mward4/deepseek-harness
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

Merge branch 'worktree/jobs-api-review-3052' into worktree/jobs-api-review-3127

Browse Source 
# Conflicts:
#	.agents/notes/implemented/feature/2026-08-08-web-background-job-display.i18n.yaml
#	docs/config-catalog.i18n.yaml
#	docs/event-producer-consumer.i18n.yaml
#	docs/module-graph.i18n.yaml
#	docs/subsystems/jobs.i18n.yaml
#	docs/subsystems/shell.i18n.yaml
#	docs/subsystems/shell.md
#	docs/subsystems/shell.zh.md
#	docs/tool-catalog.i18n.yaml
#	packages/api/session-controller/README.i18n.yaml
#	packages/context/tmux-context/tests/tmux-context.spec.ts
#	packages/e2b/e2b/tests/fixtures/composition/bin.ts
#	packages/extensions/tool-cordis/src/api-catalog.ts
#	packages/shell/bash-local/README.i18n.yaml
#	packages/shell/bash-local/README.md
#	packages/shell/bash-local/README.zh.md
#	packages/shell/bash-local/src/index.ts
#	packages/shell/bash-local/tests/executor.spec.ts
#	packages/shell/bash-sandbox/README.i18n.yaml
#	packages/shell/bash-sandbox/README.md
#	packages/shell/bash-sandbox/README.zh.md
#	packages/shell/bash-sandbox/src/index.ts
#	packages/shell/bash-sandbox/tests/bwrap.e2e.ts
#	packages/shell/bash-sandbox/tests/landlock.e2e.ts
#	packages/shell/bash-sandbox/tests/partial-landlock.spec.ts
#	packages/shell/bash-sandbox/tests/sandbox.spec.ts
#	packages/shell/bash-sandbox/tests/seatbelt.e2e.ts
#	packages/shell/pwsh-local/README.i18n.yaml
#	packages/shell/pwsh-local/README.md
#	packages/shell/pwsh-local/README.zh.md
#	packages/shell/pwsh-local/src/index.ts
#	packages/shell/pwsh-local/tests/executor.spec.ts
#	packages/shell/pwsh-sandbox/README.i18n.yaml
#	packages/shell/pwsh-sandbox/README.md
#	packages/shell/pwsh-sandbox/README.zh.md
#	packages/shell/pwsh-sandbox/src/index.ts
#	packages/shell/pwsh-sandbox/tests/sandbox.spec.ts
#	packages/shell/shell/README.i18n.yaml
#	packages/shell/shell/README.md
#	packages/shell/shell/README.zh.md
#	packages/shell/shell/src/index.ts
#	packages/shell/shell/tests/service.spec.ts
#	packages/shell/tool-bash/README.i18n.yaml
#	packages/shell/tool-bash/src/index.ts
#	packages/shell/tool-bash/tests/tools.spec.ts
#	packages/shell/tool-pwsh/README.i18n.yaml
#	packages/shell/tool-pwsh/src/index.ts
#	packages/shell/tool-pwsh/tests/background.spec.ts
#	packages/shell/tool-pwsh/tests/tools.spec.ts
#	packages/workflow/tool-workflow/README.i18n.yaml
#	packages/workflow/tool-workflow/package.json
#	packages/workflow/tool-workflow/tests/tool-workflow.spec.ts
#	snapshots/session/ptc-python-turn/system-prompt.expected.md
This commit is contained in:
Yichen Jiang committed 2026-09-17 16:48:06 +08:00
2 parents 72e6329267 5ad77a9532
commit c2fb2bd4e8
4824 files changed
+860803 -61950
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/api/session-controller/src/agent.ts
+9 -2
View File
@@ -58,7 +58,7 @@ export class ApiSessionPresetConflict extends Error {
}
/** Failures produced while resolving one ordinary Session identity to its live Agent. */
export type ApiSessionAgentError = RemoteError<'session/not-found' | 'session/agent-busy' | 'gateway/internal'>
export type ApiSessionAgentError = RemoteError<'session/not-found' | 'session/agent-busy' | 'session/writer-held' | 'gateway/internal'>
/** Result of resolving one ordinary Session identity to its live Agent. */
export type ApiSessionAgentResult =
@@ -197,7 +197,11 @@ export class ApiSessionAgentController {
this.resumes.set(sessionId, resume)
}
try {
return { agent: await resume }
const agent = await resume
// A shared resume can publish an identity that subagent routing adopts
// before every waiter observes it; apply the live ownership policy again.
const published = this.liveAgent(sessionId)
return published ?? { agent }
} catch (error: unknown) {
if (error instanceof ApiSessionNotFound) {
return { error: new RemoteError('session/not-found', error.message, { sessionId }) }
@@ -211,6 +215,9 @@ export class ApiSessionAgentController {
if (racedSession !== undefined && hasApiSessionSubagentOwner(this.ctx, racedSession, undefined)) {
return { error: apiSessionSubagentOwnershipError(sessionId) }
}
if (error instanceof Error && error.name === 'SessionAlreadyOwnedError') {
return { error: new RemoteError('session/writer-held', error.message, { sessionId }) }
}
return {
error: new RemoteError(
'gateway/internal',
packages/api/session-controller/src/client/contract/sessions.ts
+59 -20
View File
@@ -14,13 +14,64 @@ import type { SessionSearchResultItem } from '../sessions/manager.ts'
import type { SessionBinding, SessionListState } from '../sessions/service.ts'
import type { SessionFace } from './session.ts'
import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store'
import type { SessionReferenceSource } from '../index.ts'
export type { AgentContext } from '../scope.ts'
/** Known Session identity or durable direct-parent subagent address; an address owns no lifetime. */
export type SessionTarget = SessionId | SubagentAddress
/** One independent use of an exact Client generation, without Host Agent ownership. */
export interface SessionReference extends Disposable {
readonly sessionId: SessionId
/** Shared binding; access fails after reference release or generation disposal. */
readonly binding: SessionBinding
/** This reference's cancellable wait for the shared initial `Session.open()` attempt to settle. */
readonly ready: Promise<SessionBinding>
/** Release once; the final reference starts local scope and history teardown. */
release(): void
}
/** Consumer identity and optional cancellation of one acquisition waiter. */
export interface SessionRetainOptions {
readonly source: SessionReferenceSource
readonly signal?: AbortSignal | undefined
}
/** Local ownership counts, independent of catalog membership and never persisted. */
export interface SessionRetainInfo {
readonly referenceCount: number
/** Positive source counts only; a source without references is absent. */
readonly retainedBy: Readonly<Partial<Record<SessionReferenceSource, number>>>
}
/** The sessions-service face injected as `ctx.sessions`. */
export interface ISessions {
/** The useSessions standard feed (list rows + current selection; read face — writes stay inside the domain). */
/** Host catalog and local reference-source counts; navigation belongs to view owners. */
readonly list: ObservableSnapshot<SessionListState>
/**
* Retain an exact Client generation and start its shared initial history opening.
* @param target - known identity or durable direct-parent address.
* @param options - required consumer source and optional independent waiter cancellation.
* @returns an owned reference immediately; await `reference.ready` when the initial open attempt must settle first.
*/
retain(target: SessionTarget, options: SessionRetainOptions): SessionReference
/**
* Hold one reference through callback settlement, including synchronous and asynchronous failures.
* @param target - Session to acquire.
* @param options - source and acquisition cancellation.
* @param operation - callback using the reference only until its returned value or Promise settles.
* @returns the callback result after release; acquisition and callback failures propagate unchanged.
*/
using<T>(target: SessionTarget, options: SessionRetainOptions, operation: (reference: SessionReference) => T | Promise<T>): Promise<T>
/**
* Observe local reference counts without retaining, creating a scope, or opening history.
* The returned source keeps stable identity across same-id generations and remains allocated
* until the Client root is disposed, even after its final subscriber leaves.
* @param id - explicit Session identity; Host existence is not implied.
* @returns a stable read-only source across same-id generations, with zero counts when none is live.
*/
retainInfo(id: SessionId): ObservableSnapshot<SessionRetainInfo>
/**
* The `session.search` result bound the wire schema fixes, exposed to
* presentation as injected data. Not per-connection state: every transport
@@ -30,23 +81,13 @@ export interface ISessions {
/**
* Create or adopt a Session on the Host.
* @param opts - target workspace, directory, and optional preallocated identity.
* @returns the Session identity after its local binding is addressable.
* @returns the catalogued identity; retain it before borrowing its binding.
*/
create(opts?: {
workspaceId?: WorkspaceId
cwd?: string
sessionId?: SessionId
}): Promise<SessionId>
/**
* Select a session as current.
* @param id - session id (must exist in the list; unknown ids fail loud).
*/
open(id: SessionId): void
/**
* Open a healthy catalog child through its exact direct-parent address.
* @param address - catalog-derived parent and child ids.
*/
openSubagent(address: SubagentAddress): void
/**
* Resolve an already discovered direct-parent address without opening it.
* @param id - possible addressed child id.
@@ -66,8 +107,6 @@ export interface ISessions {
*/
refreshSubagents(parentSessionId: SessionId): Promise<void>
/** Clear the current selection into the no-session view state. */
clear(): void
/**
* Refresh the Host-authoritative Session list.
* @returns completion of the current or newly started Session-list refresh.
@@ -86,7 +125,7 @@ export interface ISessions {
): Promise<RemoteResult<{ items: SessionSearchResultItem[]; hasMore: boolean }>>
/**
* Fork a session from a completed-turn prefix of the source; on resolution
* the child is in the list store and `open()` can target it.
* the child is in the catalog and may be explicitly retained.
* @param opts - source session id, the optional event seq anchoring the
* cut (the boundary is the first turn/end at or after it; an in-log
* anchor in an open turn is unavailable rather than clipped backward),
@@ -96,9 +135,9 @@ export interface ISessions {
*/
fork(opts: { sessionId: SessionId; atSeq?: number; increaseTitle?: boolean }): Promise<SessionId>
/**
* Resolve an Agent-scoped context view (use-and-discard).
* Borrow an already-retained Agent-scoped Context without extending its lifetime.
* @param id - session id.
* @returns scoped ctx, or undefined for a session neither listed nor already scoped.
* @returns the live scoped Context, or undefined without a retained generation.
*/
scope(id: SessionId): AgentContext | undefined
/**
@@ -111,13 +150,13 @@ export interface ISessions {
/**
* Resolve the session face behind an Agent-scoped context.
* @param ctx - an Agent-scoped context.
* @returns the session face, or undefined when the ctx is untagged or its scope was pruned.
* @returns the matching live Session, or undefined for an untagged, foreign, or ended generation.
*/
sessionOf(ctx: Context): SessionFace | undefined
/**
* Resolve the stable session binding (scope-addressed assembly feed).
* Borrow an already-retained Session binding without extending its lifetime.
* @param id - session id.
* @returns binding, or undefined for a session neither listed nor already scoped.
* @returns the live binding, or undefined without a retained generation.
*/
binding(id: SessionId): SessionBinding | undefined
}
packages/api/session-controller/src/client/contract/snapshot.ts
-15
View File
@@ -1,24 +1,10 @@
/** Session-owned observable state excluding Conversation target data. */
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { FileAttachmentRef } from '@deepseek-ai/dsh-attachment'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
import type { RemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
import type { SessionRequestId } from '../../types.ts'
/** One transient inbox occurrence from the authoritative queue snapshot. */
export interface QueuedMessage {
readonly id: MessageId
readonly messageId: MessageId
readonly placement: 'queued' | 'steering' | 'context'
/** Prompt-RPC identity of a browser-submitted occurrence; correlates the local submission echo. */
readonly rpcId?: SessionRequestId
readonly content: readonly ContentBlock[]
readonly preview: string
readonly text: string | null
}
/** One image displayed by a local submission echo before durable admission. */
export interface PendingSubmissionImage {
/** Browser-owned preview URL; its lifecycle belongs to the submitter, never this snapshot. */
@@ -82,7 +68,6 @@ export interface PromptError {
/** Immutable Session lifecycle and control snapshot. */
export interface SessionSnapshot {
readonly sessionId: SessionId
readonly queue: readonly QueuedMessage[]
/** Local prompt-submission echoes not yet observed as durable events or queue occurrences. */
readonly pendingSubmissions: readonly PendingSubmission[]
readonly running: boolean
packages/api/session-controller/src/client/index.ts
+31 -8
View File
@@ -2,8 +2,9 @@
import type { Context } from '@deepseek-ai/cordis'
import type {} from '@deepseek-ai/dsh-agent/types'
import type {} from '@deepseek-ai/dsh-client-connection/client'
import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client'
import type {} from '@deepseek-ai/dsh-client-file-upload/client'
import { typertOwnedValue } from '@deepseek-ai/dsh-typert-protocol'
import { createSessionControlStream } from './transport.ts'
import { ClientSessions } from './sessions/service.ts'
import type { SessionRemotes } from './sessions/remotes.ts'
@@ -48,7 +49,9 @@ export type {
SessionFace,
SubmissionHandle,
} from './contract/session.ts'
export type { ISessions } from './contract/sessions.ts'
export type {
ISessions, SessionReference, SessionRetainInfo, SessionRetainOptions, SessionTarget,
} from './contract/sessions.ts'
export { MutableSessionEventSource } from './contract/events.ts'
export type {
AssistantLiveChunkEvent,
@@ -70,10 +73,20 @@ export type {
PendingSubmissionImageAttachment,
PendingSubmissionPlacement,
PromptError,
QueuedMessage,
SessionSnapshot,
} from './contract/snapshot.ts'
/** Consumer-owned reference labels; extend this map through the package's canonical /client entry. */
export interface SessionReferenceSourceMap {
/** Temporary Client Controller work, including fork-title preparation. */
controllerOperation: unknown
/** A Client Gateway invocation's synchronous Context ownership. */
gateway: unknown
}
/** Declaration-merge-extensible labels carried by independent Client references. */
export type SessionReferenceSource = Extract<keyof SessionReferenceSourceMap, string>
declare module '@deepseek-ai/cordis' {
interface Context {
/** Client Session object layer and Agent scope owner. */
@@ -98,6 +111,7 @@ export const inject = [
*/
export function apply(ctx: Context): void {
const remotes = ctx.remote as unknown as SessionRemotes
const connection = ctx.get('connection') as ConnectionHandle
const sessions = new ClientSessions(ctx, remotes)
ctx.remote.$on('api-session/added', (summary) => { sessions.handleSessionAdded(summary) })
ctx.remote.$on('api-session/removed', (sessionId) => { sessions.handleSessionRemoved(sessionId) })
@@ -115,12 +129,21 @@ export function apply(ctx: Context): void {
accept: (frame) => { sessions.handleControlFrame(frame) },
failed: (error) => { console.error('[session-controller] control stream failed:', error) },
})
control.start()
ctx.on('connection/reset', () => { sessions.handleConnected() })
if (ctx.remote.$host.home !== undefined) sessions.handleConnected()
const connected = (): void => {
if (connection.generation.getSnapshot() === undefined) return
// A ready control baseline may arrive before Cordis delivers connection/reset.
sessions.handleConnected()
control.restart()
control.start()
}
ctx.effect(() => connection.generation.subscribe(connected), 'session-controller.client.generation')
connected()
ctx.typert.contexts.registerClient('agent', {
identity: candidate => sessions.scopeOf(candidate),
resolve: sessionId => sessions.resolveAgentScope(sessionId),
identity: candidate => sessions.sessionOf(candidate)?.sessionId,
resolve: (sessionId) => {
const reference = sessions.retainAgentScope(sessionId)
return typertOwnedValue(reference.binding.ctx, () => { reference.release() })
},
})
ctx.effect(() => async () => { await control.dispose() }, 'session-controller.client.control')
}
packages/api/session-controller/src/client/scope.ts
+22 -24
View File
@@ -1,20 +1,4 @@
/**
* 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.
*/
/** Client scope generations route local events independently of Host Agent residency. */
import { Context as CordisContext } from '@deepseek-ai/cordis'
import type { Context, Fiber } from '@deepseek-ai/cordis'
import type { ClientRemote } from '@deepseek-ai/dsh-api-gateway/client'
@@ -29,11 +13,15 @@ export type AgentContext = Omit<Context, 'remote'> & {
/** Context tag written by {@link createScope}. */
const kScope = Symbol('dsh.client.scope')
interface ScopeIdentity {
readonly sessionId: SessionId
}
/** 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
* through it (passing it as the dispatch subject routes to this generation's
* tagged listeners plus every untagged one).
*/
ctx: AgentContext
@@ -47,19 +35,20 @@ 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.
* admitted globally, tagged listeners only for the same Client generation.
* 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).
* @param key - durable Session identity carried by this generation.
* @returns the tagged context and its backing fiber.
*/
export function createScope(ctx: Context, key: SessionId): AgentScopeHandle {
const fiber = ctx.plugin(agentScope)
const identity: ScopeIdentity = { sessionId: key }
const scoped = fiber.ctx.extend({
[kScope]: key,
[kScope]: identity,
[CordisContext.filter](listenerCtx: Context): boolean {
const tag = scopeOf(listenerCtx)
return tag === undefined || tag === key
const tag = scopeIdentityOf(listenerCtx)
return tag === undefined || tag === identity
},
}) as AgentContext
return {
@@ -74,5 +63,14 @@ export function createScope(ctx: Context, key: SessionId): AgentScopeHandle {
* @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]
return scopeIdentityOf(ctx)?.sessionId
}
/**
* Read the exact generation identity inherited by a Client Context.
* @param ctx - scoped or root Client Context.
* @returns the generation identity, or undefined for an unscoped Context.
*/
export function scopeIdentityOf(ctx: Context): ScopeIdentity | undefined {
return (ctx as Context & { [kScope]?: ScopeIdentity })[kScope]
}
packages/api/session-controller/src/client/sessions/lineage.ts
-5
View File
@@ -27,8 +27,6 @@ export interface SessionListEntry {
cwd?: string
/** Current host-computed projection values for list consumers. */
projectionValues?: Readonly<Partial<SessionProjectionMap>>
/** Finished running while not selected and not yet opened — the sidebar's green "done" reminder (clears on select or the next run). */
completed: boolean
/** Lineage indent depth: root = 0; the UI just multiplies by the indent width. */
depth: number
}
@@ -38,12 +36,10 @@ export interface SessionListEntry {
* follows the established input order; this projection never re-sorts a
* hydrated list from mutable timestamps.
* @param summaries - the host's session.list items.
* @param completed - sessions with a pending completion reminder (manager-owned live fact; absent = false).
* @returns display rows in render order.
*/
export function flattenLineage(
summaries: readonly TitledSessionSummary[],
completed?: ReadonlySet<SessionId>,
): SessionListEntry[] {
const byId = new Map<SessionId, TitledSessionSummary>()
for (const s of summaries) byId.set(s.sessionId, s)
@@ -70,7 +66,6 @@ export function flattenLineage(
visited.add(s.sessionId)
out.push({
...s,
completed: completed?.has(s.sessionId) ?? false,
depth,
})
const kids = children.get(s.sessionId)
packages/api/session-controller/src/client/sessions/manager.ts
+60 -175
View File
@@ -1,6 +1,4 @@
// SessionManager: the instance cluster Map<SessionId, Session> (lazy-built, resident) + the frame
// dispatch entry + list state, constructed and held by ClientSessions (one per browser client).
// List data never enters zustand; React connects via subscribe/getListSnapshot.
/** Host catalog, durable projection caches, and explicitly retained Client instances. */
import type { SubagentAddress, SubagentCatalog } from '@deepseek-ai/dsh-subagent/client'
import { SessionSeq, type SessionId, type SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
@@ -8,7 +6,6 @@ import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
import type {
SessionControlBaseline,
SessionControlFrame,
SessionQueuedItem,
SessionSummary,
} from '../../types.ts'
import { mergeOrderedBaseline } from '../ordered-baseline.ts'
@@ -24,6 +21,7 @@ import { Notifier } from './notifier.ts'
import { ProjectionValueStore } from './projection-store.ts'
import { Session } from './session.ts'
import type { SessionRemotes } from './remotes.ts'
import type { SessionTarget } from '../contract/sessions.ts'
function sessionSeqCursor(value: number): SessionSeqCursor {
return value === -1 ? -1 : SessionSeq(value)
@@ -48,14 +46,11 @@ export interface SessionSearchResultItem {
/** Immutable session-list snapshot for useSessionList. */
export interface SessionListSnapshot {
items: readonly SessionListEntry[]
/** Selected Session id (validated against items; masked to undefined while its session is off the list). */
current: SessionId | undefined
state: 'idle' | 'loading' | 'error'
/** Arrival lifecycle (see {@link SessionListPhase}); `state` stays the pull-activity axis. */
phase: SessionListPhase
error: RemoteFailure | null
subagentsByParent: Readonly<Record<SessionId, SubagentCatalogSnapshot>>
currentAddress: SubagentAddress | undefined
}
/** One parent-addressed durable catalog projected through the sessions snapshot. */
@@ -93,16 +88,6 @@ export class SessionManager {
private readonly sessions = new Map<SessionId, Session>()
/** In-flight Session disposals remain here after instances leave `sessions`, so manager disposal can await quiescence. */
private readonly sessionDisposals = new Set<Promise<void>>()
/** Latest transient queues, retained independently of Session object materialization. */
private readonly queues = new Map<SessionId, readonly SessionQueuedItem[]>()
/**
* Sessions that finished running while not selected — the sidebar's green
* "done" reminder (manager-owned, survives connection generations; cleared
* on select and session-removed, re-armed by the next completion).
*/
private readonly completedNotifications = new Set<SessionId>()
/** Last-observed running bits per session; the true→false edge here arms {@link completedNotifications}. */
private readonly prevRunning = new Map<SessionId, boolean>()
/** Per-session projection value stores, retained independently of instance arrival (the
* title-snapshot precedent, generalized): push frames land here whether or not the Session
* is instantiated (list rows read the 'title' key), and an instantiated Session adopts the
@@ -114,7 +99,7 @@ export class SessionManager {
private listPhase: SessionListPhase = 'pending'
private listError: RemoteFailure | null = null
private listInflight: Promise<void> | null = null
/** Mutations arriving after a list request starts are replayed over its response. */
/** Active list request's mutation log; its identity also fences completion after reconnect. */
private listMutations: SessionListMutation[] | null = null
private readonly addresses = new Map<SessionId, SubagentAddress>()
private readonly catalogs = new Map<SessionId, SubagentCatalogSnapshot>()
@@ -124,8 +109,6 @@ export class SessionManager {
private readonly openCatalogs = new Set<SessionId>()
private readonly catalogDebounce = new Map<SessionId, ReturnType<typeof setTimeout>>()
private selected: SessionId | undefined
private listSnapshotCache: SessionListSnapshot
/** Entry-identity cache (reference stability): list rebuilds reuse the previous entry
* object when every field matches — wire refreshes mint all-new summary objects, so identity
@@ -136,67 +119,31 @@ export class SessionManager {
this.listSnapshotCache = this.buildListSnapshot()
})
/**
* @param remote - generated Remote namespaces the Session cluster calls.
* @param restoredSelection - persisted real-Session selection candidate.
*/
constructor(
private readonly remote: SessionRemotes,
restoredSelection?: SessionId,
restoredAddress?: SubagentAddress,
) {
this.selected = restoredSelection
if (restoredAddress !== undefined) this.addresses.set(restoredAddress.childSessionId, restoredAddress)
/** @param remote - generated Remote namespaces used by catalog and history readers. */
constructor(private readonly remote: SessionRemotes) {
this.listSnapshotCache = this.buildListSnapshot()
}
// ---- Selection ----
/**
* Select a listed Session or a retained catalog-addressed child.
* @param sessionId - listed or catalog-addressed Session id.
* Resolve an acquisition target without materializing a Session.
* @param target - known identity or durable direct-parent address.
* @returns the resolved identity with its explicit or catalog-derived history route installed.
*/
select(sessionId: SessionId): void {
const address = this.navigationAddress(sessionId)
if (!this.summaries.some(summary => summary.sessionId === sessionId) && address === undefined) {
throw new Error(`sessions.select: unknown session ${sessionId}`)
resolveTarget(target: SessionTarget): SessionId {
const id = typeof target === 'string' ? target : target.childSessionId
const address = typeof target === 'string' ? this.navigationAddress(id) : target
if (typeof target === 'string'
&& !this.sessions.has(id)
&& !this.summaries.some(summary => summary.sessionId === id)
&& address === undefined) {
throw new Error(`sessions.retain: unknown session ${id}`)
}
if (address !== undefined) this.addresses.set(sessionId, address)
this.sessions.get(sessionId)?.configureSubagent(
if (address !== undefined) this.addresses.set(id, address)
this.sessions.get(id)?.configureSubagent(
address,
address === undefined
? undefined
: this.catalogs.get(address.parentSessionId)?.parentAvailable,
address === undefined ? undefined : this.catalogs.get(address.parentSessionId)?.parentAvailable,
)
this.selected = sessionId
// Looking at the session consumes its completion reminder (dot clears).
this.completedNotifications.delete(sessionId)
void this.refreshSubagents(sessionId)
this.notifier.notifyNow()
}
/**
* Select a healthy child through its durable direct-parent address.
* @param address - catalog-derived parent and child ids.
*/
selectSubagent(address: SubagentAddress): void {
const catalog = this.catalogs.get(address.parentSessionId)
const entry = catalog?.entries.find(candidate => candidate.id === address.childSessionId)
if (entry === undefined || entry.kind !== 'child' || entry.mode !== address.mode) {
throw new Error(`sessions.selectSubagent: ${address.childSessionId} is not a healthy catalog child`)
}
this.addresses.set(address.childSessionId, address)
this.sessions.get(address.childSessionId)?.configureSubagent(address, catalog?.parentAvailable)
this.selected = address.childSessionId
this.completedNotifications.delete(address.childSessionId)
void this.refreshSubagents(address.childSessionId)
this.notifier.notifyNow()
}
/** Clear the selection (the layout falls to the no-session view state). */
clearSelection(): void {
this.selected = undefined
this.notifier.notifyNow()
return id
}
/**
@@ -205,7 +152,7 @@ export class SessionManager {
* @returns The direct-parent address, when navigation discovered one.
*/
subagentAddress(sessionId: SessionId): SubagentAddress | undefined {
return this.addresses.get(sessionId)
return this.navigationAddress(sessionId)
}
/**
@@ -228,20 +175,22 @@ export class SessionManager {
// ---- Instance management ----
/**
* Drop a session instance (scope-prune companion: 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.
* Withdraw an exact Client instance before running its teardown callbacks.
* @param sessionId - identity to withdraw.
* @param expected - instance being released; a replacement is left untouched.
* @returns completion of the detached instance's stream teardown.
*/
async drop(sessionId: SessionId): Promise<void> {
drop(sessionId: SessionId, expected: Session): Promise<void> {
const session = this.sessions.get(sessionId)
if (session !== expected) return Promise.resolve()
this.sessions.delete(sessionId)
if (session !== undefined) await this.startSessionDisposal(session)
this.addresses.delete(sessionId)
return this.startSessionDisposal(session)
}
/**
* Stop owned timers and every remaining Session instance.
* @returns when every Session Remote iterator has completed teardown.
* Stop catalog requests and dispose every resident Session.
* @returns once catalog requests and every Session stream have stopped.
*/
async dispose(): Promise<void> {
for (const timer of this.catalogDebounce.values()) clearTimeout(timer)
@@ -250,6 +199,7 @@ export class SessionManager {
this.openCatalogs.clear()
const sessions = [...this.sessions.values()]
this.sessions.clear()
this.addresses.clear()
for (const session of sessions) void this.startSessionDisposal(session)
await this.drainSessionDisposals()
}
@@ -272,7 +222,7 @@ export class SessionManager {
/**
* Lazy build: return the existing instance or construct one (no auto-open —
* open is triggered by the container's select callback).
* the reference allocator opens history after binding the scope).
* @param sessionId - the session to get.
* @returns the resident instance.
*/
@@ -281,11 +231,6 @@ export class SessionManager {
if (session === undefined) {
session = this.createSession(sessionId)
this.sessions.set(sessionId, session)
// Install the latest control baseline before the running-bit sync: a
// not-running summary must sweep replayed queue
// rows the same way a live status flip would (their retirement events dropped
// while the session was uninstantiated).
session.replaceControl(this.queues.get(sessionId) ?? [])
// Sync the running and blank bits from the list snapshot into the new
// instance (consistency when the list precedes open).
const summary = this.summaries.find(s => s.sessionId === sessionId)
@@ -440,7 +385,7 @@ export class SessionManager {
// ---- List API ----
/** Full refresh via session.list (single-flight: an in-flight call is reused). */
/** Full refresh via session.list (single-flight within one Host generation). */
refreshList(): Promise<void> {
if (this.listInflight !== null) return this.listInflight
this.listState = 'loading'
@@ -452,29 +397,14 @@ export class SessionManager {
this.listInflight = (async () => {
try {
const result = await this.remote.session.list({})
if (this.listMutations !== mutations) return
if (result.ok) {
const baseline: SessionSummary[] = this.listPhase === 'pending'
? [...result.value.items]
: mergeOrderedBaseline(established, result.value.items, summary => summary.sessionId)
// Seed first observations from the pull-time baseline BEFORE replaying
// in-flight mutations, then reconcile the reminders after EVERY
// replayed mutation: an edge that happens entirely between mutations
// (baseline idle → running → idle) must still arm, which a single
// sync on the folded result would collapse away.
for (const s of baseline) {
if (!this.prevRunning.has(s.sessionId)) this.prevRunning.set(s.sessionId, s.running)
}
let summaries = baseline
for (const mutation of mutations) {
summaries = applyMutation(summaries, mutation)
this.summaries = summaries
this.syncCompletedNotifications()
}
this.summaries = summaries
this.summaries = mutations.reduce(applyMutation, baseline)
this.listState = 'idle'
this.listPhase = 'ready'
// Covers the empty-mutations pull (a plain baseline carries no edge).
this.syncCompletedNotifications()
// 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)
@@ -501,12 +431,15 @@ export class SessionManager {
}
} catch (error) {
if (!isRemoteFailure(error)) throw error
if (this.listMutations !== mutations) return
this.listState = 'error'
this.listError = error
} finally {
this.listMutations = null
this.listInflight = null
this.notifier.markDirty()
if (this.listMutations === mutations) {
this.listMutations = null
this.listInflight = null
this.notifier.markDirty()
}
}
})()
return this.listInflight
@@ -619,8 +552,6 @@ export class SessionManager {
private recordMutation(mutation: SessionListMutation): void {
this.listMutations?.push(mutation)
this.summaries = applyMutation(this.summaries, mutation)
// Eager edge reconciliation — a snapshot-build-time pass would miss consecutive status frames.
this.syncCompletedNotifications()
this.notifier.markDirty()
}
@@ -655,30 +586,17 @@ export class SessionManager {
this.replaceControlBaseline(frame.value)
return
}
if (frame.type === 'projection') {
this.projectionStore(frame.sessionId).apply(frame.key, frame.value, SessionSeq(frame.seq))
this.notifier.markDirty()
return
}
this.queues.set(frame.sessionId, frame.items)
this.sessions.get(frame.sessionId)?.handleControlFrame(frame)
this.projectionStore(frame.sessionId).apply(frame.key, frame.value, SessionSeq(frame.seq))
this.notifier.markDirty()
}
private replaceControlBaseline(baseline: SessionControlBaseline): void {
this.queues.clear()
for (const [sessionId, items] of Object.entries(baseline.queues)) {
this.queues.set(sessionId as SessionId, items)
}
for (const [sessionId, block] of Object.entries(baseline.projections)) {
const store = this.projectionStore(sessionId as SessionId)
const asOfSeq = sessionSeqCursor(block.asOfSeq)
store.truncate(asOfSeq)
store.seed({ ...block, asOfSeq })
}
for (const [sessionId, session] of this.sessions) {
session.replaceControl(this.queues.get(sessionId) ?? [])
}
this.notifier.markDirty()
}
@@ -700,7 +618,7 @@ export class SessionManager {
this.markCatalogParentExpandable(summary.parentSessionId)
}
if (summary.parentSessionId !== undefined
&& (this.selected === summary.parentSessionId || this.openCatalogs.has(summary.parentSessionId))) {
&& this.openCatalogs.has(summary.parentSessionId)) {
this.scheduleCatalogRefresh(summary.parentSessionId)
}
}
@@ -718,7 +636,6 @@ export class SessionManager {
this.updateCatalogActivity(sessionId, false)
if (durableSubagent) this.sessions.get(sessionId)?.handleRunning(false)
else this.sessions.get(sessionId)?.handleRemoved()
this.queues.delete(sessionId)
if (!durableSubagent) this.projectionStores.delete(sessionId)
const inflightCatalog = this.catalogInflight.get(sessionId)
if (inflightCatalog !== undefined) {
@@ -767,17 +684,24 @@ export class SessionManager {
/**
* Repair one re-established Host-event generation with queryable baselines.
* Discard old projection cuts before new queries, including cold Sessions
* absent from the process-local control baseline.
* Opened Session follow streams resume independently through API Gateway.
*/
handleConnected(): void {
for (const store of this.projectionStores.values()) store.clear()
this.listMutations = null
this.listInflight = null
void this.refreshList()
const selectedAddress = this.selected === undefined ? undefined : this.addresses.get(this.selected)
if (selectedAddress !== undefined) void this.refreshSubagents(selectedAddress.parentSessionId)
if (this.selected !== undefined) void this.refreshSubagents(this.selected)
for (const parentSessionId of this.openCatalogs) void this.refreshSubagents(parentSessionId)
const parents = new Set(this.openCatalogs)
for (const id of this.sessions.keys()) {
const address = this.addresses.get(id)
if (address !== undefined) parents.add(address.parentSessionId)
}
for (const parentSessionId of parents) void this.refreshSubagents(parentSessionId)
}
/** Debounce membership refetches while one parent catalog is selected or open. */
/** Debounce membership refetches for an explicitly consumed catalog. */
private scheduleCatalogRefresh(parentSessionId: SessionId): void {
if (this.catalogDebounce.has(parentSessionId)) return
const timer = setTimeout(() => {
@@ -854,38 +778,6 @@ export class SessionManager {
})
}
/**
* Reconcile completion reminders against the latest summaries, eagerly after
* every mutation and pull (a snapshot-build-time pass would collapse
* consecutive status frames into one observation). A running→idle edge of a
* non-selected session arms its reminder; running disarms it; removal drops
* it. First observation only records the running bit — sessions already
* idle at load get no reminder.
*/
private syncCompletedNotifications(): void {
const seen = new Set<SessionId>()
for (const s of this.summaries) {
seen.add(s.sessionId)
const prev = this.prevRunning.get(s.sessionId)
if (prev === undefined) {
this.prevRunning.set(s.sessionId, s.running)
continue
}
if (prev && !s.running) {
if (s.sessionId !== this.selected) this.completedNotifications.add(s.sessionId)
} else if (s.running) {
this.completedNotifications.delete(s.sessionId)
}
this.prevRunning.set(s.sessionId, s.running)
}
for (const id of this.prevRunning.keys()) {
if (!seen.has(id)) this.prevRunning.delete(id)
}
for (const id of this.completedNotifications) {
if (!seen.has(id)) this.completedNotifications.delete(id)
}
}
private buildListSnapshot(): SessionListSnapshot {
const merged: TitledSessionSummary[] = this.summaries.map((summary) => {
// List rows read the generic 'title' projection key (host-computed unit
@@ -899,7 +791,7 @@ export class SessionManager {
...(projectionValues === undefined ? {} : { projectionValues }),
}
})
const fresh = flattenLineage(merged, this.completedNotifications)
const fresh = flattenLineage(merged)
const items = fresh.map((entry) => {
const prev = this.entryCache.get(entry.sessionId)
if (
@@ -908,29 +800,22 @@ export class SessionManager {
&& prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
&& prev.origin === entry.origin && prev.title === entry.title && prev.depth === entry.depth
&& prev.projectionValues === entry.projectionValues
&& prev.completed === entry.completed
) return prev
this.entryCache.set(entry.sessionId, entry)
return entry
})
const itemIds = new Set(items.map(entry => entry.sessionId))
for (const id of this.entryCache.keys()) {
if (!items.some(e => e.sessionId === id)) this.entryCache.delete(id)
if (!itemIds.has(id)) this.entryCache.delete(id)
}
const sameOrder = items.length === this.itemsCache.length && items.every((e, i) => e === this.itemsCache[i])
if (!sameOrder) this.itemsCache = items
const selected = this.selected
const current = selected !== undefined
&& (items.some(item => item.sessionId === selected) || this.addresses.has(selected))
? selected
: undefined
return {
items: this.itemsCache,
current,
state: this.listState,
phase: this.listPhase,
error: this.listError,
subagentsByParent: Object.fromEntries(this.catalogs),
currentAddress: current === undefined ? undefined : this.addresses.get(current),
}
}
}
packages/api/session-controller/src/client/sessions/projection-store.ts
+6 -13
View File
@@ -67,9 +67,9 @@ interface Channel {
/**
* One session's projection values. Framework semantics, uniform across every
* key: a baseline seeds rows at its cut, a push frame updates one row, and in
* both paths a lower-or-equal seq loses — a replayed frame cannot regress a
* value, a stale baseline cannot overwrite a newer frame. A key the store has
* never seen reads `undefined` (capability absent). Faces are identity-stable
* both paths a lower-or-equal seq within the Host generation loses. A replayed
* frame cannot regress a value; a stale baseline cannot overwrite a newer
* frame. A key the store has never seen reads `undefined` (capability absent). Faces are identity-stable
* per key (create-on-demand, cached) so the React side binds each exactly
* once; the store-level channel (`subscribeAny`) serves coarse consumers (the
* manager's list projection reads the `title` key).
@@ -159,16 +159,9 @@ export class ProjectionValueStore {
}
}
/**
* Drop rows beyond a replacement control baseline. Such rows describe
* process state the Host lost before persisting it and would otherwise
* outrank recomputed lower-seq values forever. The caller seeds the new
* baseline immediately afterward.
* @param lastSeq - highest durable sequence reflected by the baseline.
*/
truncate(lastSeq: SessionSeqCursor): void {
for (const [key, row] of this.rows) {
if (row.seq <= lastSeq) continue
/** Discard one Host generation's values and watermarks while preserving subscribed faces. */
clear(): void {
for (const key of this.rows.keys()) {
this.rows.delete(key)
this.changed(key)
}
packages/api/session-controller/src/client/sessions/queue-mirror.ts
-71
View File
@@ -1,71 +0,0 @@
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionQueuedItem } from '../../types.ts'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { QueuedMessage } from '../contract/snapshot.ts'
const QUEUE_PREVIEW_CHARS = 200
// Attachment blocks are excluded: queue presentation renders them from
// `content`, so the text preview covers only what has no visual form.
function previewOf(content: readonly ContentBlock[]): string {
const flat = content
.filter(block => block.type !== 'image' && block.type !== 'file')
.map(block => (block.type === 'text' ? block.text : `[${block.type}]`))
.join(' ').replace(/\s+/g, ' ').trim()
const chars = Array.from(flat)
return chars.length > QUEUE_PREVIEW_CHARS ? `${chars.slice(0, QUEUE_PREVIEW_CHARS).join('')}…` : flat
}
function textOf(content: readonly ContentBlock[]): string | null {
if (!content.every(block => block.type === 'text')) return null
return content.map(block => block.text).join('')
}
type QueueItems = readonly SessionQueuedItem[]
/** Authoritative transient queue projection and durable steering handoff. */
export class SessionQueueMirror {
private current: readonly QueuedMessage[] = []
/**
* Return the current immutable queue projection.
* @returns current queue rows.
*/
snapshot(): readonly QueuedMessage[] {
return this.current
}
/**
* Replace from one authoritative stream queue frame.
* @param items - complete host queue snapshot.
*/
replace(items: QueueItems): void {
this.current = items.map((item) => {
const content = item.message.content as unknown as readonly ContentBlock[]
return {
id: item.id,
messageId: item.message.id,
placement: item.placement,
...(item.rpcId === undefined ? {} : { rpcId: item.rpcId }),
content,
preview: previewOf(content),
text: textOf(content),
}
})
}
/**
* Retire a transient steering row once its durable message enters the log.
* @param event - newly contiguous durable Session event.
* @returns whether the projection changed.
*/
acceptDurable(event: SessionEvent): boolean {
if (event.type !== 'user/message') return false
const messageId = event.data.id
const index = this.current.findIndex(item =>
item.placement === 'steering' && item.messageId === messageId)
if (index < 0) return false
this.current = this.current.filter((_item, candidate) => candidate !== index)
return true
}
}
packages/api/session-controller/src/client/sessions/service.ts
+270 -274
View File
@@ -1,19 +1,4 @@
/**
* ClientSessions: root sessions service — list snapshot store (manager
* projection; carries `current`, the persisted selection every
* session-scoped surface keys off), Agent scope tree (mintScope pattern: no-op plugin
* Fiber + ctx.extend scope tag; one scope per session, agent id === session
* id), stable SessionBinding cache, breadcrumb-route projection.
*
* 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 (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.
*/
/** Client catalog and source-labelled ownership of exact Session generations. */
import type { Context, Fiber } from '@deepseek-ai/cordis'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
import { SessionSeq, type SessionId } from '@deepseek-ai/dsh-session/types'
@@ -22,13 +7,16 @@ import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
import { SESSION_SEARCH_RESULT_LIMIT } from '../../types.ts'
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
import {
createSnapshotStore, type SnapshotStore,
createSnapshotStore, notifySubscribers, type ObservableSnapshot, type SnapshotStore,
} from '@deepseek-ai/dsh-client-store'
import type { RemoteFailure, RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
import type { SessionEventSource } from '../contract/events.ts'
import type { SessionFace } from '../contract/session.ts'
import type { AgentContext, ISessions } from '../contract/sessions.ts'
import { createScope, scopeOf as scopeTagOf } from '../scope.ts'
import type {
AgentContext, ISessions, SessionReference, SessionRetainInfo, SessionRetainOptions, SessionTarget,
} from '../contract/sessions.ts'
import type { SessionReferenceSource } from '../index.ts'
import { createScope, scopeIdentityOf, scopeOf as scopeTagOf } from '../scope.ts'
import { SessionManager } from './manager.ts'
import type { SessionRemotes } from './remotes.ts'
import type { SessionListPhase, SessionSearchResultItem, SubagentCatalogSnapshot } from './manager.ts'
@@ -46,8 +34,8 @@ export interface SessionSummary {
/** Coarse durable origin for navigation filtering; not a continuation capability. */
origin?: 'subagent'
running: boolean
/** Finished while not selected and not yet opened — the sidebar's green "done" reminder. Absent = false. */
completed?: boolean
/** Local ownership counts; Host metadata refreshes cannot overwrite them. */
readonly retainedBy: SessionRetainInfo['retainedBy']
/**
* Empty-log bit (host summary derivation mirror). New Session reuses a blank
* one targeting the same workspace. Filtering stays with the consumer: the
@@ -60,29 +48,16 @@ export interface SessionSummary {
projectionValues?: Readonly<Partial<SessionProjectionMap>>
}
/**
* Session list store shape. `current` rides the same snapshot (arbitrated:
* the single useSessions standard hook reads list and selection together —
* sidebar highlighting and current-session consumers share one fact source).
*/
/** Catalog metadata and local source counts; catalog membership owns no Client generation. */
export interface SessionListState {
/** Host-list order; addressed breadcrumb-only rows are excluded. */
ids: SessionId[]
/** Host rows plus the current addressed subagent route used by navigation. */
/** Host/catalog rows plus local fallback rows for live Client generations; only `ids` expresses Host-list membership. */
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
/** Direct durable catalogs keyed by their selected parent address. */
subagentsByParent: Readonly<Record<SessionId, SubagentCatalogSnapshot>>
/** Current session's catalog-derived address, absent on ordinary navigation. */
currentAddress: SubagentAddress | undefined
}
/** Persisted navigation cell: address survives refresh for correct history routing. */
interface SessionSelection {
sessionId?: SessionId
subagentAddress?: SubagentAddress
}
/** Structured session-create failure. */
@@ -163,15 +138,94 @@ function increasedForkTitle(title: string): string {
return `${title} (1)`
}
/** Source labels are dictionary keys, including names also present on Object.prototype. */
function freezeRetainedBy(counts: Partial<Record<SessionReferenceSource, number>>): SessionRetainInfo['retainedBy'] {
Object.setPrototypeOf(counts, null)
return Object.freeze(counts)
}
const EMPTY_RETAIN_INFO: SessionRetainInfo = Object.freeze({ referenceCount: 0, retainedBy: freezeRetainedBy({}) })
interface ScopeRecord {
fiber: Fiber
ctx: AgentContext
binding: SessionBinding
/** The concrete Session for runtime-internal entry points (staging open()); the binding carries only the outward face. */
session: Session
retention: SessionRetainInfo
live: boolean
}
/** Root sessions service: list store, current selection, object-layer manager, scope tree, bindings, and breadcrumb routes. */
interface RetentionObserver {
readonly source: ObservableSnapshot<SessionRetainInfo>
readonly listeners: Set<() => void>
published: SessionRetainInfo
}
/** A cancelled waiter releases only its own reference, not the shared opening. */
async function waitForOpen(opening: Promise<void>, signal?: AbortSignal): Promise<void> {
if (signal === undefined) return opening
const aborted = Promise.withResolvers<never>()
const onAbort = (): void => { aborted.reject(signal.reason) }
signal.addEventListener('abort', onAbort, { once: true })
try {
if (signal.aborted) onAbort()
await Promise.race([opening, aborted.promise])
} finally {
signal.removeEventListener('abort', onAbort)
}
}
class ClientSessionReference implements SessionReference {
private readonly released = new AbortController()
private readonly readiness = Promise.withResolvers<SessionBinding>()
readonly ready = this.readiness.promise
constructor(
readonly sessionId: SessionId,
private record: ScopeRecord | undefined,
private releaseReference: (() => void) | undefined,
) {
void this.ready.catch(() => {})
}
get binding(): SessionBinding {
if (this.record === undefined || !this.record.live) throw new Error(`Session reference "${this.sessionId}" is released`)
return this.record.binding
}
attachOpening(opening: Promise<void>, signal?: AbortSignal): void {
const waitSignal = signal === undefined
? this.released.signal
: AbortSignal.any([this.released.signal, signal])
void waitForOpen(opening, waitSignal).then(
() => {
try {
waitSignal.throwIfAborted()
this.readiness.resolve(this.binding)
} catch (error: unknown) {
this.readiness.reject(error)
}
},
(error: unknown) => { this.readiness.reject(error) },
)
}
release(): void {
const reason = new Error(`Session reference "${this.sessionId}" is released`)
const release = this.releaseReference
this.released.abort(reason)
this.readiness.reject(reason)
this.record = undefined
this.releaseReference = undefined
release?.()
}
[Symbol.dispose](): void {
this.release()
}
}
/** Host catalog and local reference allocator; view selection remains outside the Controller. */
export class ClientSessions implements ISessions {
/**
* The wire schema's own result bound, re-exposed for presentation plugins as
@@ -180,32 +234,15 @@ export class ClientSessions implements ISessions {
* reports the same number.
*/
readonly searchResultLimit = SESSION_SEARCH_RESULT_LIMIT
/** List snapshot store (list RPC + host stream increments; re-pulled on reconnect) — the useSessions standard feed, current included. */
/** Catalog metadata and local reference-source projection. */
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
* ClientSessions.open} / {@link ClientSessions.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<SessionSelection>
private readonly scopes = new Map<SessionId, ScopeRecord>()
/** In-flight scope drops remain here after records leave `scopes`, so root disposal can await quiescence. */
/** Stable per-id sources retained for the Client root lifetime, including across generation replacement. */
private readonly retainObservers = new Map<SessionId, RetentionObserver>()
private readonly scopeDrops = new Set<Promise<void>>()
/**
* 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>()
private closed = false
/**
* @param ctx - client root context (scope fibers mount under it).
@@ -215,61 +252,78 @@ export class ClientSessions implements ISessions {
private readonly rootCtx: Context,
remote: SessionRemotes,
) {
this.selection = createSnapshotStore<SessionSelection>(
{},
{ persist: { name: 'dsh.sessions.current' } })
const restored = this.selection.getSnapshot()
this.manager = new SessionManager(
remote,
restored.sessionId,
restored.subagentAddress,
)
this.manager = new SessionManager(remote)
this.list = createSnapshotStore<SessionListState>({
ids: [], byId: {}, current: undefined, phase: 'pending',
subagentsByParent: {}, currentAddress: undefined,
})
// The manager owns wire truth; the store is its projection. Manager
// notifications are already microtask-batched.
const disposeManagerProjection = 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.
const disposeStageFollower = this.list.subscribe(() => {
this.followCurrent()
ids: [], byId: {}, phase: 'pending', subagentsByParent: {},
})
const disposeManagerProjection = this.manager.subscribe(() => { this.projectList() })
rootCtx.effect(() => async () => {
disposeStageFollower()
this.closed = true
disposeManagerProjection()
const scopes = [...this.scopes]
this.scopes.clear()
this.deferredRemovals.clear()
this.watched = undefined
for (const [id, record] of scopes) this.startScopeDrop(id, record)
for (const [, record] of scopes) {
record.live = false
record.session.unbindScope()
}
const managerDisposal = this.manager.dispose()
for (const [id, record] of scopes) {
this.startScopeDrop(id, record)
this.publishRetention(id)
}
await this.drainScopeDrops()
await this.manager.dispose()
await managerDisposal
}, 'session-controller.client.sessions')
rootCtx.reflect.provide('sessions', this, undefined)
}
/**
* Select a listed or retained catalog-addressed session as current.
* @param id - listed or addressed session id.
*/
open(id: SessionId): void {
this.manager.select(id)
retain(target: SessionTarget, options: SessionRetainOptions): SessionReference {
const { source, signal } = options
signal?.throwIfAborted()
if (this.closed) throw new Error('Session Controller is disposed')
const id = this.manager.resolveTarget(target)
const reference = this.retainScope(id, source)
try {
reference.attachOpening(this.manager.get(id).open(), signal)
return reference
} catch (error) {
reference.release()
throw error
}
}
/**
* Open a healthy catalog child through its direct-parent address.
* @param address - catalog-derived parent and child ids.
*/
openSubagent(address: SubagentAddress): void {
this.manager.selectSubagent(address)
async using<T>(
target: SessionTarget,
options: SessionRetainOptions,
operation: (reference: SessionReference) => T | Promise<T>,
): Promise<T> {
const reference = this.retain(target, options)
try {
await reference.ready
return await operation(reference)
} finally {
reference.release()
}
}
retainInfo(id: SessionId): ObservableSnapshot<SessionRetainInfo> {
let observer = this.retainObservers.get(id)
if (observer === undefined) {
const listeners = new Set<() => void>()
observer = {
listeners,
published: this.retentionSnapshot(id),
source: {
getSnapshot: () => this.retentionSnapshot(id),
subscribe: (listener) => {
listeners.add(listener)
return () => { listeners.delete(listener) }
},
},
}
this.retainObservers.set(id, observer)
}
return observer.source
}
/**
@@ -299,17 +353,6 @@ export class ClientSessions implements ISessions {
return this.manager.refreshSubagents(parentSessionId)
}
/**
* 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.
@@ -386,12 +429,8 @@ export class ClientSessions implements ISessions {
}
/**
* Create a session on the host. Resolution guarantee: by the time the
* promise resolves, the created session is in the list store and
* {@link ClientSessions.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.
* Create a Host Session and publish its catalog row before resolving.
* Callers retain the returned identity before borrowing its binding.
* @param opts - target workspace or directory and an optional preallocated id.
* @returns the new session id.
* @throws {SessionCreateError} with the requested id.
@@ -404,9 +443,8 @@ export class ClientSessions implements ISessions {
}
/**
* Fork a session from a completed-turn prefix of the source (same
* synchronous-addressability guarantee as {@link ClientSessions.create}:
* on resolution the child is in the list store and open() can target it).
* Fork a Session from a completed-turn prefix of the source and publish
* the child in the catalog before resolving.
* @param opts - source session id, the optional event seq anchoring the
* cut (the boundary is the first turn/end at or after it; an in-log
* anchor in an open turn is unavailable rather than clipped backward),
@@ -437,33 +475,35 @@ export class ClientSessions implements ISessions {
this.projectList()
const childId = result.value.sessionId
if (sourceTitle !== undefined) {
const child = this.binding(childId)?.session
if (child === undefined) throw new Error(`fork child "${childId}" is not locally addressable`)
const renamed = await child.rename(increasedForkTitle(sourceTitle))
if (!renamed.ok) throw new Error(`fork child rename failed: ${renamed.error.code}: ${renamed.error.message}`)
const reference = this.retain(childId, { source: 'controllerOperation' })
try {
await reference.ready
const renamed = await reference.binding.session.rename(increasedForkTitle(sourceTitle))
if (!renamed.ok) throw new Error(`fork child rename failed: ${renamed.error.code}: ${renamed.error.message}`)
} finally {
reference.release()
}
}
return childId
}
/**
* Resolve an Agent-scoped context view (use-and-discard).
* Borrow an already-retained Agent-scoped Context.
* @param id - session id (the agent identity — 1:1 same axis).
* @returns scoped ctx, or undefined for a session neither listed nor already scoped.
* @returns the scoped Context, or undefined without a retained generation.
*/
scope(id: SessionId): AgentContext | undefined {
return this.resolve(id)?.ctx
return this.scopes.get(id)?.ctx
}
/**
* Materialize the Agent scope named by a validated Host Remote Event.
* The first successful Session-list baseline becomes authoritative for its
* lifetime; until then, transport streams may address the scope in either
* arrival order.
* @param id - Host-projected Agent identity (the matching Session id).
* @returns the identity-stable Agent Context.
* Retain a validated Gateway identity synchronously, without history or catalog I/O.
* @param id - Host-projected Session identity, possibly not yet catalogued.
* @returns a Gateway-source reference owned by the invocation.
*/
resolveAgentScope(id: SessionId): AgentContext {
return (this.scopes.get(id) ?? this.materializeScope(id)).ctx
retainAgentScope(id: SessionId): SessionReference {
if (this.closed) throw new Error('Session Controller is disposed')
return this.retainScope(id, 'gateway')
}
/**
@@ -485,61 +525,76 @@ export class ClientSessions implements ISessions {
* `agent.session`). Same service-method boundary as
* {@link ClientSessions.scopeOf}.
* @param ctx - an Agent-scoped context.
* @returns the session face, or undefined when the ctx is untagged or its scope was pruned.
* @returns the matching live Session, or undefined for an untagged or ended generation.
*/
sessionOf(ctx: Context): SessionFace | undefined {
const id = scopeTagOf(ctx)
if (id === undefined) return undefined
return this.scopes.get(id)?.binding.session
const record = this.scopes.get(id)
return record !== undefined && scopeIdentityOf(record.ctx) === scopeIdentityOf(ctx)
? record.binding.session
: undefined
}
/**
* 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.
* Borrow an already-retained binding without extending its lifetime.
* @param id - Session identity.
* @returns the live binding, or undefined without a retained generation.
*/
binding(id: SessionId): SessionBinding | undefined {
return this.resolve(id)?.binding
return this.scopes.get(id)?.binding
}
/**
* 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.session.open()
void this.manager.refreshSubagents(current)
private retainScope(id: SessionId, source: SessionReferenceSource): ClientSessionReference {
const record = this.scopes.get(id) ?? this.materializeScope(id)
const previous = record.retention
record.retention = Object.freeze({
referenceCount: previous.referenceCount + 1,
retainedBy: freezeRetainedBy({ ...previous.retainedBy, [source]: (previous.retainedBy[source] ?? 0) + 1 }),
})
const reference = new ClientSessionReference(id, record, () => {
if (!record.live) return
const count = record.retention.referenceCount - 1
const { [source]: sourceCount = 0, ...otherSources } = record.retention.retainedBy
const retainedBy = sourceCount > 1 ? { ...otherSources, [source]: sourceCount - 1 } : otherSources
record.retention = count === 0
? EMPTY_RETAIN_INFO
: Object.freeze({ referenceCount: count, retainedBy: freezeRetainedBy(retainedBy) })
if (count === 0) this.retireScope(id, record)
else this.publishRetention(id)
})
if (this.list.getSnapshot().byId[id] === undefined) this.projectList()
this.publishRetention(id)
return reference
}
private retentionSnapshot(id: SessionId): SessionRetainInfo {
return this.scopes.get(id)?.retention ?? EMPTY_RETAIN_INFO
}
private publishRetention(id: SessionId): void {
const state = this.list.getSnapshot()
const row = state.byId[id]
const retainedBy = this.retentionSnapshot(id).retainedBy
if (row !== undefined && row.retainedBy !== retainedBy) {
this.list.set({ ...state, byId: { ...state.byId, [id]: { ...row, retainedBy } } })
}
const observer = this.retainObservers.get(id)
const snapshot = this.retentionSnapshot(id)
if (observer === undefined || observer.published === snapshot) return
observer.published = snapshot
notifySubscribers(observer.listeners, '[session-controller] reference sources')
}
/**
* Lazily mint the scope + binding for an eligible session. Eligibility and
* prune share one predicate: listed on the host or selected
* through a retained subagent address. Breadcrumb-only ancestors remain
* summary data and do not keep scopes alive.
*/
private resolve(id: SessionId): ScopeRecord | undefined {
const existing = this.scopes.get(id)
if (existing !== undefined) return existing
if (!this.eligible(id)) return undefined
return this.materializeScope(id)
private retireScope(id: SessionId, record: ScopeRecord, disposeFiber = true): void {
if (!record.live) return
record.live = false
if (this.scopes.get(id) === record) this.scopes.delete(id)
record.session.unbindScope()
const sessionDisposal = this.manager.drop(id, record.session)
this.projectList()
this.publishRetention(id)
this.startScopeDrop(id, record, disposeFiber, sessionDisposal)
}
/** Materialize one scope after its caller establishes that the id may be addressed. */
@@ -555,21 +610,19 @@ export class ClientSessions implements ISessions {
ctx,
binding,
session,
retention: EMPTY_RETAIN_INFO,
live: true,
}
this.scopes.set(id, record)
ctx.effect(() => () => { this.retireScope(id, record, false) }, 'session-controller: exact generation')
return record
}
/** The one aliveness predicate shared by scope mint and prune: host-listed or currently addressed. */
private eligible(id: SessionId): boolean {
const { ids, current } = this.list.getSnapshot()
return current === id || ids.includes(id)
}
/** Project the manager's list snapshot into the store (title derivation is display-only). */
private projectList(): void {
const previousById = this.list.getSnapshot().byId
const {
items, current, phase, subagentsByParent, currentAddress,
items, phase, subagentsByParent,
} = this.manager.getListSnapshot()
const ids: SessionId[] = []
const byId: Record<SessionId, SessionSummary> = {}
@@ -579,7 +632,7 @@ export class ClientSessions implements ISessions {
id: entry.sessionId,
displayTitle: displayTitleOf(entry.title, entry.cwd, entry.sessionId),
running: entry.running,
...(entry.completed ? { completed: true } : {}),
retainedBy: this.retentionSnapshot(entry.sessionId).retainedBy,
blank: entry.blank,
updatedAt: entry.updatedAt,
...(entry.projectionValues === undefined
@@ -591,71 +644,46 @@ export class ClientSessions implements ISessions {
...(entry.origin !== undefined ? { origin: entry.origin } : {}),
}
}
if (current !== undefined && currentAddress !== undefined) {
const seen = new Set<SessionId>()
let address: SubagentAddress | undefined = currentAddress
while (address !== undefined && !seen.has(address.childSessionId)) {
const childId = address.childSessionId
seen.add(childId)
const child = subagentsByParent[address.parentSessionId]?.entries
.find(entry => entry.kind === 'child' && entry.id === childId)
if (child?.kind !== 'child') break
for (const [parentId, catalog] of Object.entries(subagentsByParent)) {
for (const child of catalog.entries) {
if (child.kind !== 'child') continue
const childId = child.id
const displayTitle = child.label ?? childId
const summary = byId[childId]
if (summary === undefined) {
byId[childId] = {
id: childId,
displayTitle,
parentId: address.parentSessionId,
origin: 'subagent',
running: child.activity === 'running',
blank: false,
updatedAt: 0,
id: childId, displayTitle, parentId: parentId as SessionId,
origin: 'subagent', running: child.activity === 'running', blank: false, updatedAt: 0,
retainedBy: this.retentionSnapshot(childId).retainedBy,
}
} else if (summary.displayTitle !== displayTitle) {
byId[childId] = { ...summary, displayTitle }
}
const parent = byId[address.parentSessionId]
if (parent !== undefined && parent.origin !== 'subagent') break
address = this.manager.navigationAddress(address.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.getSnapshot().subagentAddress?.childSessionId !== currentAddress?.childSessionId
|| this.selection.getSnapshot().subagentAddress?.parentSessionId !== currentAddress?.parentSessionId
|| this.selection.getSnapshot().subagentAddress?.mode !== currentAddress?.mode)) {
this.selection.set({
sessionId: current,
...(currentAddress === undefined ? {} : { subagentAddress: currentAddress }),
})
}
this.list.set({ ids, byId, current, phase, subagentsByParent, currentAddress })
this.pruneScopes()
}
/** Tear down scope + instance for no-longer-eligible sessions off stage; the staged one defers until the stage moves. */
private pruneScopes(): void {
if (this.list.getSnapshot().phase === 'pending') return
for (const [id, record] of this.scopes) {
if (this.eligible(id)) continue
if (id === this.watched) {
this.deferredRemovals.add(id)
continue
if (byId[id] !== undefined) continue
const previous = previousById[id]
const snapshot = record.session.getSnapshot()
const address = this.manager.subagentAddress(id)
byId[id] = {
...(previous ?? { id, displayTitle: id, updatedAt: 0 }),
running: snapshot.running,
retainedBy: record.retention.retainedBy,
blank: snapshot.blank,
...(address === undefined ? {} : { parentId: address.parentSessionId, origin: 'subagent' }),
}
this.scopes.delete(id)
this.deferredRemovals.delete(id)
this.startScopeDrop(id, record)
}
this.list.set({ ids, byId, phase, subagentsByParent })
}
private startScopeDrop(id: SessionId, record: ScopeRecord): void {
const drop = this.dropScope(id, record)
private startScopeDrop(
id: SessionId,
record: ScopeRecord,
disposeFiber = true,
sessionDisposal = this.manager.drop(id, record.session),
): void {
const drop = this.dropScope(record, disposeFiber, sessionDisposal)
this.scopeDrops.add(drop)
void drop.then(
() => { this.scopeDrops.delete(drop) },
@@ -669,44 +697,12 @@ export class ClientSessions implements ISessions {
}
}
/**
* One teardown for the whole per-session axis: the scope
* fiber (cascading every actx-registered effect: input shell, slash
* controller, popup, plugin stores, listeners), the session-keyed slot
* registrations and the Session instance itself — the host session log is the
* durable truth, a reopen lazily rebuilds and backfills via open().
*/
private async dropScope(id: SessionId, record: ScopeRecord): Promise<void> {
// Release the Session's dispatch point with the scope it belongs to (a
// surviving instance — the live Intent — rebinds when resolve re-mints).
record.session.unbindScope()
await Promise.allSettled([
record.fiber.dispose(),
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.startScopeDrop(id, record)
}
}
/** Await the already-withdrawn Session and scoped cleanup to quiescence. */
private async dropScope(
record: ScopeRecord,
disposeFiber: boolean,
sessionDisposal: Promise<void>,
): Promise<void> {
await Promise.allSettled([sessionDisposal, ...disposeFiber ? [record.fiber.dispose()] : []])
}
}
packages/api/session-controller/src/client/sessions/session.ts
+27 -37
View File
@@ -1,6 +1,7 @@
// Sessions remain resident after creation so their open Remote sources keep running off-screen.
import type { Context } from '@deepseek-ai/cordis'
import type { InboxState } from '@deepseek-ai/dsh-agent/types'
import { randomUUID } from '@deepseek-ai/dsh-util-crypto'
import type { AttachmentIdType, FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
@@ -13,9 +14,7 @@ import type {
QueueAction,
SessionAddress,
SessionAssistantStreamBaseline,
SessionControlFrame,
SessionProjectionBaseline,
SessionQueuedItem,
SessionRequestId,
} from '../../types.ts'
import type {
@@ -36,7 +35,6 @@ import type { SessionRemotes } from './remotes.ts'
import { ProjectionValueStore } from './projection-store.ts'
import type { ProjectionsBaseline } from './projection-store.ts'
import { resolvedClientTimeZone } from '../time-zone.ts'
import { SessionQueueMirror } from './queue-mirror.ts'
import {
ClientAssistantStream,
type ClientAssistantStreamResult,
@@ -99,8 +97,7 @@ export class Session implements SessionFace {
private jumpTargetSeq: SessionSeq | null = null
/** The running jump loop's completion, shared by retargeting callers. */
private jumpPromise: Promise<void> | null = null
/** Authoritative stream-only inbox snapshot; pending work never hits history. */
private readonly queueMirror = new SessionQueueMirror()
private readonly stopObservingInbox: () => void
private readonly assistantStream = new ClientAssistantStream()
private running = false
private address: SubagentAddress | undefined
@@ -121,7 +118,7 @@ export class Session implements SessionFace {
/** Local submission echoes, insertion-ordered (see SessionSnapshot.pendingSubmissions). */
private pendingSubmissions: readonly PendingSubmission[] = []
/** Per-echo settlement state; `retiring` latches the first observation so a
* queue frame and its durable event cannot both retire one echo. */
* Inbox projection and its durable event cannot both retire one echo. */
private readonly submissionSettlements = new Map<SessionRequestId, {
readonly onRetire?: ((retirement: PendingSubmissionRetirement) => void) | undefined
retiring: boolean
@@ -173,6 +170,9 @@ export class Session implements SessionFace {
this.snapshotCache = this.buildSnapshot()
})
this.snapshotCache = this.buildSnapshot()
this.stopObservingInbox = this.projections.faceOf('inbox').subscribe(() => {
this.observeSubmissionInbox()
})
}
/**
@@ -453,7 +453,7 @@ export class Session implements SessionFace {
}
/** Rebuild an opened history source after address replacement.
* Invalidates any in-flight open first; queue state belongs to the independently
* Invalidates any in-flight open first; projection state belongs to the independently
* reconnecting control stream and remains untouched. */
async resync(): Promise<void> {
if (this.openState === 'cold') return // never opened: no window to rebuild (doOpen flips to 'loading' synchronously, so cold implies no in-flight open)
@@ -491,26 +491,6 @@ export class Session implements SessionFace {
// ---- Manager-only entry points (@internal; never called by the UI) ----
/**
* Replace every transient control value for this Session from one stream baseline.
* @param queue - complete pending queue for this Session.
*/
replaceControl(queue: readonly SessionQueuedItem[]): void {
this.queueMirror.replace(queue)
this.observeSubmissionQueue(queue)
this.notifier.markDirty()
}
/**
* Apply one Session-addressed live control update.
* @param frame - queue replacement addressed to this Session.
*/
handleControlFrame(frame: Extract<SessionControlFrame, { type: 'queue' }>): void {
this.queueMirror.replace(frame.items)
this.observeSubmissionQueue(frame.items)
this.notifier.markDirty()
}
/**
* Running-bit relay from the host stream (list entry and snapshot stay consistent).
* @param running - the new running state.
@@ -588,6 +568,7 @@ export class Session implements SessionFace {
* @returns when the Remote iterator has completed teardown.
*/
async dispose(): Promise<void> {
this.stopObservingInbox()
// Unsettled echoes retire as failed so their owners can restore or
// release browser resources; echoes already scheduled as observed keep
// that settlement.
@@ -713,18 +694,25 @@ export class Session implements SessionFace {
const event = entry.event
const awaitingFirstTurn = this.firstPromptPendingTurn
if (event.type === 'turn/start') this.firstPromptPendingTurn = false
const queueChanged = this.queueMirror.acceptDurable(event)
this.eventSource.append(entry)
// After the feed append: the conversation assembly's animation frame is
// registered by the feed subscribers above, so the echo-retirement frame
// scheduled here always runs after the durable node became renderable.
this.observeSubmissionEvent(event)
return queueChanged || awaitingFirstTurn !== this.firstPromptPendingTurn
return awaitingFirstTurn !== this.firstPromptPendingTurn
}
/** Retire the matching echo when a durable browser-prompt `user/message` becomes visible. */
/** Observe durable acceptance even when insertion and claim share one projection notification. */
private observeSubmissionEvent(event: { readonly type: string; readonly data?: unknown }): void {
if (this.submissionSettlements.size === 0 || event.type !== 'user/message') return
if (this.submissionSettlements.size === 0) return
if (event.type === 'agent/inbox/spliced') {
const splice = event.data as { readonly inserted?: unknown } | undefined
if (Array.isArray(splice?.inserted)) {
for (const message of splice.inserted) this.observeSubmissionEvent({ type: 'user/message', data: message })
}
return
}
if (event.type !== 'user/message') return
// Structural read: window entries may be compact history records, so the
// fields are narrowed rather than trusted (same posture as Conversation
// assembly matchers).
@@ -734,12 +722,15 @@ export class Session implements SessionFace {
this.scheduleObservedRetirement(source.rpcId as SessionRequestId, attachmentRefsIn(data?.content))
}
/** Retire echoes whose prompts landed in the host inbox instead of the log (running-turn submissions). */
private observeSubmissionQueue(items: readonly SessionQueuedItem[]): void {
/** Retire local echoes when their accepted messages appear in the durable Inbox projection. */
private observeSubmissionInbox(): void {
if (this.submissionSettlements.size === 0) return
for (const item of items) {
if (item.rpcId !== undefined) {
this.scheduleObservedRetirement(item.rpcId, attachmentRefsIn(item.message.content))
const inbox = this.projections.get('inbox') as InboxState | undefined
if (inbox === undefined) return
for (const message of [...inbox['next-turn'], ...inbox['next-step']]) {
const source = message.source
if (source.kind === 'user' && 'rpcId' in source) {
this.scheduleObservedRetirement(source.rpcId, attachmentRefsIn(message.content))
}
}
}
@@ -795,7 +786,6 @@ export class Session implements SessionFace {
private buildSnapshot(): SessionSnapshot {
return {
sessionId: this.sessionId,
queue: this.queueMirror.snapshot(),
pendingSubmissions: this.pendingSubmissions,
running: this.running,
subagent: this.address === undefined
packages/api/session-controller/src/client/time-zone.ts
+1 -1
View File
@@ -1,4 +1,4 @@
/** Browser-owned time-zone sampling for prompt RPC provenance. */
/** Browser-owned time-zone sampling for one prompt RPC. */
/**
* Resolve the current browser IANA zone for one outbound operation.
packages/api/session-controller/src/commands.ts
+12 -8
View File
@@ -241,10 +241,7 @@ export class SessionCommandController {
{ sessionId: request.sessionId },
)
}
let cut = SessionLogOffset(boundary.seq + 1)
while (cut < source.events.length && source.events[cut]?.type !== 'turn/start') {
cut = SessionLogOffset(cut + 1)
}
const cut = SessionLogOffset(boundary.seq + 1)
let workspace: Workspace | undefined
try {
workspace = await this.forkWorkspace(source.header)
@@ -418,11 +415,11 @@ export class SessionCommandController {
}
/**
* Mutate one still-pending queue occurrence without resuming a cold Agent.
* Mutate one pending Inbox occurrence, restoring an ordinary cold Agent when needed.
* @param request - Session, queue item, and requested mutation.
* @returns acknowledgement that the queue mutation was applied.
*/
updateQueue(request: SessionUpdateQueueRequest): SessionUpdateQueueValue {
async updateQueue(request: SessionUpdateQueueRequest): Promise<SessionUpdateQueueValue> {
if (request.action.kind === 'edit') {
if (request.action.content.some(block => block.type !== 'text')) {
throw new RemoteError(
@@ -439,9 +436,14 @@ export class SessionCommandController {
)
}
}
const agent = this.ctx.agents.get(request.sessionId)
let agent = this.ctx.agents.get(request.sessionId)
if (agent === undefined) {
throw new RemoteError('session/queue-item-not-found', 'queued item is no longer pending', { itemId: request.itemId })
const found = await this.agents.resolveAgent(request.sessionId)
if ('error' in found) {
if (found.error.code !== 'session/not-found') throw found.error
throw new RemoteError('session/queue-item-not-found', 'queued item is no longer pending', { itemId: request.itemId })
}
agent = found.agent
}
if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
const identity = this.ctx.sessionProjections
@@ -542,6 +544,7 @@ export class SessionCommandController {
private async readSessionState(sessionId: SessionId): Promise<SessionReadState> {
const attached = this.ctx.sessions.get(sessionId)
if (attached !== undefined) {
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
return { id: attached.id, header: attached.header, events: attached.snapshotEvents() }
}
const inspected = await inspectApiSession(this.ctx, sessionId)
@@ -588,6 +591,7 @@ function hasPromptRequest(agent: Agent, requestId: SessionRequestId): boolean {
return source.kind === 'user' && 'rpcId' in source && source.rpcId === requestId
}
if (agent.inbox.nextTurn.some(matches) || agent.inbox.nextStep.some(matches)) return true
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
return agent.session.snapshotEvents().some((event) => {
if (event.type !== 'user/message') return false
const source = event.data.source
packages/api/session-controller/src/control.ts
+2 -48
View File
@@ -1,10 +1,9 @@
/** Live Session queue and projection state with reconnect baselines. */
/** Live Session projection state with reconnect baselines. */
import type { Context } from '@deepseek-ai/cordis'
import type { Agent, InboxState } from '@deepseek-ai/dsh-agent'
import { Deque } from '@deepseek-ai/dsh-deque'
import type {
Session, SessionId, UserMessage,
Session, SessionId,
} from '@deepseek-ai/dsh-session'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import type {
@@ -12,7 +11,6 @@ import type {
SessionControlFrame,
SessionProjectionBaseline,
SessionProjectionValues,
SessionQueuedItem,
} from './types.ts'
/** Owns the Host-wide Session control stream. */
@@ -29,14 +27,6 @@ export class SessionControlController {
value: value as JsonValue,
seq,
})
if (key !== 'inbox') return
const agent = this.ctx.agents.get(session.id)
if (agent?.session !== session) return
this.broadcast({
type: 'queue',
sessionId: session.id,
items: queueItemsFromInbox(value as InboxState),
})
})
ctx.effect(() => () => {
for (const stream of this.streams) stream.end()
@@ -64,13 +54,7 @@ export class SessionControlController {
private baseline(): SessionControlBaseline {
const sessions = this.ctx.sessions.list()
const queues = Object.create(null) as Record<SessionId, readonly SessionQueuedItem[]>
for (const session of sessions) {
const agent = this.ctx.agents.get(session.id)
queues[session.id] = agent?.session === session ? queueItems(agent) : []
}
return {
queues,
projections: this.projectionBaseline(sessions),
}
}
@@ -135,33 +119,3 @@ class ControlQueue {
}
}
}
function queueItems(agent: Agent): SessionQueuedItem[] {
return queueItemsFromInbox({
'next-turn': agent.inbox.nextTurn,
'next-step': agent.inbox.nextStep,
})
}
function queueItemsFromInbox(inbox: InboxState): SessionQueuedItem[] {
return [
...inbox['next-turn'].map(message => ({
id: message.id,
placement: 'queued' as const,
...promptRpcId(message),
message: { id: message.id, content: message.content as unknown as JsonValue[] },
})),
...inbox['next-step'].map(message => ({
id: message.id,
placement: message.source.kind === 'user' ? 'steering' as const : 'context' as const,
...promptRpcId(message),
message: { id: message.id, content: message.content as unknown as JsonValue[] },
})),
]
}
/** Prompt-RPC identity carried by a browser-submitted message's user source. */
function promptRpcId(message: UserMessage): Pick<SessionQueuedItem, 'rpcId'> {
const source = message.source
return source.kind === 'user' && 'rpcId' in source ? { rpcId: source.rpcId } : {}
}
packages/api/session-controller/src/history.ts
+1
View File
@@ -152,6 +152,7 @@ export class SessionHistoryController {
// Constructor seed events have no session/event notification. Normally
// only the end-seed suffix is new; if persistence advanced after the
// opening observation, replay everything beyond that snapshot cursor.
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
const suffix = session.snapshotEvents(snapshotCursor === undefined
? session.firstLiveSeq
: SessionLogOffset(snapshotCursor + 1))
packages/api/session-controller/src/index.ts
+3 -2
View File
@@ -207,6 +207,7 @@ export class SessionController extends TypertRemoteService {
return Promise.resolve({
meta: attached.header,
inheritedEventCount: attached.inheritedEventCount,
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
events: attached.snapshotEvents(),
})
}
@@ -359,12 +360,12 @@ export class SessionController extends TypertRemoteService {
}
/**
* Mutate one still-pending queue occurrence on a live Agent.
* Mutate one still-pending queue occurrence, resuming a cold Agent first.
* @param request - Session, queue item, and requested mutation.
* @returns acknowledgement that the queue mutation was applied.
*/
@Remote('updateQueue')
updateQueue(request: SessionUpdateQueueRequest): SessionUpdateQueueValue {
updateQueue(request: SessionUpdateQueueRequest): Promise<SessionUpdateQueueValue> {
return this.commands.updateQueue(request)
}
packages/api/session-controller/src/skill-catalog.ts
+1
View File
@@ -78,6 +78,7 @@ export class SessionSkillCatalog extends TypertRemoteService {
return {
skills: skills.map(skill => ({
name: skill.name,
...skill.path === undefined ? {} : { path: skill.path },
description: skill.description,
...skill.whenToUse === undefined ? {} : { whenToUse: skill.whenToUse },
modelInvocable: skill.invocation.modelInvocable,
packages/api/session-controller/src/types.ts
+3 -15
View File
@@ -190,6 +190,7 @@ declare module '@deepseek-ai/dsh-typert-protocol' {
readonly requestedCwd: string
readonly existingCwd?: string
}
'session/writer-held': { readonly sessionId: SessionId }
'session/agent-busy': { readonly reason: string }
'session/invalid-time-zone': { readonly value: string }
'session/workspace-attach-failed': { readonly sessionId: SessionId; readonly workspaceId: string }
@@ -222,6 +223,8 @@ export interface SkillListRequest {
/** One skill available to the Session's human-facing composer. */
export interface SkillEntry {
/** Absolute SKILL.md path when supplied by a filesystem provider. */
readonly path?: string
/** Kebab-case identifier referenced as `/name`. */
readonly name: string
/** Short routing description. */
@@ -522,22 +525,8 @@ export type SessionFollowFrame =
| SessionEventEntry
| { readonly type: 'assistant-stream'; readonly frame: SessionAssistantStreamFrame }
/** One pending inbox occurrence in the authoritative queue snapshot. */
export interface SessionQueuedItem {
readonly id: MessageId
readonly placement: 'queued' | 'steering' | 'context'
/** Prompt-RPC identity from the queued message's user source; clients retire the matching local submission echo on it. */
readonly rpcId?: SessionRequestId
/** JSON-safe message fields consumed by pending-queue presentation. */
readonly message: {
readonly id: MessageId
readonly content: readonly JsonValue[]
}
}
/** Complete live control baseline emitted once per control stream generation. */
export interface SessionControlBaseline {
readonly queues: Readonly<Record<SessionId, readonly SessionQueuedItem[]>>
readonly projections: Readonly<Record<SessionId, SessionProjectionBaseline>>
}
@@ -552,7 +541,6 @@ export interface SessionProjectionUpdate {
/** Host-wide live state stream. Each generation starts with exactly one baseline. */
export type SessionControlFrame =
| { readonly type: 'baseline'; readonly value: SessionControlBaseline }
| { readonly type: 'queue'; readonly sessionId: SessionId; readonly items: readonly SessionQueuedItem[] }
| ({ readonly type: 'projection' } & SessionProjectionUpdate)
declare module '@deepseek-ai/cordis' {