Merge origin/master into turtle/fork-name-suffix-5f3435925211
Co-authored-by: imccyu <[email protected]>
This commit is contained in:
commit
4731be7515
3956 files changed
+149568
-54303
No files matched your search
@@ -92,21 +92,11 @@ export type ApiSessionAgentResult =
|
||||
|
||||
/**
|
||||
* Test whether generic Session routing must leave an identity to subagent routing.
|
||||
* @param ctx - Host context carrying the Agent ownership registry.
|
||||
* @param session - attached or live Session whose ownership is tested.
|
||||
* @param agent - live Agent when one exists for the Session.
|
||||
* @param session - Attached or live Session whose origin is tested.
|
||||
* @returns whether subagent routing owns the Session identity.
|
||||
*/
|
||||
export function hasApiSessionSubagentOwner(
|
||||
ctx: Context,
|
||||
session: Pick<Session, 'header'>,
|
||||
agent: Agent | undefined,
|
||||
): boolean {
|
||||
if (session.header.origin === 'subagent') return true
|
||||
const parentId = session.header.parentSession
|
||||
if (parentId === undefined || agent === undefined) return false
|
||||
const parent = ctx.agents.get(parentId)
|
||||
return parent !== undefined && ctx.agents.isOwnedBy(agent.id, parent)
|
||||
export function hasApiSessionSubagentOwner(session: Pick<Session, 'header'>): boolean {
|
||||
return session.header.origin === 'subagent'
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -178,7 +168,7 @@ export class ApiSessionAgentController {
|
||||
const live = this.liveAgent(sessionId)
|
||||
if (live !== undefined) return live
|
||||
const attached = this.ctx.sessions.get(sessionId)
|
||||
if (attached !== undefined && hasApiSessionSubagentOwner(this.ctx, attached, undefined)) {
|
||||
if (attached !== undefined && hasApiSessionSubagentOwner(attached)) {
|
||||
return { error: apiSessionSubagentOwnershipError(sessionId) }
|
||||
}
|
||||
|
||||
@@ -203,7 +193,7 @@ export class ApiSessionAgentController {
|
||||
const raced = this.liveAgent(sessionId)
|
||||
if (raced !== undefined) return raced
|
||||
const racedSession = this.ctx.sessions.get(sessionId)
|
||||
if (racedSession !== undefined && hasApiSessionSubagentOwner(this.ctx, racedSession, undefined)) {
|
||||
if (racedSession !== undefined && hasApiSessionSubagentOwner(racedSession)) {
|
||||
return { error: apiSessionSubagentOwnershipError(sessionId) }
|
||||
}
|
||||
if (error instanceof Error && error.name === 'SessionAlreadyOwnedError') {
|
||||
@@ -286,7 +276,7 @@ export class ApiSessionAgentController {
|
||||
private liveAgent(sessionId: SessionId): ApiSessionAgentResult | undefined {
|
||||
const agent = this.ctx.agents.get(sessionId)
|
||||
if (agent === undefined) return undefined
|
||||
return hasApiSessionSubagentOwner(this.ctx, agent.session, agent)
|
||||
return hasApiSessionSubagentOwner(agent.session)
|
||||
? { error: apiSessionSubagentOwnershipError(sessionId) }
|
||||
: { agent }
|
||||
}
|
||||
@@ -312,13 +302,12 @@ export class ApiSessionAgentController {
|
||||
if (observation.header.id !== sessionId || observation.header.cwd === undefined) {
|
||||
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
|
||||
}
|
||||
if (hasApiSessionSubagentOwner(this.ctx, { header: observation.header }, undefined)) {
|
||||
if (hasApiSessionSubagentOwner({ header: observation.header })) {
|
||||
throw new ApiSessionSubagentOwnership(sessionId)
|
||||
}
|
||||
const composition = await this.composeAgent(this.presetForObservation(observation))
|
||||
const published = this.ctx.sessions.get(sessionId)
|
||||
const live = this.ctx.agents.get(sessionId)
|
||||
if (published !== undefined && hasApiSessionSubagentOwner(this.ctx, published, live)) {
|
||||
if (published !== undefined && hasApiSessionSubagentOwner(published)) {
|
||||
throw new ApiSessionSubagentOwnership(sessionId)
|
||||
}
|
||||
return (await this.ctx.agents.resume({
|
||||
@@ -343,7 +332,7 @@ export class ApiSessionAgentController {
|
||||
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)) {
|
||||
if (attached !== undefined && hasApiSessionSubagentOwner(attached)) {
|
||||
throw new ApiSessionSubagentOwnership(sessionId)
|
||||
}
|
||||
if (live !== undefined) {
|
||||
@@ -363,7 +352,7 @@ export class ApiSessionAgentController {
|
||||
let reusable: boolean
|
||||
{
|
||||
using observed = observation
|
||||
if (hasApiSessionSubagentOwner(this.ctx, { header: observed.header }, undefined)) {
|
||||
if (hasApiSessionSubagentOwner({ header: observed.header })) {
|
||||
throw new ApiSessionSubagentOwnership(sessionId)
|
||||
}
|
||||
if (observed.header.cwd !== request.cwd) {
|
||||
|
||||
@@ -4,7 +4,9 @@
|
||||
* it is restored. The work a Session still runs is reported and stopped by
|
||||
* its owners — the Agent registry (`turn`), the job registry seam (`job`),
|
||||
* the Subagent runtime (`subagent`), and the Schedule plugin (`schedule`) —
|
||||
* through the Workspace registry's archive-admission events.
|
||||
* through the Workspace registry's archive-admission events. The Subagent
|
||||
* runtime also checks its retained ancestry before each step, so a released
|
||||
* intermediate Session cannot hide an archived ancestor.
|
||||
*/
|
||||
|
||||
import type { Context, Plugin } from '@deepseek-ai/cordis'
|
||||
@@ -33,9 +35,10 @@ export const ArchivedSessionGate: Plugin.Object<void> = {
|
||||
|
||||
/**
|
||||
* Whether the Agent's Session, or a Session above it in its subagent lineage,
|
||||
* is archived. Lineage follows the durable header fields through
|
||||
* is archived. This header-only check follows the durable fields through
|
||||
* subagent-origin Sessions only: a fork of an archived Session is an
|
||||
* independent conversation.
|
||||
* independent conversation. The Subagent runtime separately checks managed
|
||||
* ancestry when an intermediate Session is unavailable here.
|
||||
* @param ctx - Host context.
|
||||
* @param agent - the Agent proposing a step.
|
||||
* @returns whether an archived Session owns the step.
|
||||
|
||||
@@ -73,8 +73,6 @@ export interface SessionSnapshot {
|
||||
readonly running: boolean
|
||||
readonly subagent: {
|
||||
readonly address: SubagentAddress
|
||||
/** Absent until the direct-parent catalog resolves. */
|
||||
readonly parentAvailable?: boolean
|
||||
} | null
|
||||
readonly removed: boolean
|
||||
readonly openState: OpenState
|
||||
|
||||
@@ -147,12 +147,7 @@ export class SessionManager {
|
||||
throw new Error(`sessions.retain: unknown session ${id}`)
|
||||
}
|
||||
if (address !== undefined) this.addresses.set(id, address)
|
||||
this.sessions.get(id)?.configureSubagent(
|
||||
address,
|
||||
address === undefined
|
||||
? undefined
|
||||
: this.agentAvailable(address.parentSessionId),
|
||||
)
|
||||
this.sessions.get(id)?.configureSubagent(address)
|
||||
return id
|
||||
}
|
||||
|
||||
@@ -276,14 +271,8 @@ export class SessionManager {
|
||||
|
||||
private createSession(sessionId: SessionId): Session {
|
||||
const address = this.addresses.get(sessionId)
|
||||
const parentAvailable = address === undefined
|
||||
? undefined
|
||||
: this.agentAvailable(address.parentSessionId)
|
||||
return new Session(sessionId, this.remote, {
|
||||
...(address === undefined ? {} : {
|
||||
address,
|
||||
...parentAvailable === undefined ? {} : { parentAvailable },
|
||||
}),
|
||||
...(address === undefined ? {} : { address }),
|
||||
// The sender's local first-send flip mirrors into the list row so the
|
||||
// session surfaces (lists filter on blank) before any host frame lands.
|
||||
onEngaged: (engaged) => {
|
||||
@@ -400,18 +389,6 @@ export class SessionManager {
|
||||
return operation
|
||||
}
|
||||
|
||||
private agentAvailable(sessionId: SessionId): boolean | undefined {
|
||||
return this.summaries.find(summary => summary.sessionId === sessionId)?.agentAvailable
|
||||
?? (this.listPhase === 'ready' ? false : undefined)
|
||||
}
|
||||
|
||||
private updateParentAvailability(): void {
|
||||
for (const [childId, address] of this.addresses) {
|
||||
const available = this.agentAvailable(address.parentSessionId)
|
||||
if (available !== undefined) this.sessions.get(childId)?.handleSubagentParentAvailable(available)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- List API ----
|
||||
|
||||
/** Full refresh via session.list (single-flight within one Host generation). */
|
||||
@@ -446,7 +423,7 @@ export class SessionManager {
|
||||
for (const sessionId of this.engagedSessions) this.pruneEngagement(sessionId, retained)
|
||||
this.listState = 'idle'
|
||||
this.listPhase = 'ready'
|
||||
this.updateParentAvailability()
|
||||
|
||||
// Resident Sessions and list rows share one display blank, reconciled with the metadata projection.
|
||||
for (const s of this.summaries) {
|
||||
const session = this.sessions.get(s.sessionId)
|
||||
@@ -599,7 +576,6 @@ export class SessionManager {
|
||||
*/
|
||||
private mergeSummary(summary: SessionSummary): void {
|
||||
this.recordMutation({ kind: 'upsert', summary })
|
||||
this.updateParentAvailability()
|
||||
}
|
||||
|
||||
/** Apply immediately and retain for replay when a list response is in flight. */
|
||||
@@ -642,15 +618,17 @@ export class SessionManager {
|
||||
// ---- Live control and Host-event sinks ----
|
||||
|
||||
/**
|
||||
* Apply a complete control baseline or one later replacement frame.
|
||||
* @param frame - baseline or live control replacement from Session Controller.
|
||||
* Apply a complete control baseline or one later per-event projection frame.
|
||||
* @param frame - baseline or live per-event projection change from Session Controller.
|
||||
*/
|
||||
handleControlFrame(frame: SessionControlFrame): void {
|
||||
if (frame.type === 'baseline') {
|
||||
this.replaceControlBaseline(frame.value)
|
||||
return
|
||||
}
|
||||
this.projectionStore(frame.sessionId).apply(frame.key, frame.value, SessionSeq(frame.seq))
|
||||
const store = this.projectionStore(frame.sessionId)
|
||||
const seq = SessionSeq(frame.seq)
|
||||
for (const change of frame.changes) store.apply(change.key, change.val, seq)
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
|
||||
@@ -719,11 +697,6 @@ export class SessionManager {
|
||||
this.projectionInflight.get(sessionId)?.controller.abort()
|
||||
this.projectionInflight.delete(sessionId)
|
||||
this.projectionLoads.delete(sessionId)
|
||||
for (const [childId, address] of this.addresses) {
|
||||
if (address.parentSessionId === sessionId) {
|
||||
this.sessions.get(childId)?.handleSubagentParentAvailable(false)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -734,7 +707,6 @@ export class SessionManager {
|
||||
handleSessionStatus(sessionId: SessionId, running: boolean): void {
|
||||
if (!this.disposed && running) this.engagedSessions.add(sessionId)
|
||||
this.recordMutation({ kind: 'status', sessionId, running, agentAvailable: true })
|
||||
this.updateParentAvailability()
|
||||
this.sessions.get(sessionId)?.handleRunning(running)
|
||||
}
|
||||
|
||||
|
||||
@@ -30,10 +30,8 @@ export interface SessionSubagentsRemote {
|
||||
request: SubagentPromptRequest,
|
||||
signal?: AbortSignal,
|
||||
): Promise<RemoteResult<SubagentPromptReceipt>>
|
||||
interruptByParent(
|
||||
interruptByUser(
|
||||
childSessionId: SessionId,
|
||||
parentSessionId: SessionId,
|
||||
mode: 'continuable',
|
||||
): Promise<RemoteResult<SubagentInterruptReceipt>>
|
||||
}
|
||||
|
||||
|
||||
@@ -71,8 +71,6 @@ interface PendingHistory {
|
||||
export interface SessionOptions {
|
||||
/** Catalog-discovered address selecting non-activating subagent transport. */
|
||||
address?: SubagentAddress
|
||||
/** Whether the exact direct parent Agent is available in Host summaries; absent until known. */
|
||||
parentAvailable?: boolean
|
||||
/**
|
||||
* Publish each accepted prompt to the Manager, including after this Session
|
||||
* object is replaced. Acceptance converts display state, but does not
|
||||
@@ -113,7 +111,6 @@ export class Session implements SessionFace {
|
||||
private readonly assistantStream = new ClientAssistantStream()
|
||||
private running = false
|
||||
private address: SubagentAddress | undefined
|
||||
private parentAvailable: boolean | undefined
|
||||
/**
|
||||
* Sticky send marker, private input of the composerPhase derivation: set
|
||||
* when the Host accepts a prompt, never reset — the blank → engaging edge of
|
||||
@@ -193,7 +190,6 @@ export class Session implements SessionFace {
|
||||
) {
|
||||
this.projections = options.projections ?? new ProjectionValueStore()
|
||||
this.address = options.address
|
||||
this.parentAvailable = options.parentAvailable
|
||||
this.notifier = new Notifier(() => {
|
||||
this.snapshotCache = this.buildSnapshot()
|
||||
})
|
||||
@@ -294,9 +290,7 @@ export class Session implements SessionFace {
|
||||
const routedContent = content as Exclude<PromptContentPart, { readonly type: 'file' }>[]
|
||||
const routed = await this.remote.subagents.prompt({
|
||||
requestId: requestId ?? randomUUID() as SessionRequestId,
|
||||
parentSessionId: this.address.parentSessionId,
|
||||
childSessionId: this.address.childSessionId,
|
||||
mode: 'continuable',
|
||||
delivery: mode,
|
||||
content: routedContent,
|
||||
clientTimeZone: resolvedClientTimeZone(),
|
||||
@@ -363,18 +357,14 @@ export class Session implements SessionFace {
|
||||
/**
|
||||
* Stop the active turn while the Host preserves pending inbox work; failures
|
||||
* land in promptError (same error-strip display slot). A subagent address
|
||||
* routes through `subagents.interruptByParent`, whose durable parent-address
|
||||
* authority works without a live parent Agent.
|
||||
* routes through `subagents.interruptByUser`, authorized on the target Session
|
||||
* without requiring a live parent Agent.
|
||||
* @returns the cancel result.
|
||||
*/
|
||||
async cancel(): Promise<RemoteResult<{ accepted: true }>> {
|
||||
const address = this.address
|
||||
const result = address !== undefined
|
||||
? await this.remote.subagents.interruptByParent(
|
||||
address.childSessionId,
|
||||
address.parentSessionId,
|
||||
'continuable',
|
||||
)
|
||||
? await this.remote.subagents.interruptByUser(address.childSessionId)
|
||||
: await this.remote.session.cancel({ sessionId: this.sessionId })
|
||||
if (!result.ok) {
|
||||
this.promptError = { op: 'stop', error: result.error }
|
||||
@@ -562,28 +552,16 @@ export class Session implements SessionFace {
|
||||
* Install or clear the catalog-discovered transport address. A changed
|
||||
* address rebuilds an already-open window through its new history route.
|
||||
* @param address - direct parent/child address, or undefined for ordinary transport.
|
||||
* @param parentAvailable - latest exact-parent availability hint, or undefined before a catalog read.
|
||||
*/
|
||||
configureSubagent(address: SubagentAddress | undefined, parentAvailable?: boolean): void {
|
||||
configureSubagent(address: SubagentAddress | undefined): void {
|
||||
const same = this.address?.parentSessionId === address?.parentSessionId
|
||||
&& this.address?.childSessionId === address?.childSessionId
|
||||
&& this.address?.mode === address?.mode
|
||||
this.address = address
|
||||
this.parentAvailable = parentAvailable
|
||||
if (!same && this.openState !== 'cold') void this.resync()
|
||||
else this.notifier.markDirty()
|
||||
}
|
||||
|
||||
/**
|
||||
* Update only the parent availability hint from a catalog refresh.
|
||||
* @param available - whether the exact direct parent is live.
|
||||
*/
|
||||
handleSubagentParentAvailable(available: boolean): void {
|
||||
if (this.parentAvailable === available) return
|
||||
this.parentAvailable = available
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply the Manager's effective display blank, further reconciled with the
|
||||
* current `sessionListMetadata` projection. Local send attempts and current
|
||||
@@ -934,7 +912,6 @@ export class Session implements SessionFace {
|
||||
address: this.address.mode === 'unknown' && identity != null
|
||||
? { ...this.address, mode: identity.mode }
|
||||
: this.address,
|
||||
...(this.parentAvailable === undefined ? {} : { parentAvailable: this.parentAvailable }),
|
||||
},
|
||||
removed: this.removed,
|
||||
openState: this.openState,
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
/** Session commands whose activation policy is explicit at each Remote method. */
|
||||
|
||||
import type { ProjectionHandle } from '@deepseek-ai/dsh-session-projection'
|
||||
import { continuationProjectionDefinition, type ContinuationState } from './continuation.ts'
|
||||
|
||||
import { modelAvailable } from './catalog.ts'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
@@ -59,7 +62,7 @@ type PromptContentCandidate =
|
||||
/**
|
||||
* 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.
|
||||
* @returns whether any part supplies ordinary prompt content.
|
||||
*/
|
||||
export function hasPromptContent(content: readonly PromptContentCandidate[]): boolean {
|
||||
return content.some(part => part.type !== 'text' || part.text.trim().length > 0)
|
||||
@@ -83,6 +86,13 @@ function latestCompletedPrefixBoundary(events: readonly SessionEvent[]): Session
|
||||
|
||||
/** Implements Session business commands delegated by the Session Controller Remote service. */
|
||||
export class SessionCommandController {
|
||||
private continuation!: ProjectionHandle<ContinuationState>
|
||||
|
||||
/** Register complete-history empty-prompt eligibility. @returns when command admission is ready. */
|
||||
async initialize(): Promise<void> {
|
||||
this.continuation = await this.ctx.sessionProjections.register(continuationProjectionDefinition)
|
||||
}
|
||||
|
||||
/**
|
||||
* @param ctx - Host context carrying Agent, model, attachment, title, and Workspace services.
|
||||
* @param agents - sole owner of create, resume, and Agent composition.
|
||||
@@ -298,12 +308,28 @@ export class SessionCommandController {
|
||||
* 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.
|
||||
* @param source - caller metadata recorded on the accepted ordinary or retry user message.
|
||||
* @returns acknowledgement that the Agent accepted the prompt.
|
||||
*/
|
||||
async prompt(request: SessionPromptRequest, source: MessageSource): Promise<SessionPromptValue> {
|
||||
const agent = await this.resolveAgent(request.sessionId)
|
||||
if (hasPromptRequest(agent, request.requestId)) return { accepted: true }
|
||||
if (request.content.length === 0) {
|
||||
try {
|
||||
return await agent.runMaintenance(() => {
|
||||
const inbox = agent.inbox
|
||||
if (request.mode !== 'queue' || this.continuation.stateOf(agent.session).ending === null
|
||||
|| inbox.nextTurn.length > 0 || inbox.nextStep.some(message => message.source.kind === 'user')) {
|
||||
throw new RemoteError('session/agent-busy', 'generation continuation is unavailable', { reason: 'NO_CONTINUABLE_FAILURE_OR_PENDING_INPUT' })
|
||||
}
|
||||
agent.followup(createUserMessage({ content: [], source }))
|
||||
return Promise.resolve({ accepted: true as const })
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
if (remoteErrorOf(error) !== undefined) throw error
|
||||
throw new RemoteError('session/agent-busy', 'generation continuation was not accepted', { reason: String(error) })
|
||||
}
|
||||
}
|
||||
const admission = resolvePromptFileReceipts(
|
||||
request.content,
|
||||
receiptId => this.ctx.fileUploads.resolve(agent, receiptId),
|
||||
@@ -409,7 +435,7 @@ export class SessionCommandController {
|
||||
}
|
||||
agent = found.agent
|
||||
}
|
||||
if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
|
||||
if (hasApiSessionSubagentOwner(agent.session)) {
|
||||
const identity = this.ctx.sessionProjections
|
||||
.snapshot(agent.session, ['subagent'])
|
||||
.values.subagent
|
||||
@@ -470,7 +496,7 @@ export class SessionCommandController {
|
||||
{ sessionId: request.sessionId },
|
||||
)
|
||||
}
|
||||
if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
|
||||
if (hasApiSessionSubagentOwner(agent.session)) {
|
||||
throw apiSessionSubagentOwnershipError(request.sessionId)
|
||||
}
|
||||
agent.cancel({ kind: 'user' }, { keepInbox: true })
|
||||
@@ -542,13 +568,12 @@ function resolvePromptFileReceipts(
|
||||
function hasPromptRequest(agent: Agent, requestId: SessionRequestId): boolean {
|
||||
const matches = (message: UserMessage): boolean => {
|
||||
const source = message.source
|
||||
return source.kind === 'user' && 'rpcId' in source && source.rpcId === requestId
|
||||
return (source.kind === 'user' || source.kind === 'retry') && 'rpcId' in source && source.rpcId === requestId
|
||||
}
|
||||
if (agent.inbox.nextTurn.some(matches) || agent.inbox.nextStep.some(matches)) return true
|
||||
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
|
||||
return agent.session.snapshotEvents().some((event) => {
|
||||
if (event.type !== 'user/message') return false
|
||||
const source = event.data.source
|
||||
return source.kind === 'user' && 'rpcId' in source && source.rpcId === requestId
|
||||
return matches(event.data)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
/** Constant-size continuation eligibility over the complete durable Session. */
|
||||
import type {} from './types.ts'
|
||||
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
|
||||
import type { SessionSeq } from '@deepseek-ai/dsh-session/types'
|
||||
import { brandNumber } from '@deepseek-ai/dsh-brand'
|
||||
import { z } from 'zod'
|
||||
|
||||
// Explicit manual requests use a fixed product policy, independent of automatic retry configuration.
|
||||
const CONTINUABLE_FAILURES: ReadonlySet<string> = new Set([
|
||||
'RATE_LIMIT', 'SERVER', 'TIMEOUT', 'TRANSPORT', 'EMPTY_RESPONSE', 'STREAM_CLOSED',
|
||||
])
|
||||
|
||||
const endingSchema = z.number().int().nonnegative().transform(brandNumber<SessionSeq>).nullable()
|
||||
const stateSchema = z.object({
|
||||
admitted: z.boolean(),
|
||||
failedAttempt: z.boolean(),
|
||||
ending: endingSchema,
|
||||
})
|
||||
|
||||
/** Recoverable manual-continuation facts over admitted user input and empty retry messages. */
|
||||
export type ContinuationState = z.infer<typeof stateSchema>
|
||||
|
||||
const EMPTY: ContinuationState = { admitted: false, failedAttempt: false, ending: null }
|
||||
|
||||
/** Eligibility requires admitted human work and a failed settlement in the last request step. */
|
||||
export const continuationProjectionDefinition = {
|
||||
key: 'conversationContinuation',
|
||||
stateVersion: 4,
|
||||
stateSchema,
|
||||
init: (): ContinuationState => EMPTY,
|
||||
apply: (state, event) => {
|
||||
switch (event.type) {
|
||||
case 'turn/start':
|
||||
return EMPTY
|
||||
case 'step/start':
|
||||
case 'assistant/message':
|
||||
return state.failedAttempt ? { ...state, failedAttempt: false } : state
|
||||
case 'user/message':
|
||||
return !state.admitted && event.surfaceOp === 'append'
|
||||
&& ((event.data.source.kind === 'user' && event.data.content.length > 0)
|
||||
|| (event.data.source.kind === 'retry' && event.data.content.length === 0))
|
||||
? { ...state, admitted: true } : state
|
||||
case 'assistant/attempt':
|
||||
return state.failedAttempt ? state : { ...state, failedAttempt: true }
|
||||
case 'turn/end': {
|
||||
const ending = event.data.reason.kind === 'error' && CONTINUABLE_FAILURES.has(event.data.reason.error.code)
|
||||
&& state.admitted && state.failedAttempt ? event.seq : null
|
||||
return ending === null ? EMPTY : { ...EMPTY, ending }
|
||||
}
|
||||
default:
|
||||
return state
|
||||
}
|
||||
},
|
||||
wire: {
|
||||
viewSchema: endingSchema,
|
||||
view: (state: ContinuationState) => state.ending,
|
||||
},
|
||||
} satisfies ProjectionDefinition<'conversationContinuation', ContinuationState>
|
||||
@@ -5,11 +5,11 @@ import { Deque } from '@deepseek-ai/dsh-deque'
|
||||
import type {
|
||||
Session, SessionId,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import type {
|
||||
SessionControlBaseline,
|
||||
SessionControlFrame,
|
||||
SessionProjectionBaseline,
|
||||
SessionProjectionUpdate,
|
||||
SessionProjectionValues,
|
||||
} from './types.ts'
|
||||
|
||||
@@ -19,13 +19,13 @@ export class SessionControlController {
|
||||
|
||||
/** @param ctx - Host context carrying live Agent and projection services. */
|
||||
constructor(private readonly ctx: Context) {
|
||||
ctx.sessionProjections.onChanged((session, key, value, seq) => {
|
||||
ctx.sessionProjections.onEventChanged((session, changes, seq) => {
|
||||
this.broadcast({
|
||||
type: 'projection',
|
||||
sessionId: session.id,
|
||||
key,
|
||||
value: value as JsonValue,
|
||||
seq,
|
||||
// Every projection definition validates its value before change publication.
|
||||
changes: changes as SessionProjectionUpdate['changes'],
|
||||
})
|
||||
})
|
||||
ctx.effect(() => () => {
|
||||
|
||||
@@ -103,10 +103,10 @@ export class HistoryController {
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to sparse projection replacements without reading a baseline.
|
||||
* Subscribe to projection changes without reading a baseline.
|
||||
* @param request - Session identity to subscribe without reading a baseline.
|
||||
* @param signal - stream cancellation.
|
||||
* @returns readiness followed by sparse per-key replacement values.
|
||||
* @returns readiness followed by one change frame per committed event, listing every key it changed.
|
||||
*/
|
||||
async *observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
|
||||
using listener = new ProjectionListener(this.ctx, request.sessionId, signal)
|
||||
|
||||
@@ -5,7 +5,6 @@ import { Deque } from '@deepseek-ai/dsh-deque'
|
||||
import { SessionLogOffset } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import { entryFor } from '../history.ts'
|
||||
import type { SessionProjectionChange, SessionEventEntry } from '../types.ts'
|
||||
|
||||
@@ -128,7 +127,7 @@ export function eventItem(item: EventItem, next?: number): {
|
||||
return { events, next }
|
||||
}
|
||||
|
||||
/** Sparse projection changes retain every key, including keys sharing a seq. */
|
||||
/** Projection changes of one Session, one change per committed event that changed a projection. */
|
||||
export class ProjectionListener implements Disposable {
|
||||
private readonly buffer: ListenerBuffer<{ session: Session; change: SessionProjectionChange }>
|
||||
|
||||
@@ -142,12 +141,12 @@ export class ProjectionListener implements Disposable {
|
||||
const input = this.buffer.createInput<{ session: Session; change: SessionProjectionChange }>(item => [item])
|
||||
const target = sessionId
|
||||
try {
|
||||
this.buffer.own(ctx.sessionProjections.onChanged((session, key, value, seq) => {
|
||||
this.buffer.own(ctx.sessionProjections.onEventChanged((session, changes, seq) => {
|
||||
if (session.id !== target) return
|
||||
this.buffer.capture(() => {
|
||||
input.append({
|
||||
session, change: { type: 'projection', key, value: value as JsonValue, seq } satisfies SessionProjectionChange,
|
||||
})
|
||||
// Every projection definition validates its value before change publication.
|
||||
const change: SessionProjectionChange = { type: 'projection', seq, changes: changes as SessionProjectionChange['changes'] }
|
||||
input.append({ session, change })
|
||||
})
|
||||
}))
|
||||
} catch (error: unknown) {
|
||||
@@ -156,8 +155,8 @@ export class ProjectionListener implements Disposable {
|
||||
}
|
||||
}
|
||||
|
||||
/** Consume sparse per-key projection replacements.
|
||||
* @returns validated replacement frames; projection seqs need not be contiguous. */
|
||||
/** Consume per-event projection changes.
|
||||
* @returns validated change frames, one per committed event; projection seqs need not be contiguous. */
|
||||
async *changes(): AsyncIterable<SessionProjectionChange> {
|
||||
try {
|
||||
for await (const { change } of this.buffer.items()) {
|
||||
|
||||
@@ -209,7 +209,10 @@ export class SessionController extends TypertRemoteService {
|
||||
this.stateController = new StateController(ctx, this.historyController)
|
||||
this.webController = new WebController(ctx, this.historyController, (observation) => { this.promote(observation) })
|
||||
this.listState = new ApiSessionList(ctx, this.listWorkSliceMs)
|
||||
await this.commands.initialize()
|
||||
ctx.fiber.assertActive()
|
||||
await this.listState.initialize()
|
||||
ctx.fiber.assertActive()
|
||||
ctx.plugin(SessionFileReferences)
|
||||
ctx.plugin(SessionMediaReferences)
|
||||
ctx.plugin(SessionSkillCatalog)
|
||||
@@ -464,7 +467,7 @@ export class SessionController extends TypertRemoteService {
|
||||
@Remote('prompt')
|
||||
async prompt(request: SessionPromptRequest, signal: AbortSignal): Promise<SessionPromptValue> {
|
||||
signal.throwIfAborted()
|
||||
if (!hasPromptContent(request.content)) {
|
||||
if (request.content.length > 0 && !hasPromptContent(request.content)) {
|
||||
throw new RemoteError('gateway/bad-request', 'prompt content must include non-whitespace text or an attachment', {})
|
||||
}
|
||||
const clientTimeZone = request.clientTimeZone === undefined
|
||||
@@ -480,7 +483,7 @@ export class SessionController extends TypertRemoteService {
|
||||
// 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',
|
||||
kind: request.content.length === 0 ? 'retry' : 'user',
|
||||
rpcId: request.requestId,
|
||||
...(clientTimeZone === undefined ? {} : { clientTimeZone }),
|
||||
...(principal === undefined ? {} : { principal }),
|
||||
@@ -553,10 +556,10 @@ export class SessionController extends TypertRemoteService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to sparse Session projection changes.
|
||||
* Subscribe to Session projection changes.
|
||||
* @param request - target Session.
|
||||
* @param signal - subscription cancellation.
|
||||
* @returns readiness and sparse projection changes.
|
||||
* @returns readiness followed by one projection change frame per committed event.
|
||||
*/
|
||||
@Remote({ mode: 'stream' })
|
||||
observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
|
||||
|
||||
@@ -6,7 +6,7 @@ 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 { ProjectionHandle, ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection'
|
||||
import type {} from '@deepseek-ai/dsh-session-projection-cache'
|
||||
import { SessionQueryError, type SessionRecord, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
|
||||
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
|
||||
@@ -76,6 +76,8 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
|
||||
|
||||
/** Owns list projection registration, bounded cold summaries, and authorized search. */
|
||||
export class ApiSessionList {
|
||||
private listMetadata!: ProjectionHandle<SessionListMetadata>
|
||||
|
||||
/**
|
||||
* @param ctx - Host context carrying Session, query, persistence, and projection services.
|
||||
* @param workSliceMs - Resolved positive integral list-work budget in milliseconds.
|
||||
@@ -87,7 +89,7 @@ export class ApiSessionList {
|
||||
* @returns when all required registrations are ready.
|
||||
*/
|
||||
async initialize(): Promise<void> {
|
||||
await this.ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
|
||||
this.listMetadata = await this.ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
|
||||
key: 'sessionListMetadata',
|
||||
stateSchema: sessionListMetadataSchema,
|
||||
init: () => ({ blank: true, lastPromptAt: null }),
|
||||
@@ -111,19 +113,25 @@ export class ApiSessionList {
|
||||
|
||||
/**
|
||||
* Build one current attached-Session summary.
|
||||
* Blankness and recency come from the `sessionListMetadata` handle; projection
|
||||
* hints carry every summary key computed through `snapshot`. The first call
|
||||
* for a Session computes these projections over its whole in-memory log, and
|
||||
* later calls advance them incrementally. The handle read and the hint with
|
||||
* the same key come from one cell; the handle read does not depend on any
|
||||
* other unit's view.
|
||||
* @param session - attached Session to summarize.
|
||||
* @returns current list metadata and available projections.
|
||||
*/
|
||||
summaryFor(session: Session): SessionSummary {
|
||||
const metadata = this.listMetadata.stateOf(session)
|
||||
const projections = this.projectionsFor(session.header, session)
|
||||
const metadata = projections?.values.sessionListMetadata
|
||||
return {
|
||||
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,
|
||||
blank: metadata.blank,
|
||||
...listFields(session.header),
|
||||
...(projections === undefined ? {} : { projections }),
|
||||
}
|
||||
@@ -304,9 +312,9 @@ export class ApiSessionList {
|
||||
// itself summary-excluded.
|
||||
const keys = this.ctx.sessionProjections.summaryKeys()
|
||||
if (session !== undefined) {
|
||||
// The live registry computed the block for this Session: its watermark
|
||||
// The live registry computes the block for this Session: its watermark
|
||||
// shares the sequence space of the Session's baselines and frames.
|
||||
return hintsOf('sequenced', this.ctx.sessionProjections.cachedSnapshot(session, keys))
|
||||
return hintsOf('sequenced', this.ctx.sessionProjections.snapshot(session, keys))
|
||||
}
|
||||
// A cold row reads the persisted cache by header alone; the cache serves
|
||||
// seeded and unseeded lifecycles alike because a listing never seeds a
|
||||
@@ -334,8 +342,8 @@ function hintsOf(
|
||||
block: ProjectionSnapshot | undefined,
|
||||
): SessionProjectionHints | undefined {
|
||||
if (block === undefined || Object.keys(block.values).length === 0) return undefined
|
||||
// Listing hints contain every summary wire value the source currently holds
|
||||
// but remain partial: missing cells and cache rows are never materialized here.
|
||||
// Listing hints contain every summary wire value the source holds; cached
|
||||
// hints remain partial because missing cache rows are never folded here.
|
||||
return { kind, asOfSeq: block.asOfSeq, values: block.values }
|
||||
}
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import type {
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { LlmAttemptId, MessageId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { LlmModelBadge, TextBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionId, SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
|
||||
import type { SessionId, SessionSeq, SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
|
||||
import type { SessionProjectionMap, SessionProjectionSummaryExcluded, SessionProjectionSummaryValues } 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'
|
||||
@@ -16,6 +16,8 @@ import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
|
||||
|
||||
declare module '@deepseek-ai/dsh-session-projection/types' {
|
||||
interface SessionProjectionMap {
|
||||
/** Eligible latest failed turn/end for an empty prompt, or null. */
|
||||
conversationContinuation: SessionSeq | null
|
||||
/** Persisted facts used to summarize a Session without activating it. */
|
||||
sessionListMetadata: SessionListMetadata
|
||||
/**
|
||||
@@ -36,9 +38,10 @@ export interface SessionListMetadata {
|
||||
}
|
||||
|
||||
/**
|
||||
* Every available summary wire value a Session-list row carries as partial,
|
||||
* possibly stale hints. The block omits the keys whose unit declares
|
||||
* `summary: false`. `kind` and `asOfSeq` are independent facts: `kind` says
|
||||
* Summary wire values a Session-list row carries. A `sequenced` block carries
|
||||
* every summary key computed at its `asOfSeq`; a `cached` block carries the
|
||||
* values the persisted cache holds and may omit keys or trail the Session.
|
||||
* The block omits the keys whose unit declares `summary: false`. `kind` and `asOfSeq` are independent facts: `kind` says
|
||||
* which sequence space `asOfSeq` belongs to, and therefore how a client may
|
||||
* merge the block; `asOfSeq` is the producer's watermark in that space.
|
||||
*/
|
||||
@@ -319,11 +322,11 @@ export interface SessionForkValue {
|
||||
|
||||
/** Session prompt request. */
|
||||
export interface SessionPromptRequest {
|
||||
/** Client-minted identity persisted on the exact accepted user message. */
|
||||
/** Client-minted identity persisted on the accepted inbox input, including empty continuation. */
|
||||
readonly requestId: SessionRequestId
|
||||
readonly sessionId: SessionId
|
||||
readonly mode: 'queue' | 'steer'
|
||||
/** At least one non-whitespace text part or attachment. */
|
||||
/** Non-whitespace text or attachments; an empty array continues an eligible failed idle Session. */
|
||||
readonly content: readonly PromptContentPart[]
|
||||
readonly clientTimeZone?: string
|
||||
}
|
||||
@@ -395,6 +398,8 @@ declare module '@deepseek-ai/dsh-llm' {
|
||||
* caller other than the owner, so an absent `principal` means the owner.
|
||||
*/
|
||||
'user-rpc': { kind: 'user'; rpcId: SessionRequestId; clientTimeZone?: string; principal?: Principal }
|
||||
/** Empty retry user message with browser prompt correlation and caller metadata. */
|
||||
'retry-rpc': { kind: 'retry'; rpcId: SessionRequestId; clientTimeZone?: string; principal?: Principal }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -569,12 +574,21 @@ export interface SessionControlBaseline {
|
||||
readonly projections: Readonly<Record<SessionId, SessionProjectionBaseline>>
|
||||
}
|
||||
|
||||
/** One finished projection value and its durable watermark. */
|
||||
/**
|
||||
* Projection changes caused by one committed Session event; the Host emits one
|
||||
* update per event that changed at least one client-visible projection view.
|
||||
*/
|
||||
export interface SessionProjectionUpdate {
|
||||
readonly sessionId: SessionId
|
||||
readonly key: string
|
||||
readonly value: JsonValue
|
||||
/** Seq of the committed event that caused every listed change. */
|
||||
readonly seq: number
|
||||
/**
|
||||
* One entry per client-visible key whose view the event changed, in
|
||||
* registration order without duplicate keys, with the key's new value as
|
||||
* `val`. A key absent from `changes` is unchanged; unlike a baseline, absence
|
||||
* does not mean an absent capability.
|
||||
*/
|
||||
readonly changes: readonly { readonly key: string; readonly val: SessionProjectionValue }[]
|
||||
}
|
||||
|
||||
/** Host-wide live state stream. Each generation starts with exactly one baseline. */
|
||||
@@ -652,10 +666,10 @@ export type SessionEventFrame =
|
||||
| { readonly type: 'ready' }
|
||||
| SessionEventEntry
|
||||
|
||||
/** One sparse per-key replacement; different keys can share a seq. */
|
||||
/** Projection changes caused by one committed event of the observed Session. */
|
||||
export type SessionProjectionChange = { readonly type: 'projection' } & Omit<SessionProjectionUpdate, 'sessionId'>
|
||||
|
||||
/** Subscription readiness without a baseline, followed by per-key changes. */
|
||||
/** Subscription readiness without a baseline, followed by one change frame per committed event. */
|
||||
export type SessionProjectionFrame =
|
||||
| { readonly type: 'ready' }
|
||||
| SessionProjectionChange
|
||||
|
||||
Reference in new issue
Block a user