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

feat(web): rewrite subagent conversations for FIFO activation

Browse Source
This commit is contained in:
Dudu-0223
2026-07-30 23:33:07 +08:00
committed by Tianyi Cui
parent f0ab04273d
commit 8a518e353b
52 changed files with 829 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -18,6 +18,8 @@ import { isAppendSurfaceEvent, lastActivityTime } from '@deepseek-ai/dsh-session
import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
import { SubagentError } from '@deepseek-ai/dsh-subagent'
import type { SubagentListEntry as CatalogSubagentListEntry } from '@deepseek-ai/dsh-subagent'
import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
import {
workspaceDomainState, workspaceRecord, WorkspaceId as brandWorkspaceId,
@@ -28,7 +30,7 @@ import type {} from '@deepseek-ai/dsh-tools'
import type {
ApiProxy, CredentialView, GoalRef, HistoryEntry, HostFrame, ModelCatalogFailure, ModelProviderGroup,
ModelReasoning, MuxFrame, QuestionResponsePayload, SessionProjectionsBlock, SessionSearchItem,
SessionSummary, SettingsNamespaceView, SubagentListEntry as SubagentCatalogEntry, ToolEventView,
SessionSummary, SettingsNamespaceView, SubagentAddress, ToolEventView,
WorkspaceId, WorkspaceView,
} from './api/index.ts'
import {
@@ -69,8 +71,6 @@ import type {
import { UserInteractionError } from '@deepseek-ai/dsh-user-interaction'
import { DirectoryPickerError } from '@deepseek-ai/dsh-host-directory-picker'
import { openNativePath } from './native-path-opener.ts'
import { SubagentError } from '@deepseek-ai/dsh-subagent'
import type { SubagentListEntry as CoreSubagentListEntry } from '@deepseek-ai/dsh-subagent'
/** Page size when history is called without maxMessages. */
const DEFAULT_MAX_MESSAGES = 50
@@ -484,92 +484,6 @@ function historyPage(
}
}
/** Map a continuation failure without exposing descriptor or provider details. */
function subagentPromptError(
request: RpcRequest<{ childSessionId: SessionId }>,
error: unknown,
signal?: AbortSignal,
): RpcResponse<never> {
const childSessionId = request.payload.childSessionId
if (signal?.aborted) {
return err(request, { code: 'cancelled', message: 'subagent prompt was cancelled', details: {} })
}
if (error instanceof SubagentError) {
switch (error.code) {
case 'NOT_RESUMABLE':
return err(request, { code: 'subagent-not-resumable', message: 'subagent cannot be resumed', details: { childSessionId } })
case 'UNAUTHORIZED':
return err(request, { code: 'subagent-unauthorized', message: 'subagent does not belong to this parent', details: { childSessionId } })
case 'ACTIVATION_CLOSING':
case 'DRAINING':
return err(request, { code: 'subagent-not-delivered', message: 'message was not delivered', details: { childSessionId } })
default:
break
}
}
return err(request, { code: 'internal', message: 'subagent prompt failed', details: {} })
}
/** Verify one address against the complete durable direct-child catalog. */
async function healthyCatalogChild(
ctx: Context,
parentSessionId: SessionId,
childSessionId: SessionId,
signal?: AbortSignal,
): Promise<{ error?: RpcError }> {
try {
const entries = await ctx.subagents.listChildren(parentSessionId, signal)
const entry = entries.find(candidate => candidate.id === childSessionId)
if (entry === undefined || (entry.kind === 'child' && entry.mode !== 'continuable')) {
return {
error: {
code: 'subagent-not-found',
message: `session "${childSessionId}" is not a continuable direct child of "${parentSessionId}"`,
details: { parentSessionId, childSessionId },
},
}
}
if (entry.kind === 'diagnostic') {
return {
error: {
code: 'subagent-catalog-diagnostic',
message: `subagent "${childSessionId}" is ${entry.reason}`,
details: { parentSessionId, childSessionId, reason: entry.reason },
},
}
}
return {}
} catch (error: unknown) {
if (error instanceof SubagentError && error.code === 'CANCELLED') {
return { error: { code: 'cancelled', message: 'subagent catalog read was cancelled', details: {} } }
}
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return {
error: {
code: 'subagent-not-found',
message: `parent session "${parentSessionId}" is unavailable`,
details: { parentSessionId, childSessionId },
},
}
}
return { error: { code: 'internal', message: 'subagent catalog read failed', details: {} } }
}
}
/** Project the durable catalog onto the continuable-only browser surface. */
function continuableCatalog(entries: readonly CoreSubagentListEntry[]): SubagentCatalogEntry[] {
return entries.flatMap((entry): SubagentCatalogEntry[] => {
if (entry.kind === 'diagnostic') return [entry]
if (entry.mode !== 'continuable') return []
return [{
kind: 'child',
id: entry.id,
label: entry.label,
activity: entry.activity,
}]
})
}
/**
* The projection baseline for one history tail page: the registry's
* watermark-cache snapshot — one fully synchronous read (no await between the
@@ -607,6 +521,107 @@ function listProjectionsFor(ctx: Context, meta: SessionHeader, session: Session
}
}
/** Projection baseline for a detached history tail without Agent activation. */
function detachedProjectionsFor(
ctx: Context,
events: readonly SessionEvent[],
): SessionProjectionsBlock | undefined {
const registry = ctx.get('sessionProjections')
if (registry === undefined) return undefined
return registry.restore({}, events, 0).snapshot
}
/** Map continuation admission failures without exposing provider details. */
function subagentPromptError(
request: RpcRequest<{ childSessionId: SessionId }>,
error: unknown,
signal: AbortSignal,
): RpcResponse<never> {
const childSessionId = request.payload.childSessionId
if (signal.aborted) {
return err(request, { code: 'cancelled', message: 'subagent prompt was cancelled', details: {} })
}
if (error instanceof SubagentError) {
switch (error.code) {
case 'NOT_RESUMABLE':
return err(request, {
code: 'subagent-not-resumable',
message: 'subagent cannot be resumed',
details: { childSessionId },
})
case 'UNAUTHORIZED':
return err(request, {
code: 'subagent-unauthorized',
message: 'subagent does not belong to this parent',
details: { childSessionId },
})
case 'DRAINING':
case 'ACTIVATION_CLOSING':
case 'CONTINUATION_UNAVAILABLE':
case 'PERSISTENCE_UNAVAILABLE':
return err(request, {
code: 'subagent-delivery-unavailable',
message: 'subagent follow-up is temporarily unavailable',
details: { childSessionId },
})
default:
break
}
}
return err(request, { code: 'internal', message: 'subagent prompt failed', details: {} })
}
/** Verify one address and mode against the complete direct-child catalog. */
async function catalogChild(
ctx: Context,
address: SubagentAddress,
signal?: AbortSignal,
): Promise<{
entry?: Extract<CatalogSubagentListEntry, { kind: 'child' }>
error?: RpcError
}> {
const { parentSessionId, childSessionId, mode } = address
try {
const entries = await ctx.subagents.listChildren(parentSessionId, signal)
const entry = entries.find(candidate => candidate.id === childSessionId)
if (entry === undefined || (entry.kind === 'child' && entry.mode !== mode)) {
return {
error: {
code: 'subagent-not-found',
message: `session "${childSessionId}" is not a ${mode} direct child of "${parentSessionId}"`,
details: { parentSessionId, childSessionId },
},
}
}
if (entry.kind === 'diagnostic') {
return {
error: {
code: 'subagent-catalog-diagnostic',
message: `subagent "${childSessionId}" is ${entry.reason}`,
details: { parentSessionId, childSessionId, reason: entry.reason },
},
}
}
return { entry }
} catch (error: unknown) {
if (signal?.aborted
|| (error instanceof SubagentError && error.code === 'CANCELLED')
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
return { error: { code: 'cancelled', message: 'subagent catalog read was cancelled', details: {} } }
}
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return {
error: {
code: 'subagent-not-found',
message: `parent session "${parentSessionId}" was not found`,
details: { parentSessionId, childSessionId },
},
}
}
return { error: { code: 'internal', message: 'subagent catalog read failed', details: {} } }
}
}
/**
* Thrown by the cold-resume path when the id names no servable session
* (absent from the store, or a pre-project legacy log without a cwd).
@@ -1698,20 +1713,34 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
try {
const entries = await ctx.subagents.listChildren(request.payload.parentSessionId, signal)
return ok(request, {
entries: continuableCatalog(entries),
entries,
parentAvailable: ctx.agents.get(request.payload.parentSessionId) !== undefined,
})
} catch (error: unknown) {
if (error instanceof SubagentError && error.code === 'CANCELLED') {
return err(request, { code: 'cancelled', message: 'subagent catalog read was cancelled', details: {} })
if (signal?.aborted
|| (error instanceof SubagentError && error.code === 'CANCELLED')
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
return err(request, {
code: 'cancelled',
message: 'subagent catalog read was cancelled',
details: {},
})
}
return err(request, { code: 'internal', message: 'subagent catalog read failed', details: {} })
return err(request, {
code: 'internal',
message: 'subagent catalog read failed',
details: {},
})
}
},
async history(request, signal) {
const { parentSessionId, childSessionId, beforeSeq, maxMessages } = request.payload
const verified = await healthyCatalogChild(ctx, parentSessionId, childSessionId, signal)
const {
parentSessionId, childSessionId, mode, beforeSeq, maxMessages,
} = request.payload
const verified = await catalogChild(ctx, {
parentSessionId, childSessionId, mode,
}, signal)
if (verified.error !== undefined) return err(request, verified.error)
try {
const snapshot = await ctx.sessionQuery.readSession(childSessionId)
@@ -1723,24 +1752,33 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
details: { childSessionId },
})
}
return ok(request, historyPage(ctx, snapshot.events, beforeSeq, maxMessages))
const page = historyPage(ctx, snapshot.events, beforeSeq, maxMessages)
const projections = beforeSeq === undefined
? detachedProjectionsFor(ctx, snapshot.events)
: undefined
return ok(request, { ...page, ...projections === undefined ? {} : { projections } })
} catch (error: unknown) {
if (signal?.aborted) {
return err(request, { code: 'cancelled', message: 'subagent history read was cancelled', details: {} })
if (signal?.aborted
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
return err(request, {
code: 'cancelled',
message: 'subagent history read was cancelled',
details: {},
})
}
if (error instanceof SessionQueryError) {
if (error.code === 'SESSION_QUERY_ABORTED') {
return err(request, { code: 'cancelled', message: 'subagent history read was cancelled', details: {} })
}
if (error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return err(request, {
code: 'subagent-not-found',
message: 'subagent disappeared during history read',
details: { parentSessionId, childSessionId },
})
}
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return err(request, {
code: 'subagent-not-found',
message: 'subagent disappeared during history read',
details: { parentSessionId, childSessionId },
})
}
return err(request, { code: 'internal', message: 'subagent history read failed', details: {} })
return err(request, {
code: 'internal',
message: 'subagent history read failed',
details: {},
})
}
},
@@ -1754,19 +1792,18 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
details: { parentSessionId },
})
}
const verified = await healthyCatalogChild(ctx, parentSessionId, childSessionId, signal)
const verified = await catalogChild(ctx, {
parentSessionId, childSessionId, mode: 'continuable',
}, signal)
if (verified.error !== undefined) return err(request, verified.error)
const operationSignal = signal ?? new AbortController().signal
try {
const messageId = await ctx.subagents.followup(
parent,
childSessionId,
content,
{ source: { kind: 'user', rpcId: request.rpcId }, signal: operationSignal },
)
const messageId = await ctx.subagents.followup(parent, childSessionId, content, {
source: { kind: 'user', rpcId: request.rpcId },
signal,
})
return ok(request, { messageId })
} catch (error: unknown) {
return subagentPromptError(request, error, operationSignal)
return subagentPromptError(request, error, signal)
}
},
},

2
packages/host/apiproxy/src/api/rpc.schema.ts
View File

@@ -65,7 +65,7 @@ export const rpcErrorSchema: z.ZodType<RpcError> = z.discriminatedUnion('code',
}) }),
z.object({ code: z.literal('subagent-not-resumable'), message: z.string(), details: z.object({ childSessionId: z.string() }) }),
z.object({ code: z.literal('subagent-unauthorized'), message: z.string(), details: z.object({ childSessionId: z.string() }) }),
z.object({ code: z.literal('subagent-not-delivered'), message: z.string(), details: z.object({ childSessionId: z.string() }) }),
z.object({ code: z.literal('subagent-delivery-unavailable'), message: z.string(), details: z.object({ childSessionId: z.string() }) }),
z.object({ code: z.literal('internal'), message: z.string(), details: z.object({}) }),
]) as unknown as z.ZodType<RpcError>

2
packages/host/apiproxy/src/api/rpc.ts
View File

@@ -80,7 +80,7 @@ export interface RpcErrorDetailsMap {
}
'subagent-not-resumable': { childSessionId: SessionId }
'subagent-unauthorized': { childSessionId: SessionId }
'subagent-not-delivered': { childSessionId: SessionId }
'subagent-delivery-unavailable': { childSessionId: SessionId }
'internal': {}
}

21
packages/host/apiproxy/src/api/subagents.schema.ts
View File

@@ -1,19 +1,29 @@
/** Zod schemas for the browser-safe subagent domain. */
import { z } from 'zod'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { RequestPayload, ResponseValue } from './rpc-map.ts'
import type { Wire } from './rpc.schema.ts'
import { contentBlockSchema, historyEntrySchema, sessionIdSchema } from './sessions.schema.ts'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import {
contentBlockSchema, historyEntrySchema, sessionIdSchema, sessionProjectionsBlockSchema,
} from './sessions.schema.ts'
import type { SubagentListEntry } from './subagents.ts'
/** Healthy and diagnostic durable catalog rows. */
export const subagentListEntrySchema = z.discriminatedUnion('kind', [
export const subagentListEntrySchema = z.union([
z.object({
kind: z.literal('child'),
id: sessionIdSchema,
label: z.string(),
mode: z.literal('one-shot'),
activity: z.union([z.literal('running'), z.literal('inactive')]),
label: z.string().optional(),
}),
z.object({
kind: z.literal('child'),
id: sessionIdSchema,
mode: z.literal('continuable'),
activity: z.union([z.literal('running'), z.literal('inactive')]),
label: z.string(),
}),
z.object({
kind: z.literal('diagnostic'),
@@ -37,6 +47,7 @@ export const subagentListValueSchema = z.object({
export const subagentHistoryRequestSchema = z.object({
parentSessionId: sessionIdSchema,
childSessionId: sessionIdSchema,
mode: z.union([z.literal('one-shot'), z.literal('continuable')]),
beforeSeq: z.number().int().nonnegative().optional(),
maxMessages: z.number().int().positive().optional(),
}) satisfies z.ZodType<Wire<RequestPayload<'subagent.history'>>>
@@ -45,12 +56,14 @@ export const subagentHistoryRequestSchema = z.object({
export const subagentHistoryValueSchema = z.object({
events: z.array(historyEntrySchema),
hasMore: z.boolean(),
projections: sessionProjectionsBlockSchema.optional(),
}) as unknown as z.ZodType<Wire<ResponseValue<'subagent.history'>>>
/** subagent.prompt request payload. */
export const subagentPromptRequestSchema = z.object({
parentSessionId: sessionIdSchema,
childSessionId: sessionIdSchema,
mode: z.literal('continuable'),
content: z.array(contentBlockSchema),
}) as unknown as z.ZodType<RequestPayload<'subagent.prompt'>>

53
packages/host/apiproxy/src/api/subagents.ts
View File

@@ -1,23 +1,31 @@
/**
* Browser-safe subagent domain contract. Persisted transcript reads never
* activate an Agent, while prompts route through the direct parent's
* Activation-backed continuation owner.
* activate an Agent, while continuable prompts route through the exact live
* direct parent into the child's Agent inbox.
*/
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import type { RpcRequest, RpcResponse } from './rpc.ts'
import type { HistoryEntry } from './sessions.ts'
import type { HistoryEntry, SessionProjectionsBlock } from './sessions.ts'
/** Complete durable direct-child catalog row. */
export type SubagentListEntry =
| {
kind: 'child'
id: SessionId
label: string
activity: 'running' | 'inactive'
}
} & (
| {
mode: 'one-shot'
label?: string
}
| {
mode: 'continuable'
label: string
}
)
| {
kind: 'diagnostic'
id: SessionId
@@ -30,10 +38,15 @@ export interface SubagentPromptReceipt {
}
/** Durable parent/child address that selects subagent transport in the client. */
export interface SubagentAddress {
parentSessionId: SessionId
childSessionId: SessionId
}
export type SubagentAddress =
& {
parentSessionId: SessionId
childSessionId: SessionId
}
& (
| { mode: 'one-shot' }
| { mode: 'continuable' }
)
/** Complete direct-child catalog plus the delivery-time parent availability hint. */
export interface SubagentCatalog {
@@ -44,8 +57,9 @@ export interface SubagentCatalog {
/** Subagent-domain unary methods. */
export interface SubagentsApi {
/**
* Lists direct continuable children without loading either side. Parent
* availability is a hint; prompt performs the authoritative check.
* Lists direct session-backed children without loading either side. Parent
* availability is a hint; continuable prompt performs the authoritative
* check.
*/
list(
request: RpcRequest<{ parentSessionId: SessionId }>,
@@ -59,14 +73,21 @@ export interface SubagentsApi {
history(
request: RpcRequest<SubagentAddress & { beforeSeq?: number; maxMessages?: number }>,
signal?: AbortSignal,
): Promise<RpcResponse<{ events: HistoryEntry[]; hasMore: boolean }>>
): Promise<RpcResponse<{
events: HistoryEntry[]
hasMore: boolean
projections?: SessionProjectionsBlock
}>>
/**
* Delivers human content through the exact live parent's continuation
* owner. Success identifies the accepted inbox message.
* Delivers human content to a continuable child through the exact live
* parent's continuation owner. Success identifies the message accepted by
* the child's FIFO inbox; later execution is independent of this request.
*/
prompt(
request: RpcRequest<SubagentAddress & { content: ContentBlock[] }>,
signal?: AbortSignal,
request: RpcRequest<
Extract<SubagentAddress, { mode: 'continuable' }> & { content: ContentBlock[] }
>,
signal: AbortSignal,
): Promise<RpcResponse<SubagentPromptReceipt>>
}