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

Merge origin/master into feat/apply-patch

Browse Source 
Regenerate config/tool/module/event catalogs, Cordis API and client slot catalogs; re-record reviewed bilingual hashes instead of merging generated bytes.

Accept deleted filesystem invariant companions. Master has no replacement event-invariant registration; keep delete-intent in typed events, policy guards, and behavior tests.

Adopt master workspace-changes intent waterfalls and tool-execution recorder association, add a prepended delete-intent listener that delegates before capture, remove argument parsing and the fs-patch dependency, and retain the real-tool Review matrix.

Keep master filesystem mutation-result metadata and tool switches alongside opt-in apply_patch. Await current-directory resolution in plan and no-follow preflight; store patch diffs as processPath. Adapt native/PTC fixtures to the supported modes.

Preserve client patch diffs and stopped states with master partial arguments and absolute navigation. Use master's rewritten ProviderEditor, including its remaining repeated model validation.

Regenerate the lockfile with pnpm install and align the new fs-patch package version. Clean retired package build outputs. Update exact rescope exceptions for verified preset and event identifiers from master; old package names remain rejected.
This commit is contained in:
sjw1231 committed 2026-10-16 11:51:43 +08:00
2 parents c5db86f864 3f621c65c3
commit e2f1208bcf
8713 files changed
+351270 -88758
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/api/session-controller/src/agent.ts
+168 -94
View File
@@ -1,5 +1,6 @@
/** Agent activation, composition, and model-selection policy owned by API Session. */
import { randomUUID } from 'node:crypto'
import { mkdir } from 'node:fs/promises'
import type { Context } from '@deepseek-ai/cordis'
import { installModelSelection } from '@deepseek-ai/dsh-agent'
@@ -8,11 +9,12 @@ import type {
} from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-agent-default-model'
import type {} from '@deepseek-ai/dsh-agent-preset-registry'
import { brandString } from '@deepseek-ai/dsh-brand'
import { ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { RemoteError, remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
import type {} from '@deepseek-ai/dsh-typert-registry'
import type { ModelSelection } from './types.ts'
@@ -42,21 +44,49 @@ export class ApiSessionCwdConflict extends Error {
}
}
/** Explicit-id creation attempted to adopt a Session under another preset. */
/** Reuse of a Session named an Agent preset while the deployment composes no preset roster. */
export class ApiSessionPresetConflict extends Error {
constructor(
readonly sessionId: SessionId,
readonly requestedPreset: string,
readonly existingPreset: string | undefined,
) {
super(
existingPreset === undefined
? `session "${sessionId}" records no agent preset and cannot be adopted under "${requestedPreset}"`
: `session "${sessionId}" runs agent preset "${existingPreset}", not "${requestedPreset}"`,
)
super(`the deployment composes no agent preset roster, so session "${sessionId}" cannot run preset "${requestedPreset}"`)
}
}
/** A blank Session chosen for reuse started a turn before this Host acquired its writer. */
class ApiSessionStartedSinceObserved extends Error {}
/** Placement and Agent preset for one Session created for a new conversation. */
export interface ApiSessionCreateRequest {
/** Blank Session to reuse, or the identity of a new Session when none is stored under it. */
readonly sessionId?: SessionId
/** Directory the Session owns. */
readonly cwd: string
/** Preset the Session runs; the configured default when omitted. */
readonly agentPreset?: string
/**
* Placement rule the caller owns, such as Workspace membership and archive state.
* @param sessionId - existing Session named by the request.
* @returns whether that Session may be reused.
*/
readonly reusable: (sessionId: SessionId) => boolean
}
/** The Session one creation request created or reused. */
export interface ApiSessionCreated {
/** Live Agent of the Session. */
readonly agent: Agent
/** Preset the Session ran when the request finished, before a later request for the same identity; absent without a roster. */
readonly agentPreset: string | undefined
}
/** Resolved preset and pre-publication Agent setup for one create or resume. */
interface ApiSessionComposition {
readonly agentPreset?: string
readonly setup: AgentSetup
}
/** Failures produced while resolving one ordinary Session identity to its live Agent. */
export type ApiSessionAgentError = RemoteError<'session/not-found' | 'session/agent-busy' | 'session/writer-held' | 'gateway/internal'>
@@ -139,7 +169,7 @@ export async function inspectApiSession(
/** Owns every operation that may create, resume, or configure a Web Agent. */
export class ApiSessionAgentController {
private readonly resumes = new Map<SessionId, Promise<Agent>>()
private readonly creations = new Map<SessionId, Promise<Agent>>()
private readonly claims = new Map<SessionId, Promise<void>>()
private readonly selections = new WeakMap<Agent, InstalledSelection>()
private readonly imageAdmissionChains = new WeakMap<Agent, Promise<void>>()
@@ -229,50 +259,26 @@ export class ApiSessionAgentController {
}
/**
* Resolve one requested identity, creating or resuming it once.
* @param sessionId - requested Session identity.
* @param cwd - directory the Session must own.
* @param checkPersistedIdentity - whether to inspect a cold identity before creation.
* @param presetId - optional Agent preset the Session must own.
* @returns the matching live ordinary Agent.
* Create a Session for a new conversation, reusing the named blank Session
* while it is still reusable. A reused Session runs the requested preset, or
* the configured default when the request names none; the preset it recorded
* is not resolved, and a change is logged as `agent-preset/selected` before
* the Agent is published. A named Session that has started a turn, fails the
* caller's placement rule, or is written by another owner stays unchanged,
* and a new Session is created instead. Requests naming one identity run in
* arrival order, including their preset resolution, and check the named
* Session's directory and ownership before resolving the preset.
* @param request - optional Session to reuse, directory, preset, and placement rule.
* @returns the reused or created Session and the preset it ran when this request finished.
* @throws ApiSessionCwdConflict when the named Session belongs to another directory.
* @throws ApiSessionSubagentOwnership when subagent routing owns the named Session.
* @throws RemoteError `agent-preset/not-found` or `agent-preset/invalid` for the requested or default preset.
* @throws ApiSessionPresetConflict when no roster is composed and the request names a preset for a reused Session.
*/
async ensureSession(
sessionId: SessionId,
cwd: string,
checkPersistedIdentity: boolean,
presetId?: string,
): Promise<Agent> {
let creation = this.creations.get(sessionId)
if (creation === undefined) {
creation = this.createOrAdopt(sessionId, cwd, checkPersistedIdentity, presetId)
.catch((error: unknown) => {
const live = this.ctx.agents.get(sessionId)
if (live !== undefined) {
if (hasApiSessionSubagentOwner(this.ctx, live.session, live)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
return live
}
const attached = this.ctx.sessions.get(sessionId)
if (attached !== undefined && hasApiSessionSubagentOwner(this.ctx, attached, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
throw error
})
.finally(() => { this.creations.delete(sessionId) })
this.creations.set(sessionId, creation)
}
const agent = await creation
if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
if (presetId !== undefined) {
this.assertPresetUnchanged(sessionId, presetId, this.presetForSession(agent.session))
}
if (agent.session.header.cwd !== cwd) {
throw new ApiSessionCwdConflict(sessionId, cwd, agent.session.header.cwd)
}
return agent
async createSession(request: ApiSessionCreateRequest): Promise<ApiSessionCreated> {
const named = request.sessionId
if (named === undefined) return this.created(await this.createMinted(request.cwd, await this.composeAgent(request.agentPreset)))
return this.inOrder(named, async () => this.created(await this.claim(named, request)))
}
/**
@@ -378,10 +384,7 @@ export class ApiSessionAgentController {
* @param presetId - requested preset or the configured default when omitted.
* @returns the resolved preset identity and Agent setup callback.
*/
async composeAgent(presetId: string | undefined): Promise<{
readonly agentPreset?: string
readonly setup: AgentSetup
}> {
async composeAgent(presetId: string | undefined): Promise<ApiSessionComposition> {
const presets = this.ctx.get('agentPresets')
if (presets === undefined) {
return { setup: (_agentCtx, agent) => { this.installSelection(agent) } }
@@ -441,48 +444,128 @@ export class ApiSessionAgentController {
})).agent
}
private async createOrAdopt(
sessionId: SessionId,
cwd: string,
checkPersistedIdentity: boolean,
presetId: string | undefined,
): Promise<Agent> {
private async inOrder<Value>(sessionId: SessionId, operation: () => Promise<Value>): Promise<Value> {
const result = (this.claims.get(sessionId) ?? Promise.resolve()).then(operation)
const settled = result.then(() => undefined, () => undefined)
this.claims.set(sessionId, settled)
try {
return await result
} finally {
if (this.claims.get(sessionId) === settled) this.claims.delete(sessionId)
}
}
/** @returns the reused Session, a new Session under `sessionId` when none is stored, or a new Session under a minted identity. */
private async claim(sessionId: SessionId, request: ApiSessionCreateRequest): Promise<Agent> {
const attached = this.ctx.sessions.get(sessionId)
const live = this.ctx.agents.get(sessionId)
if (attached !== undefined && hasApiSessionSubagentOwner(this.ctx, attached, live)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
if (live !== undefined) return live
if (checkPersistedIdentity) {
try {
using observation = await this.ctx.sessionQuery.observeSession(sessionId)
if (hasApiSessionSubagentOwner(this.ctx, { header: observation.header }, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
if (observation.header.cwd !== cwd) {
throw new ApiSessionCwdConflict(sessionId, cwd, observation.header.cwd)
}
const storedPreset = this.presetForObservation(observation)
this.assertPresetUnchanged(sessionId, presetId, storedPreset)
const composition = await this.composeAgent(storedPreset)
return (await this.ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: this.agentOptions(),
setup: composition.setup,
})).agent
} catch (error: unknown) {
if (!(error instanceof SessionQueryError)
|| error.code !== 'SESSION_QUERY_SESSION_NOT_FOUND') throw error
if (live !== undefined) {
if (live.session.header.cwd !== request.cwd) {
throw new ApiSessionCwdConflict(live.id, request.cwd, live.session.header.cwd)
}
const composition = await this.composeAgent(request.agentPreset)
return await this.reuseLive(live, request, composition) ?? this.createMinted(request.cwd, composition)
}
let observation: SessionObservation
try {
observation = await this.ctx.sessionQuery.observeSession(sessionId)
} catch (error: unknown) {
if (!(error instanceof SessionQueryError) || error.code !== 'SESSION_QUERY_SESSION_NOT_FOUND') throw error
return this.createAgent(sessionId, request.cwd, await this.composeAgent(request.agentPreset))
}
let reusable: boolean
{
using observed = observation
if (hasApiSessionSubagentOwner(this.ctx, { header: observed.header }, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
if (observed.header.cwd !== request.cwd) {
throw new ApiSessionCwdConflict(sessionId, request.cwd, observed.header.cwd)
}
reusable = request.reusable(sessionId) && !observed.events.some(event => event.type === 'turn/start')
}
const composition = await this.composeAgent(request.agentPreset)
if (!reusable) return this.createMinted(request.cwd, composition)
this.assertPresetApplicable(sessionId, request.agentPreset)
try {
return (await this.ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: this.agentOptions(),
setup: this.reuseSetup(composition),
})).agent
} catch (error: unknown) {
if (error instanceof ApiSessionStartedSinceObserved) return this.createMinted(request.cwd, composition)
if (!(error instanceof Error) || error.name !== 'SessionAlreadyOwnedError') throw error
// Another resume in this process may hold the writer for the Agent it published.
return this.ctx.agents.get(sessionId) === undefined
? this.createMinted(request.cwd, composition)
: this.claim(sessionId, request)
}
}
/** @returns the live blank Session under the requested preset, or undefined when it cannot be reused. */
private async reuseLive(
live: Agent,
request: ApiSessionCreateRequest,
composition: ApiSessionComposition,
): Promise<Agent | undefined> {
if (!request.reusable(live.id) || !this.isBlank(live.session)) return undefined
this.assertPresetApplicable(live.id, request.agentPreset)
const current = this.presetForSession(live.session)
const presets = this.ctx.get('agentPresets')
const target = composition.agentPreset
if (presets === undefined || target === undefined || target === current) return live
try {
await presets.select(live, target)
} catch (error: unknown) {
if (remoteErrorOf(error)?.code !== 'agent-preset/locked') throw error
return undefined
}
return live
}
/** Setup that re-checks blankness under the acquired writer, then records a preset change before publication. */
private reuseSetup(composition: ApiSessionComposition): AgentSetup {
return async (agentCtx, agent) => {
if (!this.isBlank(agent.session)) throw new ApiSessionStartedSinceObserved()
const commit = await composition.setup(agentCtx, agent)
const target = composition.agentPreset
if (target !== undefined && this.presetForSession(agent.session) !== target) {
agent.session.append('agent-preset/selected', { agentPreset: target })
}
return commit
}
}
/** Whether no turn has started, by the predicate the preset registry's `select` enforces. */
private isBlank(session: Session): boolean {
const boundary = this.ctx.sessionProjections.stateOf(session, 'turnBoundary')
return boundary === undefined || (boundary.openTurnStartSeq === null && boundary.lastTurn === 0)
}
/** Without a roster, reuse cannot apply a named preset. */
private assertPresetApplicable(sessionId: SessionId, requested: string | undefined): void {
if (requested === undefined || this.ctx.get('agentPresets') !== undefined) return
throw new ApiSessionPresetConflict(sessionId, requested)
}
private created(agent: Agent): ApiSessionCreated {
return { agent, agentPreset: this.presetForSession(agent.session) }
}
private createMinted(cwd: string, composition: ApiSessionComposition): Promise<Agent> {
return this.createAgent(brandString<SessionId>(`session-${randomUUID()}`), cwd, composition)
}
private async createAgent(sessionId: SessionId, cwd: string, composition: ApiSessionComposition): Promise<Agent> {
try {
await mkdir(cwd, { recursive: true })
} catch (error: unknown) {
throw new Error(`failed to ensure project directory "${cwd}": ${String(error)}`, { cause: error })
}
const composition = await this.composeAgent(presetId)
return (await this.ctx.agents.create({
sessionId,
agentOptions: this.agentOptions(),
@@ -514,15 +597,6 @@ export class ApiSessionAgentController {
}
return observation.projections.values.agentPreset ?? undefined
}
private assertPresetUnchanged(
sessionId: SessionId,
requested: string | undefined,
existing: string | undefined,
): void {
if (requested === undefined || requested === existing) return
throw new ApiSessionPresetConflict(sessionId, requested, existing)
}
}
function agentModelSelection(selection: ModelSelection): AgentModelSelection {
packages/api/session-controller/src/client/contract/sessions.ts
+10 -5
View File
@@ -79,24 +79,27 @@ export interface ISessions {
*/
readonly searchResultLimit: number
/**
* Create or adopt a Session on the Host.
* @param opts - target workspace, directory, and optional preallocated identity.
* @returns the catalogued identity; retain it before borrowing its binding.
* Create a Session for a new conversation on the Host. The Host reuses the
* named blank Session while it is reusable and otherwise creates another; the
* Session runs the named preset, or the Host default when none is named.
* @param opts - target workspace or directory, an optional blank Session to reuse, and an optional Agent preset.
* @returns the catalogued identity the Host created or reused; retain it before borrowing its binding.
*/
create(opts?: {
workspaceId?: WorkspaceId
cwd?: string
sessionId?: SessionId
agentPreset?: string
}): Promise<SessionId>
/**
* Resolve an already discovered direct-parent address without opening it.
* @param id - possible addressed child id.
* @returns a retained or loaded-catalog address, without retaining a new selection or scope.
* @returns a retained, catalog-derived, or unknown-mode listed address, without retaining a new selection or scope.
*/
subagentAddress(id: SessionId): SubagentAddress | undefined
/**
* Load all Session projections once per connection; retry an unsuccessful initial read.
* Read projections without starting migration; retry failed or migration-deferred reads.
* @param sessionId - Session to inspect without opening its conversation.
* @returns completion of the current or newly started refresh.
*/
@@ -134,6 +137,8 @@ export interface ISessions {
atSeq?: number
increaseTitle?: boolean
onCreated?: (childId: SessionId) => void
/** False refuses source migration; omission preserves the Host default. */
allowMigration?: boolean
}): Promise<SessionId>
/**
* Borrow an already-retained Agent-scoped Context without extending its lifetime.
packages/api/session-controller/src/client/sessions/lineage.ts
+2
View File
@@ -27,6 +27,8 @@ export interface SessionListEntry {
parentSessionId?: SessionId
/** Coarse durable origin for navigation filtering; not a continuation capability. */
origin?: 'subagent'
/** Format classification supplied by the Host list. */
formatStatus?: SessionSummary['formatStatus']
cwd?: string
/** Current host-computed projection values for list consumers. */
projectionValues?: Readonly<Partial<SessionProjectionMap>>
packages/api/session-controller/src/client/sessions/manager.ts
+43 -15
View File
@@ -61,7 +61,7 @@ export interface SessionListSnapshot {
/** Shared projection values and the lifecycle of their explicit baseline read. */
export interface SessionProjectionSnapshot {
readonly values: Readonly<Partial<SessionProjectionMap>>
readonly state: 'idle' | 'loading' | 'ready' | 'error'
readonly state: 'idle' | 'loading' | 'ready' | 'migration-required' | 'error'
readonly error: RemoteFailure | null
}
@@ -153,8 +153,9 @@ export class SessionManager {
/**
* Resolve an address for breadcrumb navigation without retaining transport authority.
* @param sessionId - possible child id in an already-loaded catalog.
* @returns A retained or catalog-derived direct-parent address.
* @param sessionId - possible child id in received Session metadata.
* @returns A retained address, then a catalog address, then an unknown-mode address
* from a listed subagent header; undefined without that evidence.
*/
subagentAddress(sessionId: SessionId): SubagentAddress | undefined {
const retained = this.addresses.get(sessionId)
@@ -162,7 +163,7 @@ export class SessionManager {
for (const parentSessionId of this.projectionStores.keys()) {
const child = this.projectionStores.get(parentSessionId)?.values().subagentCatalog
?.find(entry => entry.id === sessionId)
if (child !== undefined) {
if (child !== undefined && child.mode !== 'external') {
return {
parentSessionId,
childSessionId: sessionId,
@@ -170,7 +171,13 @@ export class SessionManager {
}
}
}
return undefined
const summary = this.summaries.find(summary => summary.sessionId === sessionId)
if (summary?.origin !== 'subagent' || summary.parentSessionId === undefined) return undefined
return {
parentSessionId: summary.parentSessionId,
childSessionId: sessionId,
mode: 'unknown',
}
}
// ---- Instance management ----
@@ -248,7 +255,7 @@ export class SessionManager {
const address = this.addresses.get(sessionId)
const child = address === undefined ? undefined : this.projectionStores.get(address.parentSessionId)?.values().subagentCatalog
?.find(entry => entry.id === sessionId)
if (child !== undefined) {
if (child !== undefined && child.mode !== 'external') {
// A catalogued child exists only after its delegated session has
// durable history, even though child rows do not carry `blank`.
session.handleBlank(false)
@@ -332,7 +339,7 @@ export class SessionManager {
}
/**
* Load a complete projection baseline once per connection; retry unsuccessful reads.
* Read projections without starting migration; retry failed or migration-deferred reads.
* @param sessionId - Session to inspect without opening its conversation.
* @returns completion of the current or newly started read.
*/
@@ -350,13 +357,26 @@ export class SessionManager {
const result = await this.remote.session.projections({ sessionId }, controller.signal)
if (controller.signal.aborted) return
if (result.ok) {
if (result.value !== null) {
store.seed({ ...result.value, asOfSeq: sessionSeqCursor(result.value.asOfSeq) })
const value = result.value
if (value !== null) {
switch (value.kind) {
case 'sequenced':
store.seed({ values: value.values, asOfSeq: sessionSeqCursor(value.asOfSeq) })
break
case 'migration-required':
store.applyCached(value.values)
break
default:
assertNever(value, 'session projections result')
}
} else if (store.values() === initialValues) {
// A later frame proves existence independently of an earlier missing read.
store.clear()
}
this.projectionLoads.set(sessionId, { state: 'ready', error: null })
this.projectionLoads.set(sessionId, {
state: value?.kind === 'migration-required' ? 'migration-required' : 'ready',
error: null,
})
} else {
this.projectionLoads.set(sessionId, { state: 'error', error: result.error })
}
@@ -482,17 +502,21 @@ export class SessionManager {
* Contract session.create; on success merge into summaries immediately (no
* wait for the next refresh). A created session is blank by definition
* (entity birth precedes the first message).
* @param opts - target workspace or working directory, plus an optional caller-owned id.
* @returns the create result.
* @param opts - target workspace or working directory, an optional blank Session to reuse, and an optional Agent preset.
* @returns the create result, naming the Session the Host created or reused.
*/
async create(
opts: {
workspaceId?: WorkspaceId
cwd?: string
sessionId?: SessionId
agentPreset?: string
} = {},
): Promise<RemoteResult<{ sessionId: SessionId }>> {
const shared = opts.sessionId === undefined ? {} : { sessionId: opts.sessionId }
const shared = {
...(opts.sessionId === undefined ? {} : { sessionId: opts.sessionId }),
...(opts.agentPreset === undefined ? {} : { agentPreset: opts.agentPreset }),
}
const payload = opts.workspaceId !== undefined
? { workspaceId: opts.workspaceId, ...shared }
: { ...(opts.cwd === undefined ? {} : { cwd: opts.cwd }), ...shared }
@@ -526,16 +550,17 @@ export class SessionManager {
* preserve it or lower it after an exact cut before the first `turn/start`;
* lineage rides parentSessionId. A child published before Workspace
* attachment fails is also reconciled into the list.
* @param opts - source session and the optional exact inclusive boundary seq.
* @param opts - source session, optional exact inclusive boundary seq, and permission to start migration.
* @returns the fork result (the child session id).
*/
async fork(
opts: { sessionId: SessionId; atSeq?: SessionSeq },
opts: { sessionId: SessionId; atSeq?: SessionSeq; allowMigration?: boolean },
): Promise<RemoteResult<{ sessionId: SessionId }>> {
const source = this.summaries.find(s => s.sessionId === opts.sessionId)
const result = await this.remote.session.fork({
sessionId: opts.sessionId,
...opts.atSeq === undefined ? {} : { atSeq: opts.atSeq },
...opts.allowMigration === undefined ? {} : { allowMigration: opts.allowMigration },
})
const childId = result.ok
? result.value.sessionId
@@ -774,6 +799,7 @@ export class SessionManager {
&& prev.blank === entry.blank
&& prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
&& prev.origin === entry.origin && prev.title === entry.title && prev.depth === entry.depth
&& prev.formatStatus === entry.formatStatus
&& prev.projectionValues === entry.projectionValues
) return prev
this.entryCache.set(entry.sessionId, entry)
@@ -813,6 +839,7 @@ function applyMutation(summaries: readonly SessionSummary[], mutation: SessionLi
...(mutation.kind === 'upsert' ? {
agentAvailable: mutation.summary.agentAvailable,
running: mutation.summary.running,
...(mutation.summary.formatStatus === undefined ? {} : { formatStatus: mutation.summary.formatStatus }),
} : {}),
...(existing.cwd === undefined && mutation.summary.cwd !== undefined ? { cwd: mutation.summary.cwd } : {}),
...(existing.parentSessionId === undefined && mutation.summary.parentSessionId !== undefined
@@ -823,6 +850,7 @@ function applyMutation(summaries: readonly SessionSummary[], mutation: SessionLi
if (filled.cwd === existing.cwd && filled.parentSessionId === existing.parentSessionId
&& filled.origin === existing.origin && filled.blank === existing.blank
&& filled.agentAvailable === existing.agentAvailable && filled.running === existing.running
&& filled.formatStatus === existing.formatStatus
) return [...summaries]
return summaries.map(summary => summary.sessionId === mutation.summary.sessionId ? filled : summary)
}
packages/api/session-controller/src/client/sessions/projection-store.ts
+3 -3
View File
@@ -6,8 +6,8 @@
* (`{ value, seq }`) comes from the connected Host — a follow opening
* baseline, a Session Controller `projection` frame, a list block the Host
* computed for an attached Session — and merges under **higher seq wins**
* against other sequenced rows. A `cached` row (`{ value }`) comes from the
* session list's view of the persisted checkpoint, carries no comparable
* against other sequenced rows. A `cached` row (`{ value }`) comes from
* a Host view of the persisted checkpoint, carries no comparable
* seq, and yields to every sequenced write. No client-side domain folding
* exists: a domain ships projection support with zero client code. Per-key
* bare observable faces feed `useProjection` (ui-renderer binds them).
@@ -166,7 +166,7 @@ export class ProjectionValueStore {
}
/**
* Fill keys from a session-list block the Host labeled `cached`: a zero-I/O
* Fill keys from a Host block labeled `cached`: a zero-I/O
* view of the persisted checkpoint. A cached value lands only where no
* sequenced row exists: a connected Session has already answered for such
* a key, and the list's view of the persisted checkpoint cannot be newer
packages/api/session-controller/src/client/sessions/service.ts
+19 -7
View File
@@ -21,6 +21,7 @@ import { SessionManager } from './manager.ts'
import type { SessionRemotes } from './remotes.ts'
import type { SessionListPhase, SessionSearchResultItem, SessionProjectionSnapshot } from './manager.ts'
import type { Session } from './session.ts'
import type { SessionListEntry } from './lineage.ts'
/** Session list row projected from the host list RPC plus live stream increments. */
export interface SessionSummary {
@@ -33,6 +34,8 @@ export interface SessionSummary {
parentId?: SessionId
/** Coarse durable origin for navigation filtering; not a continuation capability. */
origin?: 'subagent'
/** Format classification supplied by the Host list, when available. */
formatStatus?: SessionListEntry['formatStatus']
/** Host running state for `ids` members; a display fallback for other rows. */
running: boolean
/** Local ownership counts; Host metadata refreshes cannot overwrite them. */
@@ -40,7 +43,7 @@ export interface SessionSummary {
/**
* New Session presentation and reuse eligibility, derived from the Host
* summary, `sessionListMetadata`, and client acceptance/running observations.
* New Session reuses a blank one targeting the same workspace. Filtering
* Workspace reconnection may reuse a blank in the same workspace. Filtering
* stays with the consumer: the store carries every row, while the Workspace
* browser shows only the selected blank entry.
*/
@@ -332,14 +335,14 @@ export class ClientSessions implements ISessions {
* Resolve an already discovered direct-parent address without opening it.
* Feature plugins use this to avoid Agent-bound RPCs in persisted child views.
* @param id - possible addressed child id.
* @returns A retained or loaded-catalog address, without retaining a new selection or scope.
* @returns A retained, catalog-derived, or unknown-mode listed address, without retaining a new selection or scope.
*/
subagentAddress(id: SessionId): SubagentAddress | undefined {
return this.manager.subagentAddress(id)
}
/**
* Load all Session projections once per connection; retry an unsuccessful initial read.
* Read projections without starting migration; retry failed or migration-deferred reads.
* @param sessionId - Session to inspect without opening its conversation.
*/
refreshProjections(sessionId: SessionId): Promise<void> {
@@ -424,11 +427,16 @@ export class ClientSessions implements ISessions {
/**
* 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.
* @param opts - target workspace or directory, an optional blank Session to reuse, and an optional Agent preset.
* @returns the created or reused session id.
* @throws {SessionCreateError} with the requested id.
*/
async create(opts: { workspaceId?: WorkspaceId; cwd?: string; sessionId?: SessionId } = {}): Promise<SessionId> {
async create(opts: {
workspaceId?: WorkspaceId
cwd?: string
sessionId?: SessionId
agentPreset?: string
} = {}): Promise<SessionId> {
const result = await this.manager.create(opts)
if (!result.ok) throw new SessionCreateError(result.error, opts.sessionId)
this.projectList()
@@ -443,7 +451,7 @@ export class ClientSessions implements ISessions {
* seq (a real event seq the caller already knows; a cut inside an open
* turn is balanced Host-side with synthetic closers, and omission selects
* the latest completed-turn prefix), and whether to increment an
* inherited durable title before resolving.
* inherited durable title before resolving, plus permission to start source migration.
* @returns the child session id.
* @throws {SessionForkError} with the source id.
* @throws {Error} when a requested child-title rename fails after creation.
@@ -453,6 +461,7 @@ export class ClientSessions implements ISessions {
atSeq?: number
increaseTitle?: boolean
onCreated?: (childId: SessionId) => void
allowMigration?: boolean
}): Promise<SessionId> {
const sourceTitle = opts.increaseTitle
? this.list.getSnapshot().byId[opts.sessionId]?.title
@@ -460,6 +469,7 @@ export class ClientSessions implements ISessions {
const result = await this.manager.fork({
sessionId: opts.sessionId,
...(opts.atSeq === undefined ? {} : { atSeq: SessionSeq(opts.atSeq) }),
...(opts.allowMigration === undefined ? {} : { allowMigration: opts.allowMigration }),
})
if (!result.ok) throw new SessionForkError(result.error, opts.sessionId)
this.projectList()
@@ -622,6 +632,7 @@ export class ClientSessions implements ISessions {
retainedBy: this.retentionSnapshot(entry.sessionId).retainedBy,
blank: entry.blank,
updatedAt: entry.updatedAt,
formatStatus: entry.formatStatus,
...(entry.projectionValues === undefined
? {}
: { projectionValues: entry.projectionValues }),
@@ -633,6 +644,7 @@ export class ClientSessions implements ISessions {
}
for (const [parentId, projection] of Object.entries(projectionsBySession)) {
for (const child of projection.values.subagentCatalog ?? []) {
if (child.mode === 'external') continue
const childId = child.id
const summary = byId[childId]
const projectionValues = summary?.projectionValues ?? this.manager.projectionValues(childId)
packages/api/session-controller/src/client/sessions/session.ts
+1 -1
View File
@@ -289,7 +289,7 @@ export class Session implements SessionFace {
// wire type is used; this array is not filtered or reordered.
const routedContent = content as Exclude<PromptContentPart, { readonly type: 'file' }>[]
const routed = await this.remote.subagents.prompt({
requestId: randomUUID() as SessionRequestId,
requestId: requestId ?? randomUUID() as SessionRequestId,
parentSessionId: this.address.parentSessionId,
childSessionId: this.address.childSessionId,
mode: 'continuable',
packages/api/session-controller/src/commands.ts
+28 -46
View File
@@ -20,7 +20,6 @@ import { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title'
import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time'
import { assertNever } from '@deepseek-ai/dsh-util-values'
import { RemoteError, remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
import type { Workspace } from '@deepseek-ai/dsh-workspace'
@@ -33,6 +32,7 @@ import {
apiSessionSubagentOwnershipError,
hasApiSessionSubagentOwner,
inspectApiSession,
type ApiSessionCreated,
} from './agent.ts'
import type {
SessionAttachmentRequest,
@@ -64,7 +64,12 @@ type PromptContentCandidate =
| SessionPromptRequest['content'][number]
| Extract<SessionUpdateQueueRequest['action'], { readonly kind: 'edit' }>['content'][number]
function hasPromptContent(content: readonly PromptContentCandidate[]): boolean {
/**
* Whether prompt content carries non-whitespace text or an attachment.
* @param content - submitted or edited prompt parts.
* @returns false for content the Agent must not receive.
*/
export function hasPromptContent(content: readonly PromptContentCandidate[]): boolean {
return content.some(part => part.type !== 'text' || part.text.trim().length > 0)
}
@@ -98,15 +103,15 @@ export class SessionCommandController {
) {}
/**
* Create or idempotently adopt one ordinary Session.
* @param request - requested identity, location, and Agent preset.
* @returns the Session identity and resolved preset when configured.
* Create one ordinary Session for a new conversation, reusing the named
* blank Session while it is an unarchived member of the requested Workspace.
* @param request - optional blank Session to reuse, location, and Agent preset.
* @returns the created or reused Session identity and its preset when configured.
*/
async create(request: SessionCreateRequest): Promise<SessionCreateValue> {
if (request.workspaceId !== undefined && request.cwd !== undefined) {
throw new RemoteError('gateway/bad-request', 'session.create accepts workspaceId or cwd, not both', {})
}
const sessionId = request.sessionId ?? brandString<SessionId>(`session-${randomUUID()}`)
let workspace: Workspace | undefined
if (request.workspaceId !== undefined) {
workspace = this.ctx.workspaceRegistry.get(request.workspaceId)
@@ -117,17 +122,19 @@ export class SessionCommandController {
}
}
const cwd = workspace?.path ?? request.cwd ?? this.defaultCwd
let adopted: Agent
let created: ApiSessionCreated
try {
adopted = await this.agents.ensureSession(
sessionId,
created = await this.agents.createSession({
...(request.sessionId === undefined ? {} : { sessionId: request.sessionId }),
cwd,
request.sessionId !== undefined,
request.agentPreset,
)
...(request.agentPreset === undefined ? {} : { agentPreset: request.agentPreset }),
reusable: sessionId => !this.ctx.workspaceRegistry.archivedSessionIds.includes(sessionId)
&& (workspace === undefined || workspace.sessionIds.includes(sessionId)),
})
} catch (error) {
this.rejectCreation(sessionId, error)
this.rejectCreation(error)
}
const sessionId = created.agent.id
if (workspace !== undefined) {
try {
await workspace.attachSession(sessionId)
@@ -139,8 +146,7 @@ export class SessionCommandController {
)
}
}
const agentPreset = this.agents.presetForSession(adopted.session)
return { sessionId, ...(agentPreset === undefined ? {} : { agentPreset }) }
return { sessionId, ...(created.agentPreset === undefined ? {} : { agentPreset: created.agentPreset }) }
}
/**
@@ -304,35 +310,15 @@ export class SessionCommandController {
}
/**
* Reject empty content, then admit one prompt after Agent and attachment validation.
* @param request - Session identity, prompt content, source metadata, and delivery mode.
* Admit one prompt after Agent and attachment validation; the Remote method
* has already validated the content and built the durable message source.
* @param request - Session identity, prompt content, and delivery mode.
* @param source - the `user-rpc` source recorded on the accepted message.
* @returns acknowledgement that the Agent accepted the prompt.
*/
async prompt(request: SessionPromptRequest): Promise<SessionPromptValue> {
if (!hasPromptContent(request.content)) {
throw new RemoteError(
'gateway/bad-request',
'prompt content must include non-whitespace text or an attachment',
{},
)
}
const clientTimeZone = request.clientTimeZone === undefined
? undefined
: canonicalClientTimeZone(request.clientTimeZone)
if (request.clientTimeZone !== undefined && clientTimeZone === undefined) {
throw new RemoteError(
'session/invalid-time-zone',
'clientTimeZone must be UTC or a valid IANA Area/Location name',
{ value: request.clientTimeZone },
)
}
async prompt(request: SessionPromptRequest, source: MessageSource): Promise<SessionPromptValue> {
const agent = await this.resolveAgent(request.sessionId)
if (hasPromptRequest(agent, request.requestId)) return { accepted: true }
const source: MessageSource = {
kind: 'user',
rpcId: request.requestId,
...(clientTimeZone === undefined ? {} : { clientTimeZone }),
}
const hasImage = request.content.some(part => part.type === 'image')
const admit = async (): Promise<SessionPromptValue> => {
try {
@@ -530,16 +516,12 @@ export class SessionCommandController {
return found.agent
}
private rejectCreation(sessionId: SessionId, error: unknown): never {
private rejectCreation(error: unknown): never {
if (remoteErrorOf(error) !== undefined) throw error
if (error instanceof Error && error.name === 'SessionAlreadyOwnedError') {
throw new RemoteError('session/writer-held', error.message, { sessionId })
}
if (error instanceof ApiSessionPresetConflict) {
throw new RemoteError('agent-preset/conflict', error.message, {
sessionId: error.sessionId,
requestedPreset: error.requestedPreset,
...(error.existingPreset === undefined ? {} : { existingPreset: error.existingPreset }),
})
}
if (error instanceof ApiSessionCwdConflict) {
@@ -552,7 +534,7 @@ export class SessionCommandController {
if (error instanceof ApiSessionSubagentOwnership) {
throw apiSessionSubagentOwnershipError(error.sessionId)
}
throw new RemoteError('gateway/internal', `failed to create session "${sessionId}": ${String(error)}`, {})
throw new RemoteError('gateway/internal', `failed to create session: ${String(error)}`, {})
}
private async readSessionState(sessionId: SessionId): Promise<SessionReadState> {
packages/api/session-controller/src/index.ts
+123 -29
View File
@@ -6,19 +6,23 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-fs'
import { Context } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'
import { errorChain, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import { errorChain, ReasoningEffortId, type MessageSource } from '@deepseek-ai/dsh-llm'
import type {} from '@deepseek-ai/dsh-client-file-upload'
import { canOpenNativePath, nativeFileManager, nativeFileApplications, openNativeFileApplication, openNativeAssociatedPath, revealNativePath } from '@deepseek-ai/dsh-native-command'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence'
import type {} from '@deepseek-ai/dsh-session-projection-cache'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import { Remote, RemoteError, TypertRemoteService } from '@deepseek-ai/dsh-typert-protocol'
import { Remote, RemoteError, TypertRemoteService, principalOf } from '@deepseek-ai/dsh-typert-protocol'
import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time'
import { assertNever } from '@deepseek-ai/dsh-util-values'
import type {} from '@deepseek-ai/dsh-api-remote-access'
import {
ApiSessionAgentController,
inspectApiSession,
type ApiSessionAgentResult,
} from './agent.ts'
import { SessionCommandController } from './commands.ts'
import { SessionCommandController, hasPromptContent } from './commands.ts'
import { SessionControlController } from './control.ts'
import { SessionHistoryController } from './history.ts'
import { SessionFileReferences } from './file-references.ts'
@@ -31,6 +35,7 @@ import { ArchivedSessionGate } from './archived-session-gate.ts'
import type {
ModelCatalog,
SessionWorkspacePathApplication,
SessionAddress,
SessionAttachmentRequest,
SessionAttachmentValue,
SessionCancelRequest,
@@ -75,12 +80,27 @@ declare module '@deepseek-ai/cordis' {
}
}
/**
* Read the Session resource a history address names: the Session itself, or
* the child Session whose transcript a subagent address reads.
* @param request - history request carrying the durable address.
* @returns the Session whose access the read requires.
*/
function addressedSession(request: { readonly address: SessionAddress }): SessionId {
return request.address.kind === 'session' ? request.address.sessionId : request.address.childSessionId
}
/** Session Controller deployment policy. */
export interface Config {
/** Override platform desktop-opener detection. */
readonly nativeOpen?: boolean
/** Positive integral milliseconds of list work before yielding between complete rows. */
readonly listWorkSliceMs?: number
}
/** Deployment policy after schema defaults have been applied. */
type ResolvedConfig = Config & { readonly listWorkSliceMs: number }
/** Host integrations replaceable by direct unit tests. */
export interface SessionControllerInternals {
/** Native default-application handoff. */
@@ -104,6 +124,7 @@ export class SessionController extends TypertRemoteService {
'fileUploads',
'fs',
'llm',
'remoteAccess',
'sessions',
'sessionProjections',
'sessionQuery',
@@ -111,8 +132,9 @@ export class SessionController extends TypertRemoteService {
'workspaceRegistry',
]
static Config: z<Config> = z.object({
static Config: z<Config, ResolvedConfig> = z.object({
nativeOpen: z.boolean(),
listWorkSliceMs: z.natural().min(1).default(16),
})
private readonly agents: ApiSessionAgentController
@@ -129,11 +151,13 @@ export class SessionController extends TypertRemoteService {
/**
* @param ctx - Host context containing the Session capability assembly.
* @param config - native-opener deployment policy.
* @param config - native-opener and list-scheduling deployment policy.
* @param internals - host integrations replaceable by direct unit tests.
*/
constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) {
super(ctx, 'sessionController', { namespace: 'session' })
ctx.typert.resources.register('session', { field: 'sessionId' })
const resolved = SessionController.Config(config)
installModelSelectionProjection(ctx)
this.agents = new ApiSessionAgentController(ctx)
this.commands = new SessionCommandController(ctx, this.agents, process.cwd())
@@ -149,7 +173,7 @@ export class SessionController extends TypertRemoteService {
await Promise.allSettled([...this.promotions])
}, 'session-controller.promotions')
this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
this.listState = new ApiSessionList(ctx)
this.listState = new ApiSessionList(ctx, resolved.listWorkSliceMs)
this.fileApplications = internals.fileApplications ?? nativeFileApplications
this.openFileApplication = internals.openFileApplication ?? openNativeFileApplication
this.openPath = internals.openPath ?? openNativeAssociatedPath
@@ -245,29 +269,33 @@ export class SessionController extends TypertRemoteService {
/**
* Read all visible Session rows without resuming an Agent.
* @param _request - reserved empty list request.
* @param signal - cancellation for persistence reads.
* @param signal - cancellation for persistence reads and summary generation.
* @returns visible Session summaries ordered by activity.
*/
@Remote('list')
async list(_request: SessionListRequest, signal: AbortSignal): Promise<SessionListValue> {
return { items: await this.listState.list(signal) }
const items = await this.listState.list(signal)
return { items: items.filter(item => this.ctx.remoteAccess.allows({ kind: 'session', id: item.sessionId })) }
}
/**
* Search visible Session content without resuming an Agent.
* Search visible Session content without resuming an Agent. A non-owner
* receives only the hits whose Session `allows()` admits; `hasMore`
* describes the bounded result before that filtering.
* @param request - literal message-content query.
* @param signal - cancellation for list and search reads.
* @returns authorized bounded Session search results.
*/
@Remote('search')
search(request: SessionSearchRequest, signal: AbortSignal): Promise<SessionSearchValue> {
return this.listState.search(request.query, signal)
async search(request: SessionSearchRequest, signal: AbortSignal): Promise<SessionSearchValue> {
const result = await this.listState.search(request.query, signal)
return { ...result, items: result.items.filter(item => this.ctx.remoteAccess.allows({ kind: 'session', id: item.sessionId })) }
}
/**
* Create or idempotently adopt one ordinary Session.
* @param request - requested identity, location, and Agent preset.
* @returns the Session identity and resolved preset when configured.
* Create one ordinary Session for a new conversation, reusing the named blank Session while it is reusable.
* @param request - optional blank Session to reuse, location, and Agent preset.
* @returns the created or reused Session identity and its preset when configured.
*/
@Remote('create')
create(request: SessionCreateRequest): Promise<SessionCreateValue> {
@@ -408,11 +436,19 @@ export class SessionController extends TypertRemoteService {
* Fork one cold-readable exact event prefix into a new Session. An omitted
* boundary selects the latest completed-turn prefix; an open cut receives
* synthetic fork closers.
* @param request - source Session and optional exact inclusive event boundary.
* @param request - source Session, optional exact inclusive event boundary, and migration preflight choice.
* @returns the new Session identity.
*/
@Remote('fork')
fork(request: SessionForkRequest): Promise<SessionForkValue> {
async fork(request: SessionForkRequest): Promise<SessionForkValue> {
if (request.allowMigration === false && this.ctx.sessions.get(request.sessionId) === undefined) {
const stored = await this.ctx.get('sessionPersistence')?.stat(request.sessionId)
if (stored?.formatStatus === 'migration-required' && this.ctx.sessions.get(request.sessionId) === undefined) {
throw new RemoteError('session/migration-required', 'Open the source Session to complete migration before forking', {
sessionId: request.sessionId,
})
}
}
return this.commands.fork(request)
}
@@ -423,9 +459,30 @@ export class SessionController extends TypertRemoteService {
* @returns acknowledgement that the Agent accepted the prompt.
*/
@Remote('prompt')
prompt(request: SessionPromptRequest, signal: AbortSignal): Promise<SessionPromptValue> {
async prompt(request: SessionPromptRequest, signal: AbortSignal): Promise<SessionPromptValue> {
signal.throwIfAborted()
return this.commands.prompt(request)
if (!hasPromptContent(request.content)) {
throw new RemoteError('gateway/bad-request', 'prompt content must include non-whitespace text or an attachment', {})
}
const clientTimeZone = request.clientTimeZone === undefined
? undefined
: canonicalClientTimeZone(request.clientTimeZone)
if (request.clientTimeZone !== undefined && clientTimeZone === undefined) {
throw new RemoteError(
'session/invalid-time-zone',
'clientTimeZone must be UTC or a valid IANA Area/Location name',
{ value: request.clientTimeZone },
)
}
// The caller is recorded only when it is not the owner, so owner messages keep their exact source.
const principal = principalOf(this.ctx.invocation)
const source: MessageSource = {
kind: 'user',
rpcId: request.requestId,
...(clientTimeZone === undefined ? {} : { clientTimeZone }),
...(principal === undefined ? {} : { principal }),
}
return this.commands.prompt(request, source)
}
/**
@@ -464,7 +521,7 @@ export class SessionController extends TypertRemoteService {
* @param signal - cancellation for persistence reads.
* @returns one chronological page.
*/
@Remote('page')
@Remote({ name: 'page', resources: { session: addressedSession } })
page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionPage> {
return this.history.page(request, signal)
}
@@ -476,16 +533,16 @@ export class SessionController extends TypertRemoteService {
* @returns a complete opening snapshot followed by gap-free durable event
* frames and optional cursorless assistant-stream frames.
*/
@Remote({ mode: 'stream' })
@Remote({ mode: 'stream', resources: { session: addressedSession } })
follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
return this.history.follow(request, signal)
}
/**
* Read all registered projections without activating an Agent.
* Read exact projections, returning cached hints only when migration is required.
* @param request - Session whose current values are required.
* @param signal - cancellation for the Session observation.
* @returns complete baseline, or null when the Session does not exist.
* @param signal - cancellation for the Session read.
* @returns a sequenced baseline, cached hints when migration is deferred, or null when absent.
*/
@Remote('projections')
async projections(request: SessionProjectionsRequest, signal: AbortSignal): Promise<SessionProjectionsValue> {
@@ -494,16 +551,29 @@ export class SessionController extends TypertRemoteService {
throw new RemoteError('gateway/bad-request', 'sessionId must not be empty', {})
}
try {
signal.throwIfAborted()
const live = this.liveProjections(sessionId)
if (live !== undefined) return live
const stored = await this.ctx.get('sessionPersistence')?.stat(sessionId, { signal })
signal.throwIfAborted()
const attached = this.liveProjections(sessionId)
if (attached !== undefined) return attached
if (stored === undefined) return null
if (stored.formatStatus === 'migration-required') {
const cached = this.ctx.get('sessionProjectionCache')?.cachedSnapshot(stored.header)
return { kind: 'migration-required', values: { ...cached?.values } as SessionProjectionValues }
}
using observation = await this.ctx.sessionQuery.observeSession(sessionId, { signal })
const projections = observation.projections
if (projections === undefined) {
throw new RemoteError('session/projections-unavailable', 'Session projections are unavailable', {})
}
return { asOfSeq: projections.asOfSeq, values: projections.values as SessionProjectionValues }
return { kind: 'sequenced', asOfSeq: projections.asOfSeq, values: projections.values as SessionProjectionValues }
} catch (error: unknown) {
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') return null
if (signal.aborted
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return this.liveProjections(sessionId) ?? null
}
if (signal.aborted || (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
throw new RemoteError('gateway/cancelled', 'Session projection read was cancelled', {}, { cause: error })
}
if (error instanceof RemoteError) throw error
@@ -511,14 +581,38 @@ export class SessionController extends TypertRemoteService {
}
}
private liveProjections(
sessionId: SessionId,
): Extract<SessionProjectionsValue, { readonly kind: 'sequenced' }> | undefined {
const session = this.ctx.sessions.get(sessionId)
if (session === undefined) return undefined
const snapshot = this.ctx.sessionProjections.snapshot(session)
return { kind: 'sequenced', asOfSeq: snapshot.asOfSeq, values: snapshot.values as SessionProjectionValues }
}
/**
* Stream a complete live-control baseline followed by replacement frames.
* @param signal - cancellation owned by the Remote stream carrier.
* @returns one complete baseline followed by live replacement frames.
*/
@Remote({ mode: 'stream' })
control(signal: AbortSignal): AsyncIterable<SessionControlFrame> {
return this.controlState.control(signal)
async *control(signal: AbortSignal): AsyncIterable<SessionControlFrame> {
for await (const frame of this.controlState.control(signal)) {
switch (frame.type) {
case 'baseline': {
const projections = Object.fromEntries(Object.entries(frame.value.projections)
.filter(([sessionId]) => this.ctx.remoteAccess.allows({ kind: 'session', id: sessionId as SessionId })))
yield { type: 'baseline', value: { ...frame.value, projections } }
break
}
case 'projection':
if (this.ctx.remoteAccess.allows({ kind: 'session', id: frame.sessionId })) yield frame
break
/* v8 ignore next -- closed control-frame union. */
default:
assertNever(frame)
}
}
}
packages/api/session-controller/src/list.ts
+32 -10
View File
@@ -1,12 +1,14 @@
/** Cold-safe Session list and search projection. */
import { performance } from 'node:perf_hooks'
import { scheduler } from 'node:timers/promises'
import type { Context } from '@deepseek-ai/cordis'
import type {} from '@deepseek-ai/dsh-agent-preset-registry'
import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection'
import type {} from '@deepseek-ai/dsh-session-projection-cache'
import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
import { SessionQueryError, type SessionRecord, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { z } from 'zod'
import {
@@ -74,8 +76,11 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
/** Owns list projection registration, bounded cold summaries, and authorized search. */
export class ApiSessionList {
/** @param ctx - Host context carrying Session, query, persistence, and projection services. */
constructor(private readonly ctx: Context) {
/**
* @param ctx - Host context carrying Session, query, persistence, and projection services.
* @param workSliceMs - Resolved positive integral list-work budget in milliseconds.
*/
constructor(private readonly ctx: Context, private readonly workSliceMs: number) {
ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
key: 'sessionListMetadata',
stateSchema: sessionListMetadataSchema,
@@ -111,6 +116,7 @@ export class ApiSessionList {
sessionId: session.id,
updatedAt: updatedAt(session.header, metadata),
agentAvailable: this.ctx.agents.get(session.id)?.session === session,
formatStatus: 'current',
running: this.ctx.agents.get(session.id)?.status === 'running',
blank: metadata?.blank ?? session.seq === 0,
...listFields(session.header),
@@ -120,7 +126,7 @@ export class ApiSessionList {
/**
* Read every visible attached and persisted Session without activating an Agent.
* @param signal - optional cancellation for persistence reads.
* @param signal - optional cancellation for persistence reads and summary generation.
* @returns visible Session summaries ordered by activity.
*/
async list(signal?: AbortSignal): Promise<SessionSummary[]> {
@@ -128,28 +134,44 @@ export class ApiSessionList {
const records = await this.ctx.sessionQuery.listSessions(signal)
signal?.throwIfAborted()
const items: SessionSummary[] = []
const cold: SessionHeader[] = []
const cold: SessionRecord[] = []
let yieldDeadline = performance.now() + this.workSliceMs
for (const record of records) {
signal?.throwIfAborted()
const live = this.ctx.sessions.get(record.header.id)
if (live !== undefined) {
items.push(this.summaryFor(live))
continue
} else if (record.header.cwd !== undefined) {
cold.push(record)
}
if (performance.now() >= yieldDeadline) {
await scheduler.yield()
signal?.throwIfAborted()
yieldDeadline = performance.now() + this.workSliceMs
}
if (record.header.cwd === undefined) continue
cold.push(record.header)
}
for (const header of cold) items.push(this.summarizeCold(header))
for (const record of cold) {
signal?.throwIfAborted()
items.push(this.summarizeCold(record))
if (performance.now() >= yieldDeadline) {
await scheduler.yield()
signal?.throwIfAborted()
yieldDeadline = performance.now() + this.workSliceMs
}
}
signal?.throwIfAborted()
items.sort((left, right) => right.updatedAt - left.updatedAt)
return items
}
private summarizeCold(header: SessionHeader): SessionSummary {
private summarizeCold({ header, formatStatus }: SessionRecord): SessionSummary {
const projections = this.projectionsFor(header, undefined)
const metadata = projections?.values.sessionListMetadata
return {
sessionId: header.id,
updatedAt: updatedAt(header, metadata),
agentAvailable: false,
...(formatStatus === undefined ? {} : { formatStatus }),
running: false,
// A large, metadata-less, or inaccessible cache miss remains unknown and visible.
blank: metadata?.blank ?? false,
packages/api/session-controller/src/skill-catalog.ts
+8 -7
View File
@@ -4,6 +4,7 @@ import type { Context } from '@deepseek-ai/cordis'
import type {} from '@deepseek-ai/dsh-agent-preset-registry/types'
import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
import { isUserInvocable } from '@deepseek-ai/dsh-skill'
import type {} from '@deepseek-ai/dsh-working-directory'
import type { ScopeKey } from '@deepseek-ai/dsh-scope'
import { Remote, RemoteError, TypertRemoteService } from '@deepseek-ai/dsh-typert-protocol'
import type { SkillListRequest, SkillListValue } from './types.ts'
@@ -17,7 +18,7 @@ declare module '@deepseek-ai/cordis' {
/** Host service backing `ctx.remote.skills` without activating a cold Agent. */
export class SessionSkillCatalog extends TypertRemoteService {
static inject = ['agents', 'sessionQuery', 'typert']
static inject = ['agents', 'sessionQuery', 'typert', 'workingDirectory']
/** @param ctx - Host context carrying Session reads and optional skill/preset services. */
constructor(ctx: Context) {
@@ -33,17 +34,20 @@ export class SessionSkillCatalog extends TypertRemoteService {
*/
@Remote
async list(request: SkillListRequest, signal: AbortSignal): Promise<SkillListValue> {
void signal
const { sessionId } = request
let cwd: string | undefined
let cwd: string
let agentPreset: string | undefined
try {
using observation = await this.ctx.sessionQuery.observeSession(sessionId)
if (observation.projections === undefined) {
throw new Error('skill catalog requires a projected Session observation')
}
cwd = observation.header.cwd
cwd = observation.projections.values.workingDirectory
?? observation.header.cwd
?? this.ctx.workingDirectory.defaultDirectory
agentPreset = observation.projections.values.agentPreset ?? undefined
const live = this.ctx.agents.get(sessionId)
if (live !== undefined) cwd = await this.ctx.workingDirectory.ensure(live, signal)
} catch (error: unknown) {
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
@@ -55,9 +59,6 @@ export class SessionSkillCatalog extends TypertRemoteService {
{},
)
}
if (cwd === undefined) {
throw new RemoteError('gateway/internal', `session "${sessionId}" has no project cwd`, {})
}
const live = this.ctx.agents.get(sessionId)
const presets = this.ctx.get('agentPresets')
packages/api/session-controller/src/types.ts
+24 -6
View File
@@ -9,6 +9,7 @@ import type { LlmAttemptId, MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { TextBlock } from '@deepseek-ai/dsh-llm'
import type { SessionId, SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
import type { Principal } from '@deepseek-ai/dsh-typert-protocol'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
@@ -177,6 +178,8 @@ export type QueueAction =
export interface SessionSummary {
/** Whether this Session currently owns a live Agent. */
readonly agentAvailable: boolean
/** Current for attached Sessions; otherwise the persistence format status when reported. */
readonly formatStatus?: 'current' | 'migration-required'
readonly sessionId: SessionId
readonly updatedAt: number
readonly running: boolean
@@ -217,13 +220,13 @@ declare module '@deepseek-ai/dsh-typert-protocol' {
'agent-preset/conflict': {
readonly sessionId: SessionId
readonly requestedPreset: string
readonly existingPreset?: string
}
'session/attachment-invalid': { readonly reason: string }
'session/queue-item-not-found': { readonly itemId: MessageId }
'session/steer-unavailable': { readonly itemId: MessageId }
'session/title-invalid': { readonly sessionId: SessionId }
'session/fork-unavailable': { readonly sessionId: SessionId }
'session/migration-required': { readonly sessionId: SessionId }
'subagent/not-found': {
readonly parentSessionId: SessionId
readonly childSessionId: SessionId
@@ -281,17 +284,23 @@ export interface SessionSearchValue {
readonly hasMore: boolean
}
/** Session creation or explicit-id adoption request. */
/** Request for a Session for a new conversation. */
export interface SessionCreateRequest {
/** Workspace the Session joins; exclusive with `cwd`. */
readonly workspaceId?: WorkspaceId
/** Directory the Session owns when no Workspace is named. */
readonly cwd?: string
/** Blank Session to reuse while it is reusable, or the identity of a new Session when none is stored under it. */
readonly sessionId?: SessionId
/** Preset the Session runs, replacing a reused Session's recorded preset; the configured default when omitted. */
readonly agentPreset?: string
}
/** Session creation response value. */
export interface SessionCreateValue {
/** The created or reused Session, which may differ from the requested `sessionId`. */
readonly sessionId: SessionId
/** Preset the Session runs, when the deployment composes a preset roster. */
readonly agentPreset?: string
}
@@ -322,6 +331,8 @@ export interface SessionForkRequest {
readonly sessionId: SessionId
/** Exact inclusive source event seq; omission selects the latest completed-turn prefix. */
readonly atSeq?: number
/** False asks the Host to check for required migration before forking. Defaults to true. */
readonly allowMigration?: boolean
}
/** Identity of a newly forked Session. */
@@ -399,8 +410,12 @@ export type SessionRequestId = Branded<'session-request-id'>
declare module '@deepseek-ai/dsh-llm' {
interface MessageSourceMap {
/** Browser prompt correlation and optional Host-validated time zone. */
'user-rpc': { kind: 'user'; rpcId: SessionRequestId; clientTimeZone?: string }
/**
* Browser prompt correlation, optional Host-validated time zone, and the
* caller that submitted the prompt. `principal` is written only for a
* caller other than the owner, so an absent `principal` means the owner.
*/
'user-rpc': { kind: 'user'; rpcId: SessionRequestId; clientTimeZone?: string; principal?: Principal }
}
}
@@ -419,8 +434,11 @@ export interface SessionProjectionsRequest {
readonly sessionId: SessionId
}
/** Complete Session projection baseline; null when the Session does not exist. */
export type SessionProjectionsValue = SessionProjectionBaseline | null
/** Exact projections carry their event cut; required migration returns sequence-free cache hints. Null means absent. */
export type SessionProjectionsValue =
| (SessionProjectionBaseline & { readonly kind: 'sequenced' })
| { readonly kind: 'migration-required'; readonly values: SessionProjectionValues }
| null
/** One raw Session event in the Remote journal. */
export interface SessionEventEntry {