refactor(session-controller): own Client Session generations
This commit is contained in:
1 parent
c52ef9fada
commit
6830e1460d
6 files changed
+413
-471
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import type { Context } from '@deepseek-ai/cordis'
|
||||
import type {} from '@deepseek-ai/dsh-agent/types'
|
||||
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,
|
||||
@@ -73,6 +76,17 @@ export type {
|
||||
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. */
|
||||
@@ -125,8 +139,11 @@ export function apply(ctx: Context): void {
|
||||
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')
|
||||
}
|
||||
@@ -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]
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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'
|
||||
@@ -24,6 +22,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,8 +47,6 @@ 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
|
||||
@@ -57,7 +54,6 @@ export interface SessionListSnapshot {
|
||||
subagentsByParent: Readonly<Record<SessionId, SubagentCatalogSnapshot>>
|
||||
/** Background jobs per session; an absent key is an empty set. */
|
||||
jobsBySession: Readonly<Record<SessionId, readonly JobView[]>>
|
||||
currentAddress: SubagentAddress | undefined
|
||||
}
|
||||
|
||||
/** One parent-addressed durable catalog projected through the sessions snapshot. */
|
||||
@@ -95,14 +91,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>>()
|
||||
/**
|
||||
* 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
|
||||
@@ -130,8 +118,6 @@ export class SessionManager {
|
||||
*/
|
||||
private readonly jobsBySession = new Map<SessionId, readonly JobView[]>()
|
||||
|
||||
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
|
||||
@@ -142,67 +128,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
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -211,7 +161,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)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -234,20 +184,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)
|
||||
@@ -256,6 +208,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()
|
||||
}
|
||||
@@ -278,7 +231,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.
|
||||
*/
|
||||
@@ -458,25 +411,9 @@ export class SessionManager {
|
||||
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)
|
||||
@@ -624,8 +561,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()
|
||||
}
|
||||
|
||||
@@ -702,7 +637,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)
|
||||
}
|
||||
}
|
||||
@@ -778,13 +713,15 @@ export class SessionManager {
|
||||
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(() => {
|
||||
@@ -861,38 +798,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
|
||||
@@ -906,7 +811,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 (
|
||||
@@ -915,7 +820,6 @@ 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
|
||||
@@ -926,20 +830,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 selected = this.selected
|
||||
const current = selected !== undefined
|
||||
&& (itemIds.has(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),
|
||||
jobsBySession: Object.fromEntries(this.jobsBySession),
|
||||
currentAddress: current === undefined ? undefined : this.addresses.get(current),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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'
|
||||
@@ -23,13 +8,16 @@ import { SESSION_SEARCH_RESULT_LIMIT } from '../../types.ts'
|
||||
import type { SessionJob as JobView } 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'
|
||||
@@ -47,8 +35,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
|
||||
@@ -61,17 +49,12 @@ 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. */
|
||||
@@ -82,14 +65,6 @@ export interface SessionListState {
|
||||
* set, so consumers read absence rather than a sentinel.
|
||||
*/
|
||||
jobsBySession: Readonly<Record<SessionId, readonly JobView[]>>
|
||||
/** 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. */
|
||||
@@ -170,15 +145,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
|
||||
@@ -187,32 +241,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).
|
||||
@@ -222,61 +259,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: {}, jobsBySession: {}, 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: {}, jobsBySession: {},
|
||||
})
|
||||
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
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -306,17 +360,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.
|
||||
@@ -393,12 +436,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.
|
||||
@@ -411,9 +450,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),
|
||||
@@ -444,33 +482,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')
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -492,61 +532,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. */
|
||||
@@ -562,21 +617,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, jobsBySession, currentAddress,
|
||||
items, phase, subagentsByParent, jobsBySession,
|
||||
} = this.manager.getListSnapshot()
|
||||
const ids: SessionId[] = []
|
||||
const byId: Record<SessionId, SessionSummary> = {}
|
||||
@@ -586,7 +639,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
|
||||
@@ -598,71 +651,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, jobsBySession, 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, jobsBySession })
|
||||
}
|
||||
|
||||
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) },
|
||||
@@ -676,44 +704,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()] : []])
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user