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

Merge branch 'stack/agent-profiles-3-wire' into stack/agent-profiles-4-settings

Browse Source
This commit is contained in:
Yichen Jiang
2026-08-07 13:39:40 +08:00
parent a0194bdab2 f3f2129154
commit e95e413115
259 changed files with 8059 additions and 1533 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -538,6 +538,28 @@ function detachedProjectionsFor(
return registry.restore({}, events, 0).snapshot
}
/**
* Best-effort projections for one subagent history page, fail-soft like
* {@link listProjectionsFor}: a registered unit throwing on a corrupt payload
* never blocks transcript reading — the page is served without the block.
* @param ctx - context carrying the logger for the degradation warning.
* @param childSessionId - the child whose page is being decorated.
* @param compute - the arm-specific fold (live watermark or detached restore).
* @returns the projections block, or undefined when the fold failed.
*/
function subagentHistoryProjections(
ctx: Context,
childSessionId: SessionId,
compute: () => SessionProjectionsBlock | undefined,
): SessionProjectionsBlock | undefined {
try {
return compute()
} catch (error) {
ctx.logger.warn(`subagent.history: projections for "${childSessionId}" failed (serving the page without them): ${String(error)}`)
return undefined
}
}
/** Map continuation admission failures without exposing provider details. */
function subagentPromptError(
request: RpcRequest<{ childSessionId: SessionId }>,
@@ -578,6 +600,15 @@ function subagentPromptError(
return err(request, { code: 'internal', message: 'subagent prompt failed', details: {} })
}
/** Stable RPC face of the missing projections capability, shared by every catalog read path. */
function projectionsUnavailableError(): RpcError {
return {
code: 'internal',
message: 'subagent catalog is unavailable: this deployment does not mount the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
details: {},
}
}
/** Verify one address and mode against the complete direct-child catalog. */
async function catalogChild(
ctx: Context,
@@ -611,19 +642,11 @@ async function catalogChild(
}
return { entry }
} catch (error: unknown) {
if (signal?.aborted
|| (error instanceof SubagentError && error.code === 'CANCELLED')
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
if (signal?.aborted || (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}" was not found`,
details: { parentSessionId, childSessionId },
},
}
if (error instanceof SubagentError && error.code === 'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE') {
return { error: projectionsUnavailableError() }
}
return { error: { code: 'internal', message: 'subagent catalog read failed', details: {} } }
}
@@ -1018,28 +1041,16 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
})
}
/** Whether the session's own suffix carries the durable subagent discriminator. */
function hasSubagentDescriptor(session: Pick<Session, 'events' | 'header'>): boolean {
const events = session.events
// Indexed scan from the own-suffix start: slicing copies the whole suffix
// on every Agent-bound RPC, including each `session.prompt` on long
// transcripts.
for (let index = session.header.seedLength ?? 0; index < events.length; index += 1) {
if (events[index]?.type === 'subagent/descriptor') return true
}
return false
}
/**
* Generic Host interaction cannot claim a durably classified subagent or an
* Agent created through its live parent. The runtime-owner arm also covers
* descriptor-less child publication windows and older stored headers.
* Generic Host interaction cannot claim a durably classified subagent
* (`origin: 'subagent'` in the header) or an Agent runtime-owned by its
* live parent.
*/
function hasSubagentOwner(
session: Pick<Session, 'events' | 'header'>,
session: Pick<Session, 'header'>,
agent: Agent | undefined,
): boolean {
if (session.header.origin === 'subagent' || hasSubagentDescriptor(session)) return true
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)
@@ -1095,7 +1106,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
resume = (async () => {
try {
const inspected = await inspectServable(sessionId)
if (hasSubagentOwner({ header: inspected.meta, events: inspected.events }, undefined)) {
if (hasSubagentOwner({ header: inspected.meta }, undefined)) {
throw new SubagentSessionOwnership(sessionId)
}
const publishedSession = ctx.sessions.get(sessionId)
@@ -1224,7 +1235,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
// Ownership first: explicit-id adoption of a session-backed
// subagent must answer `agent-busy` regardless of the requested
// cwd (the api/commands.ts contract), not a cwd conflict.
if (hasSubagentOwner({ header: inspected.meta, events: inspected.events }, undefined)) {
if (hasSubagentOwner({ header: inspected.meta }, undefined)) {
throw new SubagentSessionOwnership(sessionId)
}
if (inspected.meta.cwd !== cwd) {
@@ -2067,15 +2078,16 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
parentAvailable: ctx.agents.get(request.payload.parentSessionId) !== undefined,
})
} catch (error: unknown) {
if (signal?.aborted
|| (error instanceof SubagentError && error.code === 'CANCELLED')
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
if (signal?.aborted || (error instanceof SubagentError && error.code === 'CANCELLED')) {
return err(request, {
code: 'cancelled',
message: 'subagent catalog read was cancelled',
details: {},
})
}
if (error instanceof SubagentError && error.code === 'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE') {
return err(request, projectionsUnavailableError())
}
return err(request, {
code: 'internal',
message: 'subagent catalog read failed',
@@ -2092,44 +2104,65 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
parentSessionId, childSessionId, mode,
}, signal)
if (verified.error !== undefined) return err(request, verified.error)
try {
const snapshot = await ctx.sessionQuery.readSession(childSessionId)
signal?.throwIfAborted()
if (snapshot.session.parentSession !== parentSessionId) {
return err(request, {
code: 'subagent-unauthorized',
message: 'subagent parent changed during history read',
details: { childSessionId },
})
}
const page = historyPage(ctx, snapshot.events, beforeSeq, maxMessages, ctx.agents.get(childSessionId))
const projections = beforeSeq === undefined
? detachedProjectionsFor(ctx, snapshot.events)
// The generic-history data plane: an attached child serves its
// in-memory snapshot and the registry's live watermark projections; a
// cold child is one persistence inspection plus a detached fold.
let header: SessionHeader
let events: SessionEvent[]
let projections: SessionProjectionsBlock | undefined
const attached = ctx.sessions.get(childSessionId)
if (attached !== undefined) {
header = attached.header
events = [...attached.events]
projections = beforeSeq === undefined
? subagentHistoryProjections(ctx, childSessionId, () => projectionsFor(ctx, attached))
: undefined
return ok(request, { ...page, ...projections === undefined ? {} : { projections } })
} catch (error: unknown) {
if (signal?.aborted
|| (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
} else {
try {
const inspected = await inspectServable(childSessionId)
header = inspected.meta
events = inspected.events
projections = beforeSeq === undefined
? subagentHistoryProjections(ctx, childSessionId, () => detachedProjectionsFor(ctx, inspected.events))
: undefined
} catch (error: unknown) {
if (signal?.aborted) {
return err(request, {
code: 'cancelled',
message: 'subagent history read was cancelled',
details: {},
})
}
if (error instanceof SessionNotFound) {
return err(request, {
code: 'subagent-not-found',
message: 'subagent disappeared during history read',
details: { parentSessionId, childSessionId },
})
}
return err(request, {
code: 'cancelled',
message: 'subagent history read was cancelled',
code: 'internal',
message: 'subagent history read failed',
details: {},
})
}
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 },
})
}
}
if (signal?.aborted) {
return err(request, {
code: 'internal',
message: 'subagent history read failed',
code: 'cancelled',
message: 'subagent history read was cancelled',
details: {},
})
}
if (header.parentSession !== parentSessionId) {
return err(request, {
code: 'subagent-unauthorized',
message: 'subagent parent changed during history read',
details: { childSessionId },
})
}
const page = historyPage(ctx, events, beforeSeq, maxMessages)
return ok(request, { ...page, ...projections === undefined ? {} : { projections } })
},
async prompt(request, signal) {

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

@@ -70,7 +70,8 @@ export interface SubagentsApi {
): Promise<RpcResponse<SubagentCatalog>>
/**
* Reads one healthy catalog child's persisted raw log with ordinary
* Reads one healthy catalog child's transcript — the in-memory snapshot of
* a live child, the persisted log of a cold one — with ordinary
* message-aligned pagination and render intents, without Agent activation.
*/
history(

42
packages/host/apiproxy/tests/api-proxy-cold.spec.ts
View File

@@ -190,6 +190,7 @@ describe('subagent ownership fence', () => {
const meta = header('session-child', 1000, {
parentSession: sid('session-parent'),
seedLength: 0,
origin: 'subagent',
})
const events = [
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
@@ -245,6 +246,47 @@ describe('subagent ownership fence', () => {
expect(inspect).toHaveBeenCalledTimes(3)
})
it('no longer treats a descriptor-only cold child without origin as subagent-owned', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserInteractionService)
const sessionId = sid('session-legacy-child')
const meta = header('session-legacy-child', 1000, {
parentSession: sid('session-parent'),
seedLength: 0,
})
const events = [
{
type: 'subagent/descriptor',
seq: 0,
time: 1,
data: { version: 2, mode: 'continuable', provider: 'spawn', label: 'child' },
},
] as SessionEvent[]
ctx.provide('sessionPersistence', {
list: () => Promise.resolve([meta]),
inspect: () => Promise.resolve({ meta, events }),
locate: () => undefined,
} as never)
// Pre-#1569 stores classify a child only through the descriptor event and
// carry no header `origin`; the pre-release decision stops recognizing
// them, so the ownership fence lets generic resume reach the registry
// instead of answering `agent-busy`.
const resume = vi.spyOn(ctx.agents, 'resume')
.mockRejectedValue(new Error('registry unavailable in this bench'))
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const prompt = await api.sessions.prompt(request({
sessionId,
mode: 'queue',
content: [{ type: 'text', text: 'follow up' }],
}))
expect(resume).toHaveBeenCalledTimes(1)
expect(prompt.result.ok).toBe(false)
if (!prompt.result.ok) expect(prompt.result.error.code).toBe('internal')
})
it('rejects origin-marked and runtime-owned live children from generic controls', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)

141
packages/host/apiproxy/tests/api-proxy-subagents.spec.ts
View File

@@ -1,7 +1,6 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
import { SubagentError } from '@deepseek-ai/dsh-subagent'
import { RpcId } from '../src/api/rpc.ts'
import type { RpcRequest } from '../src/api/rpc.ts'
@@ -21,7 +20,12 @@ function bench(options: {
entries?: object[]
followupError?: Error
listError?: Error
readError?: Error
/** Persistence forgets the child entirely (the vanished-mid-read race). */
storedChild?: false
/** Attach the child to the live session store instead of persistence only. */
liveChild?: true
/** Every registered projection unit throws on this child's payloads. */
projectionsThrow?: true
historyParent?: SessionId
} = {}) {
const parent = { id: PARENT }
@@ -49,25 +53,44 @@ function bench(options: {
) => options.followupError === undefined
? Promise.resolve('message-1')
: Promise.reject(options.followupError))
const readSession = vi.fn(() => options.readError === undefined
? Promise.resolve({
session: {
version: 0, id: CHILD, createdAt: 1, parentSession: options.historyParent ?? PARENT,
} satisfies SessionHeader,
events: [
{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'work' }], source: { kind: 'user' } } },
] as unknown as SessionEvent[],
})
: Promise.reject(options.readError))
const childHeader = {
version: 0, id: CHILD, createdAt: 1, cwd: '/proj', parentSession: options.historyParent ?? PARENT,
} satisfies SessionHeader
const childEvents = [
{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'work' }], source: { kind: 'user' } } },
] as unknown as SessionEvent[]
const inspect = vi.fn(() => Promise.resolve({ meta: childHeader, events: childEvents }))
const liveBlock = { values: {}, asOfSeq: 3 }
const coldBlock = { values: {}, asOfSeq: 0 }
const snapshot = vi.fn(() => {
if (options.projectionsThrow === true) throw new Error('hostile unit')
return liveBlock
})
const restore = vi.fn(() => {
if (options.projectionsThrow === true) throw new Error('hostile unit')
return { snapshot: coldBlock }
})
const ctx = new Context()
ctx.provide('agents', { get: getAgent })
ctx.provide('subagents', { listChildren, followup })
ctx.provide('sessionQuery', { readSession })
ctx.provide('sessions', {
get: (id: SessionId) => options.liveChild === true && id === CHILD
? { id: CHILD, header: childHeader, events: childEvents }
: undefined,
})
ctx.provide('sessionPersistence', {
list: () => Promise.resolve(options.storedChild === false ? [] : [childHeader]),
inspect,
locate: () => undefined,
})
// The gateway's own projection push feed subscribes at construction; the
// no-op disposer keeps that seam quiet while these tests pin history reads.
ctx.provide('sessionProjections', { snapshot, restore, onChanged: () => () => {} })
ctx.provide('userInteraction', { registerProvider: () => () => {} })
const api = createApiProxy(ctx, {
provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp',
})
return { api, getAgent, listChildren, readSession, followup, parent }
return { api, getAgent, listChildren, inspect, snapshot, restore, followup, parent }
}
describe('subagent gateway', () => {
@@ -112,13 +135,8 @@ describe('subagent gateway', () => {
.toMatchObject({ ok: true, value: { entries: [{ activity: 'running' }] } })
})
it('reads a healthy direct child without acquiring an Agent owner', async () => {
// `bench()` leaves the child with no live Agent at all, so the response
// below is produced cold — which is the invariant: the read never creates
// or resumes one. It may still CONSULT the live registry, because tool
// presenters live with the per-agent definitions and rendering this
// child's own cards needs its layer.
const { api, getAgent, readSession } = bench()
it('reads a healthy direct child without looking up or activating any Agent', async () => {
const { api, getAgent, inspect, restore } = bench()
const response = await api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', maxMessages: 10,
}))
@@ -126,8 +144,46 @@ describe('subagent gateway', () => {
ok: true,
value: { hasMore: false, events: [{ event: { type: 'user/message', seq: 0 } }] },
})
expect(readSession).toHaveBeenCalledWith(CHILD)
expect(getAgent).not.toHaveBeenCalledWith(PARENT)
expect(inspect).toHaveBeenCalledWith(CHILD)
expect(restore).toHaveBeenCalledTimes(1)
expect(getAgent).not.toHaveBeenCalled()
})
it('serves a live child from the in-memory snapshot and the watermark projections', async () => {
const { api, inspect, snapshot, restore } = bench({ liveChild: true })
const response = await api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))
expect(response.result).toMatchObject({
ok: true,
value: { hasMore: false, projections: { asOfSeq: 3 } },
})
expect(snapshot).toHaveBeenCalledTimes(1)
expect(restore).not.toHaveBeenCalled()
expect(inspect).not.toHaveBeenCalled()
})
it('serves the page without projections when a hostile unit breaks the fold', async () => {
const cold = bench({ projectionsThrow: true })
const coldResponse = await cold.api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))
expect(coldResponse.result).toMatchObject({
ok: true,
value: { hasMore: false, events: [{ event: { type: 'user/message', seq: 0 } }] },
})
if (coldResponse.result.ok) expect('projections' in coldResponse.result.value).toBe(false)
const live = bench({ projectionsThrow: true, liveChild: true })
const liveResponse = await live.api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))
expect(liveResponse.result).toMatchObject({
ok: true,
value: { hasMore: false, events: [{ event: { type: 'user/message', seq: 0 } }] },
})
if (liveResponse.result.ok) expect('projections' in liveResponse.result.value).toBe(false)
expect(live.snapshot).toHaveBeenCalledTimes(1)
})
it('reads one-shot history and rejects an address with the wrong mode', async () => {
@@ -135,18 +191,18 @@ describe('subagent gateway', () => {
kind: 'child', id: CHILD, mode: 'one-shot', label: 'batch',
activity: 'inactive', hasChildren: false,
}
const { api, readSession } = bench({ entries: [oneShot] })
const { api, inspect } = bench({ entries: [oneShot] })
expect((await api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'one-shot',
}))).result).toMatchObject({ ok: true })
expect((await api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))).result).toMatchObject({ ok: false, error: { code: 'subagent-not-found' } })
expect(readSession).toHaveBeenCalledTimes(1)
expect(inspect).toHaveBeenCalledTimes(1)
})
it('rejects a diagnostic address before reading history', async () => {
const { api, readSession } = bench({ entries: [
const { api, inspect } = bench({ entries: [
{ kind: 'diagnostic', id: CHILD, reason: 'unsupported' },
] })
const response = await api.subagents.history(request({
@@ -159,7 +215,34 @@ describe('subagent gateway', () => {
details: { parentSessionId: PARENT, childSessionId: CHILD, reason: 'unsupported' },
},
})
expect(readSession).not.toHaveBeenCalled()
expect(inspect).not.toHaveBeenCalled()
})
it('maps the missing projections capability to one wire face on list, history, and prompt', async () => {
const listError = () => new SubagentError(
'listing subagents requires the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE',
)
const expected = {
code: 'internal',
message: 'subagent catalog is unavailable: this deployment does not mount the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
}
const list = bench({ listError: listError() })
expect((await list.api.subagents.list(request({ parentSessionId: PARENT }))).result)
.toMatchObject({ ok: false, error: expected })
const history = bench({ listError: listError() })
expect((await history.api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))).result).toMatchObject({ ok: false, error: expected })
expect(history.inspect).not.toHaveBeenCalled()
const prompt = bench({ listError: listError() })
expect((await prompt.api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content: [],
}), new AbortController().signal)).result).toMatchObject({ ok: false, error: expected })
expect(prompt.followup).not.toHaveBeenCalled()
})
it('routes human content through the exact live parent with rpc attribution', async () => {
@@ -198,9 +281,7 @@ describe('subagent gateway', () => {
})
it('maps history disappearance and hides unexpected backend details', async () => {
const disappeared = bench({
readError: new SessionQueryError('secret path', 'SESSION_QUERY_SESSION_NOT_FOUND'),
})
const disappeared = bench({ storedChild: false })
expect((await disappeared.api.subagents.history(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable',
}))).result).toMatchObject({