refactor(subagent): consume shared session observations

This commit is contained in:
imccyu
2026-08-25 06:09:25 +08:00
parent b8dfa8b892
commit f5f0448bee
48 changed files with 572 additions and 302 deletions
+1
View File
@@ -124,6 +124,7 @@
"@deepseek-ai/dsh-session-checkpoint-policy": "workspace:^",
"@deepseek-ai/dsh-session-log-deepseek": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-settings": "workspace:^",
"@deepseek-ai/dsh-settings-file": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
+10 -15
View File
@@ -28,8 +28,8 @@ interface SessionList {
items: Array<{
sessionId: string
cwd?: string
agentPreset?: string
blank: boolean
projections?: { values: { agentPreset?: string | null } }
}>
}
@@ -222,23 +222,18 @@ async function workspaceBaseline(baseUrl: string): Promise<WorkspaceBaseline> {
return frame.value as WorkspaceBaseline
}
/** Read the explicit page cut from a fresh Session follow generation. */
async function sessionCursor(baseUrl: string, sessionId: string): Promise<number> {
/** Read the complete opening page from a fresh Session follow generation. */
async function history(baseUrl: string, sessionId: string): Promise<HistoryPage> {
const frame = await openingStreamItem(
baseUrl,
'session/follow',
{ request: { address: { kind: 'session', sessionId } } },
value => isRecord(value) && value.type === 'opened' && Number.isSafeInteger(value.cursor),
{ request: { address: { kind: 'session', sessionId }, maxMessages: 100 } },
value => isRecord(value)
&& value.type === 'snapshot'
&& Array.isArray(value.events)
&& typeof value.hasMore === 'boolean',
)
return frame.cursor as number
}
/** Read Session history at the cursor explicitly opened for this page. */
async function history(baseUrl: string, sessionId: string): Promise<HistoryPage> {
const throughSeq = await sessionCursor(baseUrl, sessionId)
return remoteRpc<HistoryPage>(baseUrl, 'session/page', {
request: { address: { kind: 'session', sessionId }, throughSeq, maxMessages: 100 },
})
return { events: frame.events as HistoryPage['events'], hasMore: frame.hasMore as boolean }
}
/** Poll a public observation until it satisfies the test's behavior predicate. */
@@ -379,9 +374,9 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('GitHub webhook through the real
const sessions = await remoteRpc<SessionList>(baseUrl, 'session/list', { _request: {} })
expect(sessions.items.find(session => session.sessionId === sessionId)).toMatchObject({
agentPreset: 'minimal',
blank: false,
cwd: canonicalWorkspacePath,
projections: { values: { agentPreset: 'minimal' } },
})
const admitted = await eventually(
@@ -10,6 +10,9 @@
root: './.sessions'
compression: none
- id: session-query
name: './tests/fixtures/subagent-diagnostic-query.ts'
# file/override both default to their DSH_SNAPSHOT_* env vars.
- id: replay
name: '@deepseek-ai/dsh-llm-replay'
@@ -0,0 +1,14 @@
/** Exact-read Session query used by the descriptor-less child snapshot. */
import SessionQueryEngine from '@deepseek-ai/dsh-session-query'
/** Search is outside this fixture; inherited corpus and observation reads stay real. */
export default class SubagentDiagnosticQuery extends SessionQueryEngine {
override searchSessions(): Promise<never> {
return Promise.reject(new Error('session search is unavailable in this fixture'))
}
override searchEvents(): Promise<never> {
return Promise.reject(new Error('event search is unavailable in this fixture'))
}
}
+3 -18
View File
@@ -12,7 +12,7 @@ import type { PatchOptions } from '@deepseek-ai/cordis-plugin-include'
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
import { settingsNamespace } from '@deepseek-ai/dsh-settings'
import { SUBAGENT_MODEL_SELECTION_SETTINGS_NAMESPACE } from '@deepseek-ai/dsh-tool-subagent/model-selection-settings'
import { resolveSessionPreset, SETTINGS_NAMESPACE, SHIPPED_PRESET_ROOT } from '@deepseek-ai/dsh-agent-presets'
import { SETTINGS_NAMESPACE, SHIPPED_PRESET_ROOT } from '@deepseek-ai/dsh-agent-presets'
import { applyChildComposition, childSessionMeta } from '@deepseek-ai/dsh-subagent'
import { CallId } from '@deepseek-ai/dsh-llm'
import type {} from '@deepseek-ai/dsh-compaction-basic'
@@ -629,27 +629,12 @@ describe('a switch survives the session', () => {
// The header keeps the creation fact; the log carries what it runs.
expect(handle.agent.session.header.agentPreset).toBe('standard')
expect(resolveSessionPreset(handle.agent.session)).toBe('minimal')
expect(ctx.sessionProjections.stateOf(handle.agent.session, 'agentPreset')).toBe('minimal')
} finally {
await handle.dispose()
}
})
it('rebuilds a switched session from the log, not the creation header', () => {
// The exact shape a resume reads back from disk: the header says standard,
// the log records the switch the user made while the session was blank.
const rebuilt = resolveSessionPreset({
header: { version: 0, id: SessionId('x'), createdAt: 0, agentPreset: 'standard' },
events: [
{ type: 'agent-preset/selected', seq: 1, time: 0, data: { agentPreset: 'minimal' } },
{ type: 'turn/start', seq: 2, time: 0, data: { turn: 0, trigger: { kind: 'message', source: { kind: 'user' } } } },
] as never,
})
// Reading the header alone would compose the creation-time preset over a
// history another one produced — the replay the blank-only lock prevents.
expect(rebuilt).toBe('minimal')
})
})
describe('a forked session', () => {
@@ -659,7 +644,7 @@ describe('a forked session', () => {
meta: { agentPreset: 'minimal' },
setup: agentCtx => ctx.agentPresets.mount(agentCtx, 'minimal').then(() => undefined),
})
const inherited = resolveSessionPreset(parent.agent.session)
const inherited = ctx.sessionProjections.stateOf(parent.agent.session, 'agentPreset') ?? undefined
const child = await ctx.agents.create({
sessionId: SessionId('preset-fork-child'),
meta: {
+3
View File
@@ -65,6 +65,9 @@
{
"path": "../../packages/session-query/session-query-sqlite"
},
{
"path": "../../packages/session-query/session-query"
},
{
"path": "../../packages/shell/shell-env"
},
+12
View File
@@ -69,6 +69,8 @@ describe('web e2e: Cordis tools use their owned cards', () => {
let page: Page
let tripwire: ReturnType<typeof watchConsole>
const sessionEvents: SessionEvent[] = []
const modelFrames: string[] = []
const modelChanges: string[] = []
beforeAll(async () => {
scaffold = await launchWebScaffold({
@@ -77,8 +79,17 @@ describe('web e2e: Cordis tools use their owned cards', () => {
...(MODE === 'record' ? {} : { replayFixture: FIXTURE, paceMs: 15 }),
})
scaffold.ctx.on('session/event', (_session, event: SessionEvent) => { sessionEvents.push(event) })
scaffold.ctx.sessionProjections.onChanged((_session, key, value, seq) => {
if (key === 'modelSelection') modelChanges.push(`${String(seq)}:${JSON.stringify(value)}`)
})
browser = await chromium.launch()
page = await newEnglishPage(browser)
page.on('websocket', (socket) => {
socket.on('framereceived', (frame) => {
const payload = String(frame.payload)
if (payload.includes('modelSelection')) modelFrames.push(payload)
})
})
tripwire = watchConsole(page)
await page.goto(scaffold.baseUrl, { waitUntil: 'load' })
await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
@@ -169,6 +180,7 @@ describe('web e2e: Cordis tools use their owned cards', () => {
it.skipIf(MODE === 'record')('matches the conversation aria golden', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-cordis-aria'))
console.log('MODEL_TRACE', { modelChanges, frameCount: modelFrames.length, modelFrames })
await page.locator('[data-conversation-scroll]').evaluate((host) => { host.scrollTop = host.scrollHeight })
await expect.poll(
async () => page.getByRole('button', { name: 'Back to bottom', exact: true }).count(),
@@ -42,8 +42,8 @@
- button "Commands":
- img
- 'button "Access mode, current: Read Only"': Read Only
- button "Select model":
- text: Select model
- button "Select model, current github-webhook-review-test/reply":
- text: github-webhook-review-test/reply
- img
- button "Send message" [disabled]
- text: 1 turns · 1 steps LLM {{duration}}
+1 -1
View File
@@ -233,7 +233,7 @@ describe.skipIf(MODE === 'record')('web e2e: file and session references through
const group = page.getByRole('treeitem', { name: /Ungrouped/ })
await group.waitFor({ timeout: 15_000 })
if (await group.getAttribute('aria-expanded') !== 'true') await group.click()
const target = page.getByRole('treeitem').filter({ hasText: /^dsh-web-e2e-ws-/ }).first()
const target = page.getByRole('treeitem', { name: /Reference order target/ })
await target.waitFor({ timeout: 15_000 })
await target.click()
await page.getByRole('button', { name: /^Session recall\s*Research notes$/ }).waitFor({ timeout: 15_000 })
+16 -25
View File
@@ -228,44 +228,35 @@ describe('web e2e: seeded history renders through cold resume', () => {
await recordFixture(scaffold, sessionId, SEED)
}, 200_000)
it.skipIf(MODE === 'record')('serves the projections baseline on the real composition tail page', async () => {
it.skipIf(MODE === 'record')('serves the projections baseline on the real composition opening snapshot', async () => {
// Composition regression tripwire: the projection registry must be a row
// in the SHIPPED cordis.yml — with it absent every domain unit's optional
// injection stays silent and this block disappears (no titles/todos on
// the web), while fixture-level suites stay green. Assert through the
// real HTTP wire against the booted real host.
const response = await fetch(`${scaffold.baseUrl}/api/session/page`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({
type: 'client-request', rpcId: 'seeded-projections', method: 'session/page',
payload: {
args: { request: {
address: { kind: 'session', sessionId: SEED_ID },
throughSeq: seededThroughSeq,
} },
},
}),
})
expect(response.ok).toBe(true)
const body = await response.json() as {
result: { ok: boolean; value?: { projections?: { asOfSeq: number; values: Record<string, unknown> } } }
// production Session Controller against the booted real host.
const controller = new AbortController()
const stream = scaffold.ctx.sessionController.follow({
address: { kind: 'session', sessionId: SessionId(SEED_ID) },
}, controller.signal)[Symbol.asyncIterator]()
const first = await stream.next()
controller.abort()
if (first.done || first.value.type !== 'snapshot') {
throw new Error('session follow did not publish its opening snapshot')
}
expect(body.result.ok).toBe(true)
const projections = body.result.value?.projections
expect(projections).toBeDefined()
expect(projections?.asOfSeq).toBeGreaterThanOrEqual(0)
expect(first.value.cursor).toBe(seededThroughSeq)
const projections = first.value.projections
expect(projections.asOfSeq).toBe(seededThroughSeq)
// The seed carries a session/title event: the title unit is host-plane, so
// it folds the detached log and serves the value with nothing composed.
expect(typeof projections?.values.title).toBe('string')
expect(typeof projections.values.title).toBe('string')
// `todos` is absent because its unit belongs to the agent preset and this
// directly seeded session never composed that preset. History computes
// the baseline through the standard projection registry without mounting
// an Agent composition as a read side effect.
expect(projections?.values).not.toHaveProperty('todos')
expect(projections.values).not.toHaveProperty('todos')
// The session-stats unit is a shipped web-app bundle row: whole-log
// turn/step counts ride the same tail block (the stats strip's source).
const sessionStats = projections?.values.sessionStats as { turns: number; steps: number } | undefined
const sessionStats = projections.values.sessionStats as { turns: number; steps: number } | undefined
expect(sessionStats).toBeDefined()
expect(sessionStats?.turns).toBeGreaterThanOrEqual(1)
expect(sessionStats?.steps).toBeGreaterThanOrEqual(sessionStats?.turns ?? 0)
+1 -1
View File
@@ -125,7 +125,7 @@ async function sessionCursor(baseUrl: string, sessionId: string): Promise<number
}
const value = frame.value
if (frame.type === 'item' && isRecord(value)
&& value.type === 'opened' && Number.isSafeInteger(value.cursor)) {
&& value.type === 'snapshot' && Number.isSafeInteger(value.cursor)) {
finish(undefined, value.cursor as number)
}
} catch (error) {
@@ -353,6 +353,38 @@ describe('web e2e: persisted subagent conversation and human continuation', () =
await compareOrRefreshGolden(SIDEBAR_EXPECTED, sidebar, MODE)
})
it('keeps a restored child neutral until its parent availability arrives', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-restore'))
const pattern = '**/api/subagent.list'
let requested = false
let releaseCatalog = (): void => {}
const catalogHeld = new Promise<void>((resolve) => { releaseCatalog = resolve })
await page.route(pattern, async (route) => {
const response = await route.fetch()
requested = true
await catalogHeld
await route.fulfill({ response })
})
const warningStart = tripwire.warnings.length
try {
await page.reload({ waitUntil: 'load' })
await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
await expect.poll(() => requested, { timeout: 15_000 }).toBe(true)
expect(await page.getByText('This subagent is read-only for now', { exact: true }).count()).toBe(0)
expect(await page.locator('[data-composer-seat]').evaluate(element =>
getComputedStyle(element).visibility)).toBe('hidden')
releaseCatalog()
const input = page.getByRole('textbox', { name: 'Message the agent' })
await input.waitFor({ timeout: 15_000 })
await expect.poll(() => input.isEnabled(), { timeout: 15_000 }).toBe(true)
acknowledgeReloadConnectionLoss(tripwire, warningStart)
} finally {
releaseCatalog()
await page.unroute(pattern)
}
})
it('continues through FIFO follow-up admission and receives the child follow events', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-followup'))
const ended = new Promise<void>((resolveEnded, reject) => {
@@ -80,8 +80,15 @@ export function ConversationRoot({
// The exemption is deliberately open-state-wide, not loading-only: a
// summary-blank session is the hero before its open starts (`cold`) and
// after one fails (`error`) for the same reason — there is no history.
const settling = sessionId !== undefined && shellPhase === 'blank' && openState === 'loading'
&& summaryBlank !== true
// A restored continuable subagent also stays settled until its eagerly
// loaded parent catalog establishes availability. This keeps the composer
// hidden instead of briefly rendering the parent-offline takeover.
const parentAvailabilityPending = session?.subagent?.address.mode === 'continuable'
&& session.subagent.parentAvailable === undefined
const settling = sessionId !== undefined && (
(shellPhase === 'blank' && openState === 'loading' && summaryBlank !== true)
|| parentAvailabilityPending
)
const hero = sessionId === undefined
|| (shellPhase === 'blank' && (openState === 'open' || summaryBlank === true))
const zone: InputZone | undefined =
@@ -157,7 +157,7 @@ export function InputBar({
// A continuable child without its live parent cannot accept human input,
// but its independent Stop below stays available while it runs.
const continuable = subagent?.address.mode === 'continuable'
const parentOffline = continuable && !subagent.parentAvailable
const parentOffline = continuable && subagent.parentAvailable !== true
// Running input stays free; locked = session removed, the
// inert no-workspace state, the machine faces absent (no session), or a
// parent-offline continuable child. An owner block also disables input;
@@ -79,7 +79,6 @@ function fakeSessions(ctx: Context): { sessions: ISessions; binding: SessionBind
subagentAddress: () => undefined,
setSubagentCatalogOpen: () => {},
refreshSubagents: () => Promise.reject(new Error('unused fake Sessions operation')),
noteAgentPreset: () => {},
clear: () => {},
refresh: () => Promise.reject(new Error('unused fake Sessions operation')),
search: () => Promise.reject(new Error('unused fake Sessions operation')),
@@ -499,7 +499,6 @@ function CatalogDropdown({
const hoverOpenTimer = useRef<ReturnType<typeof setTimeout> | undefined>(undefined)
const hoverCloseTimer = useRef<ReturnType<typeof setTimeout> | undefined>(undefined)
const observedCatalogs = useRef(new Set<SessionId>())
const requestedInitialCatalog = useRef<SessionId>()
const setCatalogOpenRef = useRef(setCatalogOpen)
setCatalogOpenRef.current = setCatalogOpen
const currentEntry = currentSessionId === undefined
@@ -531,16 +530,6 @@ function CatalogDropdown({
}
: catalog
useEffect(() => {
if (
variant !== 'switcher'
|| catalog !== undefined
|| requestedInitialCatalog.current === rootSessionId
) return
requestedInitialCatalog.current = rootSessionId
refresh(rootSessionId)
}, [catalog, refresh, rootSessionId, variant])
const observeCatalog = (parentSessionId: SessionId, next: boolean): void => {
if (next) observedCatalogs.current.add(parentSessionId)
else observedCatalogs.current.delete(parentSessionId)
@@ -34,7 +34,10 @@ function selectReadOnlySubagent(owner: ComposerChainProps): SubagentReadOnlyMatc
const subagent = owner.session?.subagent
if (subagent === undefined || subagent === null) return null
if (subagent.address.mode === 'one-shot') return { reason: 'one-shot' }
if (subagent.parentAvailable) return null
// The parent catalog is fetched ahead of the selected Session. Until it
// resolves, leave the normal disabled composer in place instead of briefly
// claiming that the parent is offline.
if (subagent.parentAvailable !== false) return null
// A RUNNING parent-offline continuable child keeps the default composer:
// its input is disabled there, but the same primary Stop stays available so
// the child can be interrupted. Once it stops, this takeover returns.
@@ -133,6 +133,7 @@ describe('apply', () => {
// One-shot stays read-only even while running: it has no stop action.
expect(select(owner({ address: { ...address, mode: 'one-shot' }, parentAvailable: true }, true)))
.toEqual({ reason: 'one-shot' })
expect(select(owner({ address }))).toBeNull()
expect(select(owner({ address, parentAvailable: true }))).toBeNull()
expect(select(owner({ address, parentAvailable: false })))
.toEqual({ reason: 'parent-unavailable' })
@@ -770,7 +770,7 @@ describe('SubagentHeaderLineage', () => {
it.each([
['ancestor', vi.fn()],
['current', undefined],
] as const)('refreshes an absent %s switcher catalog without waiting for hover', (_kind, openTitle) => {
] as const)('keeps an absent %s switcher catalog lazy until interaction', (_kind, openTitle) => {
const input = {
...props(undefined, {}, {
[CHILD]: {
@@ -784,7 +784,7 @@ describe('SubagentHeaderLineage', () => {
}
render(<SubagentHeaderLineage {...input} />)
expect(input.refresh).toHaveBeenCalledWith(PARENT)
expect(input.refresh).not.toHaveBeenCalled()
})
it('keeps a nested title switcher scoped to its direct-parent catalog', () => {
@@ -54,6 +54,7 @@
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-session-persistence-sqlite": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/dsh-subagent-fork-in-process": "workspace:^",
"@deepseek-ai/dsh-subagent-spawn-in-process": "workspace:^",
@@ -16,6 +16,7 @@ import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import TeamService, { foldTeam, TeamId, TeamMessageId } from '../src/index.ts'
import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/index.ts'
import { TestSessionQuery } from './test-session-query.ts'
const SIGNAL = new AbortController().signal
const PERSISTENCE_TEST_TIMEOUT_MS = 15_000
@@ -94,6 +95,7 @@ async function stack(
contexts.add(ctx)
await mountAgentLoopTestDependencies(ctx)
await backend.mount(ctx, root)
await ctx.plugin(TestSessionQuery)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
@@ -16,6 +16,7 @@ import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-a
import TeamService, { foldTeam, TeamError, TeamId, TeamMessageId, TeamTaskId } from '../src/index.ts'
import { TeamRuntimeLifecycle } from '../src/lifecycle.ts'
import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/index.ts'
import { TestSessionQuery } from './test-session-query.ts'
const SIGNAL = new AbortController().signal
const roots: string[] = []
@@ -48,6 +49,7 @@ async function setup(
const storageRoot = mkdtempSync(join(tmpdir(), 'dsh-team-'))
roots.push(storageRoot)
await ctx.plugin(JsonlSessionPersistence, { root: storageRoot })
await ctx.plugin(TestSessionQuery)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
@@ -0,0 +1,14 @@
/** Minimal concrete Session query for Agent Team continuation tests. */
import SessionQueryEngine from '@deepseek-ai/dsh-session-query'
/** Session query implementation whose search faces are outside these tests. */
export class TestSessionQuery extends SessionQueryEngine {
override searchSessions(): Promise<never> {
return Promise.reject(new Error('session search is not configured in this test'))
}
override searchEvents(): Promise<never> {
return Promise.reject(new Error('event search is not configured in this test'))
}
}
@@ -48,6 +48,7 @@
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/dsh-subagent-fork-in-process": "workspace:^",
@@ -10,6 +10,7 @@ import { CallId } from '@deepseek-ai/dsh-llm'
import { scopeOf } from '@deepseek-ai/dsh-scope'
import { SessionId } from '@deepseek-ai/dsh-session'
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
import SessionQueryEngine from '@deepseek-ai/dsh-session-query'
import SubagentService from '@deepseek-ai/dsh-subagent'
import * as SubagentFork from '@deepseek-ai/dsh-subagent-fork-in-process'
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
@@ -37,6 +38,17 @@ const TOOL_NAMES = [
const roots: string[] = []
let callNumber = 0
/** Session query implementation whose search faces are outside these tests. */
class TestSessionQuery extends SessionQueryEngine {
override searchSessions(): Promise<never> {
return Promise.reject(new Error('session search is not configured in this test'))
}
override searchEvents(): Promise<never> {
return Promise.reject(new Error('event search is not configured in this test'))
}
}
afterEach(() => {
for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
})
@@ -47,6 +59,7 @@ async function setup(script: ConstructorParameters<typeof MockAdapter>[0], legac
const storageRoot = mkdtempSync(join(tmpdir(), 'dsh-tool-team-'))
roots.push(storageRoot)
await ctx.plugin(JsonlSessionPersistence, { root: storageRoot })
await ctx.plugin(TestSessionQuery)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
if (legacyControl) await ctx.plugin(ToolSubagentControl)
+20 -63
View File
@@ -10,19 +10,18 @@ import type { Agent, ModelSelection } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-agent-presets/types'
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
import { SubagentError } from '@deepseek-ai/dsh-subagent'
import type { SubagentListEntry as CatalogSubagentListEntry } from '@deepseek-ai/dsh-subagent'
import { isUserInvocable } from '@deepseek-ai/dsh-skill'
import {
InvalidPresetIdError, PresetExistsError, PresetMountError,
PresetNotWritableError, resolveSessionPreset, UnknownPresetError,
PresetNotWritableError, UnknownPresetError,
} from '@deepseek-ai/dsh-agent-presets'
import type { PresetBearingSession } from '@deepseek-ai/dsh-agent-presets'
import type {
ApiProxy, ConfigurableProviderView, CredentialView, GoalRef,
SettingsNamespaceView, SubagentAddress,
SettingsNamespaceView,
} from './api/index.ts'
import type { SessionRequestId } from '@deepseek-ai/dsh-api-session-controller/types'
import { ApiSessionNotFound, buildModelCatalog } from '@deepseek-ai/dsh-api-session-controller'
import { buildModelCatalog } from '@deepseek-ai/dsh-api-session-controller'
import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
import {
DEFAULT_SESSION_LOG_COMPRESSION_LEVEL,
flushLiveSessionLog,
@@ -199,49 +198,6 @@ function projectionsUnavailableError(): RpcError {
}
}
/** 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')) {
return { error: { code: 'cancelled', message: 'subagent catalog read was cancelled', details: {} } }
}
if (error instanceof SubagentError && error.code === 'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE') {
return { error: projectionsUnavailableError() }
}
return { error: { code: 'internal', message: 'subagent catalog read failed', details: {} } }
}
}
/**
* The requested preset differs from the one this session already runs.
*
@@ -300,14 +256,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
/** Resolve a Session's live or standing preset scope without resuming it. */
async function sessionScopeFor(
sessionId: SessionId,
session: PresetBearingSession,
agentPreset: string | undefined,
): Promise<ScopeKey | undefined> {
const live = ctx.get('agents')?.get(sessionId)
if (live !== undefined) return live
const presets = ctx.get('agentPresets')
if (presets === undefined) return undefined
try {
return await presets.standingKeyFor(resolveSessionPreset(session))
return await presets.standingKeyFor(agentPreset)
} catch {
// An unknown or unusable recorded preset falls back to the global registry.
return undefined
@@ -536,10 +492,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
details: { parentSessionId },
})
}
const verified = await catalogChild(ctx, {
parentSessionId, childSessionId, mode: 'continuable',
}, signal)
if (verified.error !== undefined) return err(request, verified.error)
try {
const messageId = await ctx.subagents.followup(parent, childSessionId, content, {
source: {
@@ -869,15 +821,21 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
// the view scope is the live agent or the preset's standing key.
async list(request) {
const { sessionId } = request.payload
let session: PresetBearingSession
let cwd: string | undefined
let agentPreset: string | undefined
try {
const inspected = await ctx.sessionController.inspect(sessionId)
session = { header: inspected.meta, events: inspected.events }
using observation = await ctx.sessionQuery.observeSession(sessionId)
if (observation.projections === undefined) {
throw new Error('skill catalog requires a projected Session observation')
}
cwd = observation.header.cwd
agentPreset = observation.projections.values.agentPreset ?? undefined
} catch (error: unknown) {
if (error instanceof ApiSessionNotFound) {
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return err(request, {
code: 'session-not-found',
message: error.message,
message: `session "${sessionId}" not found`,
details: { sessionId },
})
}
@@ -887,12 +845,11 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
details: {},
})
}
if (session.header.cwd === undefined) {
if (cwd === undefined) {
// Every served session records its project at create time; a
// cwd-less header is a pre-project legacy log (not served).
return err(request, { code: 'internal', message: `session "${sessionId}" has no project cwd`, details: {} })
}
const cwd = session.header.cwd
// The host registry is layered per scope and serves every session. A
// composition may still realm-mount its own registry instead; that
// instance is invisible to host contexts, so address it through the
@@ -910,7 +867,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}
// Resolve the live or recorded preset scope so the catalog matches the
// Session composition without resuming its Agent.
const scope = await sessionScopeFor(sessionId, session)
const scope = await sessionScopeFor(sessionId, agentPreset)
try {
const skills = (await skillRegistry.list({ cwd, scope })).filter(isUserInvocable)
return ok(request, {
@@ -1065,7 +1022,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
},
async models(request) {
return ok(request, await buildModelCatalog(ctx))
return ok(request, await buildModelCatalog(ctx, defaults.defaultModelSelection()))
},
async discoverModels(request, signal) {
+1 -1
View File
@@ -30,7 +30,7 @@ export interface ApiProxy {
// ---- Domain interfaces and payload entities ----
export type {
ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelCatalog, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelSelection,
} from '@deepseek-ai/dsh-api-session-controller/types'
export type { DirectoryEntry, DirectoryListing, HostApi } from './host.ts'
@@ -10,6 +10,7 @@ import type { ConfigurableProviderView, DiscoveredModelView } from './llm.ts'
import type {
ModelCatalogFailure,
ModelCatalogModel,
ModelSelection,
ModelProviderGroup,
ModelReasoning,
ModelReasoningEffort,
@@ -50,6 +51,13 @@ const modelCatalogFailureSchema = z.object({
message: z.string(),
}) satisfies z.ZodType<Wire<ModelCatalogFailure>>
/** Complete model selection used as the Host default. */
const modelSelectionSchema = z.object({
provider: z.string().min(1),
model: z.string().min(1),
reasoningEffort: z.string().min(1).optional(),
}) satisfies z.ZodType<Wire<ModelSelection>>
/** ConfigurableProviderView row of llm.providers. */
export const configurableProviderViewSchema = z.object({
provider: z.string().min(1),
@@ -73,6 +81,8 @@ export const llmModelsRequestSchema = z.object({}) satisfies z.ZodType<Wire<Requ
/** llm.models response value. */
export const llmModelsValueSchema = z.object({
default: modelSelectionSchema,
routableProviders: z.array(z.string().min(1)),
groups: z.array(modelProviderGroupSchema),
failures: z.array(modelCatalogFailureSchema),
}) satisfies z.ZodType<Wire<ResponseValue<'llm.models'>>>
+2 -3
View File
@@ -9,8 +9,7 @@
import type { RpcRequest, RpcResponse } from './rpc.ts'
import type {
ModelCatalogFailure,
ModelProviderGroup,
ModelCatalog,
} from '@deepseek-ai/dsh-api-session-controller/types'
/** Wire view of one configurable provider. */
@@ -48,7 +47,7 @@ export interface LlmApi {
* settings surface's models view, needing no session. Per-provider listing
* failures ride `failures` without failing the sound groups.
*/
models(request: RpcRequest<{}>): Promise<RpcResponse<{ groups: ModelProviderGroup[]; failures: ModelCatalogFailure[] }>>
models(request: RpcRequest<{}>): Promise<RpcResponse<ModelCatalog>>
/**
* Interrogate a provider endpoint the configuration surface is still
@@ -10,12 +10,13 @@ import SessionStore, { SessionId, type Session } from '@deepseek-ai/dsh-session'
import { RpcId, type RpcRequest } from '../src/api/rpc.ts'
import type { ApiProxy } from '../src/api/index.ts'
import {
InvalidPresetIdError, PresetExistsError, resolveSessionPreset, UnknownPresetError,
agentPresetProjectionDefinition, InvalidPresetIdError, PresetExistsError, UnknownPresetError,
} from '@deepseek-ai/dsh-agent-presets'
import type {} from '@deepseek-ai/dsh-agent-presets/types'
import { GoalId } from '@deepseek-ai/dsh-goal'
import { createApiProxy } from '../src/api-proxy.ts'
import { describe, expect, it } from 'vitest'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
let nextRpc = 0
function request<P>(payload: P): RpcRequest<P> {
@@ -122,6 +123,35 @@ async function harness(
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
if (presets !== undefined) ctx.provide('agentPresets', roster(presets, options.userIds) as never)
ctx.provide('sessionQuery', {
observeSession: (sessionId: SessionId) => {
const session = ctx.sessions.get(sessionId)
if (session === undefined) {
return Promise.reject(new SessionQueryError(
`session "${sessionId}" not found`,
'SESSION_QUERY_SESSION_NOT_FOUND',
))
}
let preset = agentPresetProjectionDefinition.init(session.header)
for (const event of session.events) {
preset = agentPresetProjectionDefinition.apply(preset, event)
}
const events = Object.freeze([...session.events])
const lease = (): SessionObservation => ({
source: 'live' as const,
header: session.header,
events,
cursor: events.at(-1)?.seq ?? -1,
projections: {
asOfSeq: events.at(-1)?.seq ?? -1,
values: { agentPreset: preset },
},
retain: lease,
[Symbol.dispose]: () => {},
})
return Promise.resolve(lease())
},
} as never)
const factory: AgentFactory = {
async createAgent(_ownerCtx, options) {
@@ -289,7 +319,8 @@ describe('agentPreset.select', () => {
const session = ctx.sessions.get(SessionId('sel-log'))
if (session === undefined) throw new Error('unreachable')
expect(session.header.agentPreset).toBe('standard')
expect(resolveSessionPreset(session)).toBe('minimal')
expect(session.events.findLast(event => event.type === 'agent-preset/selected')?.data)
.toEqual({ agentPreset: 'minimal' })
})
it('serializes two concurrent selects on one session', async () => {
@@ -309,7 +340,8 @@ describe('agentPreset.select', () => {
const session = ctx.sessions.get(SessionId('sel-race'))
if (session === undefined) throw new Error('unreachable')
// One winner, and the log agrees with it: the last committed switch.
expect(resolveSessionPreset(session)).toBe('standard')
expect(session.events.findLast(event => event.type === 'agent-preset/selected')?.data)
.toEqual({ agentPreset: 'standard' })
})
it('refuses once the conversation has started', async () => {
@@ -679,6 +679,8 @@ describe('llm domain', () => {
ctx.llm.registerAdapter(['broken'], new BrokenCatalogAdapter('Broken', []))
const api = createApiProxy(ctx, DEFAULTS)
const value = expectOk(await api.llm.models(request({})))
expect(value.default).toEqual({ provider: 'p', model: 'm' })
expect(value.routableProviders).toEqual(['deepseek-official', 'broken'])
expect(value.groups).toEqual([{
id: 'deepseek-official',
name: 'DeepSeek',
@@ -1,7 +1,7 @@
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import { ApiSessionNotFound } from '@deepseek-ai/dsh-api-session-controller'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type {} from '@deepseek-ai/dsh-skill'
import { describe, expect, it, vi } from 'vitest'
import { createApiProxy } from '../src/api-proxy.ts'
@@ -14,11 +14,18 @@ describe('skill catalog Session inspection', () => {
await ctx.plugin(AgentRegistry)
const sessionId = SessionId('cold-skills')
const resolveAgent = vi.fn()
const inspect = vi.fn(() => Promise.resolve({
meta: { version: 0 as const, id: sessionId, createdAt: 1, cwd: '/cold/project' },
const dispose = vi.fn()
const observeSession = vi.fn(() => Promise.resolve({
source: 'live',
header: { version: 0 as const, id: sessionId, createdAt: 1, cwd: '/cold/project' },
events: [],
}))
ctx.provide('sessionController', { inspect, resolveAgent } as never)
cursor: -1,
projections: { asOfSeq: -1, values: {} },
retain: () => { throw new Error('not retained') },
[Symbol.dispose]: dispose,
} satisfies SessionObservation))
ctx.provide('sessionQuery', { observeSession } as never)
ctx.provide('sessionController', { resolveAgent } as never)
const list = vi.fn(() => Promise.resolve([{
name: 'review',
description: 'Review the current change.',
@@ -42,7 +49,8 @@ describe('skill catalog Session inspection', () => {
}],
},
})
expect(inspect).toHaveBeenCalledWith(sessionId)
expect(observeSession).toHaveBeenCalledWith(sessionId)
expect(dispose).toHaveBeenCalledOnce()
expect(resolveAgent).not.toHaveBeenCalled()
expect(list).toHaveBeenCalledWith({ cwd: '/cold/project', scope: undefined })
})
@@ -51,7 +59,10 @@ describe('skill catalog Session inspection', () => {
const sessionId = SessionId('missing-skills')
for (const fixture of [
{
error: new ApiSessionNotFound('session "missing-skills" not found'),
error: new SessionQueryError(
'session "missing-skills" not found',
'SESSION_QUERY_SESSION_NOT_FOUND',
),
code: 'session-not-found',
},
{ error: new Error('storage offline'), code: 'internal' },
@@ -59,9 +70,8 @@ describe('skill catalog Session inspection', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
ctx.provide('sessionController', {
inspect: () => Promise.reject(fixture.error),
resolveAgent: vi.fn(),
ctx.provide('sessionQuery', {
observeSession: () => Promise.reject(fixture.error),
} as never)
ctx.provide('skills', { list: vi.fn() } as never)
const api = createApiProxy(ctx, {
@@ -107,7 +107,7 @@ describe('subagent gateway', () => {
.toMatchObject({ ok: true, value: { entries: [{ activity: 'running' }] } })
})
it('maps the missing projections capability to one wire face on list and prompt', async () => {
it('maps missing catalog projections on list without preflighting prompt delivery', async () => {
const listError = () => new SubagentError(
'listing subagents requires the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE',
@@ -124,8 +124,9 @@ describe('subagent gateway', () => {
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()
}), new AbortController().signal)).result).toMatchObject({ ok: true })
expect(prompt.listChildren).not.toHaveBeenCalled()
expect(prompt.followup).toHaveBeenCalledOnce()
})
it('routes human content through the exact live parent with rpc attribution', async () => {
@@ -81,7 +81,12 @@ function scriptedApi(overrides: {
},
llm: {
providers: r => ok(r, { providers: [] }),
models: r => ok(r, { groups: [], failures: [] }),
models: r => ok(r, {
default: { provider: 'test', model: 'test' },
routableProviders: [],
groups: [],
failures: [],
}),
discoverModels: err,
...overrides.llm,
},
@@ -439,7 +444,12 @@ describe('config unary surface', () => {
},
llm: {
providers: record('llm.providers', r => ok(r, { providers: [providerRow] })),
models: record('llm.models', r => ok(r, { groups: [group], failures: [] })),
models: record('llm.models', r => ok(r, {
default: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
routableProviders: ['deepseek-official'],
groups: [group],
failures: [],
})),
discoverModels: record('llm.discoverModels', r => ok(r, { models: [{ id: 'acme-large', contextWindow: 65536 }] })),
},
})
@@ -465,7 +475,15 @@ describe('config unary surface', () => {
const providers = await c.llm.providers({})
expect(providers.result).toEqual({ ok: true, value: { providers: [providerRow] } })
const models = await c.llm.models({})
expect(models.result).toEqual({ ok: true, value: { groups: [group], failures: [] } })
expect(models.result).toEqual({
ok: true,
value: {
default: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
routableProviders: ['deepseek-official'],
groups: [group],
failures: [],
},
})
const discovered = await c.llm.discoverModels({
settingsNs: 'llm-pi-ai',
baseURL: 'https://gateway.acme.example/v1',
@@ -140,7 +140,18 @@ function fakeApi(overrides: Partial<{ crashOn: string }> = {}): ApiProxy {
return { rpcId: request.rpcId, result: { ok: true, value: { providers: [] } } }
},
async models(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { groups: [], failures: [] } } }
return {
rpcId: request.rpcId,
result: {
ok: true,
value: {
default: { provider: 'test', model: 'test' },
routableProviders: [],
groups: [],
failures: [],
},
},
}
},
async discoverModels(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { models: [] } } }
+5
View File
@@ -52,6 +52,7 @@
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"@deepseek-ai/dsh-session-projection": "workspace:^",
"@deepseek-ai/dsh-session-projection-cache": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-jobs": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-user-approval": "workspace:^",
@@ -76,6 +77,9 @@
"@deepseek-ai/dsh-session-projection-cache": {
"optional": true
},
"@deepseek-ai/dsh-session-query": {
"optional": true
},
"@deepseek-ai/dsh-jobs": {
"optional": true
},
@@ -96,6 +100,7 @@
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"@deepseek-ai/dsh-session-projection": "workspace:^",
"@deepseek-ai/dsh-session-projection-cache": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-storage": "workspace:^",
"@deepseek-ai/dsh-storage-domain": "workspace:^",
"@deepseek-ai/dsh-jobs": "workspace:^",
+26 -8
View File
@@ -35,6 +35,7 @@ import type { ContentBlock, MessageId, MessageSource } from '@deepseek-ai/dsh-ll
import { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import type { SessionObservation, SessionQueryEngine } from '@deepseek-ai/dsh-session-query'
import type { ToolRestriction } from '@deepseek-ai/dsh-tools'
import { foldSubagentDescriptor, snapshotSubagentDescriptor } from './descriptor.ts'
import type { SubagentDescriptorData } from './descriptor.ts'
@@ -939,7 +940,7 @@ export class SubagentContinuationManager {
}
/**
* Cold-resume a persisted child: inspect and authorize its Session, fold the
* Cold-resume a persisted child: retain and authorize its prepared Session, fold the
* generic descriptor, create the Activation through `ctx.agents.resume()`,
* and submit the waiting turn. This never dispatches through a subagent
* provider the persisted Session already holds the initial prefix and the
@@ -951,23 +952,27 @@ export class SubagentContinuationManager {
content: ContentBlock[],
options: SubagentFollowupOptions,
): Promise<MessageId> {
const persistence = this.requirePersistence()
let loaded: Awaited<ReturnType<typeof persistence.inspect>>
const query = this.requireSessionQuery()
let observation: SessionObservation
try {
loaded = await persistence.inspect(childId, options.signal)
observation = await query.observeSession(childId, {
signal: options.signal,
})
} catch (error: unknown) {
options.signal.throwIfAborted()
throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error })
}
options.signal.throwIfAborted()
using source = observation
this.assertAdmitting(parent)
// Authorize the persisted header before folding: only the durable child's
// exact live direct parent may continue it.
this.authorizeLineage(parent, childId, loaded.meta.parentSession)
this.authorizeLineage(parent, childId, source.header.parentSession)
// Fold only the child's own suffix: a fork seed replays the parent's log,
// which may carry an ANCESTOR's descriptor when the parent is itself a
// continuable child.
const descriptor = foldSubagentDescriptor(loaded.events.slice(loaded.meta.seedLength ?? 0))
const descriptor = foldSubagentDescriptor(
source.events.slice(source.header.seedLength ?? 0),
)
if (descriptor === undefined || descriptor.mode !== 'continuable') {
throw new SubagentError(
`subagent "${childId}" has no supported continuation state and cannot be resumed; `
@@ -996,7 +1001,7 @@ export class SubagentContinuationManager {
if (error instanceof SubagentError) throw error
throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error })
}
return this.submitMaterialized(activation, content, options.source, parent, options.signal)
return await this.submitMaterialized(activation, content, options.source, parent, options.signal)
}
/**
@@ -1545,6 +1550,19 @@ export class SubagentContinuationManager {
}
return persistence
}
/** Resolve the Session query service used for cold child observations. */
private requireSessionQuery(): SessionQueryEngine {
const query = this.ctx.get('sessionQuery')
if (query === undefined) {
throw new SubagentError(
'continuable subagents require session query (load @deepseek-ai/dsh-session-query)',
'CONTINUATION_UNAVAILABLE',
)
}
return query
}
}
export type { SubagentDescriptorData }
+7 -18
View File
@@ -325,27 +325,16 @@ export class SubagentRuntime extends Service {
/**
* Enumerate the parent's direct session-backed subagents without loading or
* resuming an Agent and without any query service: the listing merges the live
* session store with optional session persistence (live-preferred) and
* serves each child's durable mode/label from the registered `subagent`
* projection unit down a three-rung ladder the registry's watermark
* snapshot for a live child; for a cold one, a durable projection-cache
* row when the optional cache serves an own-suffix identity (its `seq`
* gate proves the value postdates the fork seed, where a child's own
* descriptor is immutable once appended), else one persistence inspection
* folded through the registry. The
* projection fold is the single classification authority; per-child
* diagnostics relay a fold that served no identity or a failed inspection,
* never a list-time descriptor parse. Absent persistence, enumeration is
* live-only (a cold child cannot be resumed then either, so its absence is
* capability absence, not an error). This service consults no Agent
* registrations, Activations, or providers.
* resuming an Agent. The Session query service supplies one live-preferred
* corpus and shared point observations; the projection cache supplies
* immutable descriptor hits without opening cold logs. The registered
* `subagent` projection remains the sole mode/label classifier.
*
* Every persistence read receives `signal`, and the listing rechecks
* cancellation around each of those awaits. Read rejections that settle
* Every query receives `signal`, and the listing rechecks cancellation
* around each await. Read rejections that settle
* after an abort become a stable `SubagentError` with code `CANCELLED`.
* @param parentSessionId - parent session whose direct children are listed.
* @param signal - caller-owned cancellation forwarded to persistence reads
* @param signal - caller-owned cancellation forwarded to Session queries
* and observed around every read await.
* @returns children and per-child diagnostics ordered by `createdAt`, then id.
* @throws {@link SubagentError} when the projection registry or the session
+76 -69
View File
@@ -1,13 +1,13 @@
/**
* Read-only enumeration of durable subagent children and descendant trees
* straight from the live session store and optional session persistence no
* query service. Candidates come from one live-preferred corpus; each child's
* mode/label is the registered `subagent` projection unit's value, resolved
* through the Session query service. Candidates come from one live-preferred
* corpus; each child's mode/label is the registered `subagent` projection
* unit's value, resolved
* down a three-rung ladder: the registry's watermark cache for a live child,
* a durable projection-cache row when it serves an own-suffix identity (the
* seq gate), and one persistence inspection folded through the registry
* otherwise, validated against the enumerated lifecycle. The projection fold
* is the single classification authority this module parses no descriptor
* seq gate), and one shared Session observation otherwise, validated against
* the enumerated lifecycle. The projection fold is the single classification
* authority this module parses no descriptor
* itself. Absent persistence, enumeration is live-only: a cold child is
* unreachable for resume anyway, so its absence is capability absence, not an
* error. The module owns no catalog state and does not consult Activation,
@@ -17,17 +17,17 @@
*/
import type { Context } from '@deepseek-ai/cordis'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import type { Session, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionProjectionRegistry } from '@deepseek-ai/dsh-session-projection'
import type { SessionProjectionCache } from '@deepseek-ai/dsh-session-projection-cache'
import type { SessionObservation, SessionQueryEngine } from '@deepseek-ai/dsh-session-query'
import { SubagentError } from './error.ts'
import type { SubagentIdentityProjection } from './projection-types.ts'
/**
* Concurrent cold inspections per listing; a constant because it bounds one
* read-only scan of local media, not deployment behavior. Should a networked
* persistence backend appear, promote it to a validated `Config` field.
* Concurrent cold observations per explicit catalog listing. Current Session
* persistence providers are local; a networked provider must promote this to
* a validated deployment setting.
*/
const COLD_READ_CONCURRENCY = 4
@@ -79,8 +79,8 @@ export type SubagentListEntry =
* unrecognized-version descriptor deliberately undistinguished), and
* for any candidate whose log makes a registered unit's fold or schema
* throw (deterministic data damage, contained per child); `unavailable`
* when the candidate's persistence inspection failed (retried on the
* next listing). `unsupported` is never produced; it remains in the
* when the candidate's Session observation was absent or transiently
* unreadable (retried on the next listing). `unsupported` is never produced; it remains in the
* union for consumers that route on it.
*/
readonly reason: 'corrupt' | 'unsupported' | 'unavailable'
@@ -102,7 +102,7 @@ type CorpusRecord = { readonly header: SessionHeader; readonly live: Session | u
interface ListingRuntime {
readonly projections: SessionProjectionRegistry
readonly persistence: SessionPersistence | undefined
readonly query: SessionQueryEngine
readonly cache: SessionProjectionCache | undefined
readonly corpus: ReadonlyMap<SessionId, CorpusRecord>
readonly subagentParents: ReadonlySet<SessionId>
@@ -120,8 +120,7 @@ interface PositionedCandidate {
* serving each identity from the `subagent` projection unit: the registry's
* watermark snapshot for a live child; for a cold one, a durable
* projection-cache row when it serves an own-suffix identity (the seq gate),
* else one bounded-concurrency persistence inspection folded through the
* registry.
* else one bounded-concurrency shared Session observation.
* @see SubagentRuntime.listChildren for the public cancellation and failure contract.
* @param ctx - context carrying the session store, the projection registry,
* optional persistence, and the optional projection cache.
@@ -206,29 +205,34 @@ async function prepareListing(
)
}
assertListingNotCancelled(signal)
const persistence = ctx.get('sessionPersistence')
const query = ctx.get('sessionQuery')
if (query === undefined) {
throw new SubagentError(
'listing subagents requires the sessionQuery service (load @deepseek-ai/dsh-session-query)',
'SUBAGENT_CONTROL_QUERY_UNAVAILABLE',
)
}
// Optional acceleration only: an absent cache service just means every
// cold candidate takes the authoritative preparation rung, so it carries
// no error code and no configuration check.
const cache = ctx.get('sessionProjectionCache')
let persistedHeaders: readonly SessionHeader[] = []
if (persistence !== undefined) {
try {
persistedHeaders = await persistence.list(signal)
} catch (error: unknown) {
// The backend may reject with its own abort failure after observing the
// forwarded signal; cancellation stays a stable subagent failure.
assertListingNotCancelled(signal)
throw error
}
let records: Awaited<ReturnType<SessionQueryEngine['listSessions']>>
try {
records = await query.listSessions(signal)
} catch (error: unknown) {
assertListingNotCancelled(signal)
throw error
}
assertListingNotCancelled(signal)
// Live-preferred merge without header reconciliation: a live record wins
// its id wholesale, exactly as a live-preferred corpus would serve it.
const corpus = new Map<SessionId, CorpusRecord>()
for (const header of persistedHeaders) corpus.set(header.id, { header, live: undefined })
for (const session of sessions.list()) {
corpus.set(session.header.id, { header: session.header, live: session })
for (const record of records) {
const live = sessions.get(record.header.id)
corpus.set(record.header.id, {
header: live?.header ?? record.header,
live,
})
}
const subagentParents = new Set<SessionId>()
for (const record of corpus.values()) {
@@ -236,7 +240,7 @@ async function prepareListing(
subagentParents.add(record.header.parentSession)
}
}
return { projections, persistence, cache, corpus, subagentParents }
return { projections, query, cache, corpus, subagentParents }
}
/** Resolve projection-backed rows for aligned candidates with bounded cold reads. */
@@ -245,7 +249,7 @@ async function resolveCandidateRows(
listing: ListingRuntime,
signal: AbortSignal | undefined,
): Promise<(SubagentListEntry | undefined)[]> {
const { projections, persistence, cache, subagentParents } = listing
const { projections, query, cache, subagentParents } = listing
const rows: (SubagentListEntry | undefined)[] = Array.from({ length: candidates.length })
const coldReads: { index: number; header: SessionHeader }[] = []
candidates.forEach((candidate, index) => {
@@ -254,36 +258,33 @@ async function resolveCandidateRows(
coldReads.push({ index, header: candidate.header })
return
}
// The registry's watermark cache serves the live value with zero log
// reads; a live child without an identity yet is the creation window
// before the establishing provider appends its descriptor.
// Read only the identity unit. A live child without an identity yet is the
// creation window before the establishing provider appends its descriptor.
let identity: SubagentIdentityProjection | null | undefined
try {
identity = projections.snapshot(candidate.live).values.subagent
identity = projections.snapshot(candidate.live, ['subagent']).values.subagent
} catch {
// The snapshot folds EVERY registered unit over this child's log, so
// any unit's fold or schema can reject damaged payloads. That is
// deterministic data damage in this one child; it degrades to one
// corrupt diagnostic instead of failing the whole listing.
// A rejecting identity fold is deterministic data damage in this child;
// contain it as one diagnostic instead of failing the whole listing.
rows[index] = { kind: 'diagnostic', id: childId, reason: 'corrupt' }
return
}
// The unit's serializable no-value sentinel is `null`; `undefined` can
// only mean the key was dropped at a JSON boundary. Both are no value.
if (identity === undefined || identity === null) return
if (identity === undefined || identity === null
|| identity.seq < (candidate.header.seedLength ?? 0)) return
rows[index] = childRow(childId, identity, 'running', subagentParents.has(childId))
})
// Cold candidates exist only when persistence listed them, so the narrow
// re-check is about types, not reachability.
if (persistence !== undefined && coldReads.length > 0) {
// Cold candidates came from the query corpus and are resolved concurrently.
if (coldReads.length > 0) {
const queue = [...coldReads]
await Promise.all(Array.from(
{ length: Math.min(COLD_READ_CONCURRENCY, queue.length) },
async () => {
for (let job = queue.shift(); job !== undefined; job = queue.shift()) {
rows[job.index] = await resolveColdIdentity(
persistence, projections, cache, job.header,
query, cache, job.header,
subagentParents.has(job.header.id), signal,
)
}
@@ -338,16 +339,14 @@ function compareCorpusRecords(a: CorpusRecord, b: CorpusRecord): number {
/**
* Resolve one cold candidate down the remaining ladder: a durable
* projection-cache row when it serves an own-suffix identity (the seq gate),
* otherwise one persistence inspection folded through the projection
* registry (the same detached recipe the API proxy uses for detached session
* projections). A failed inspection is one transient `unavailable` row
* retried on the next listing; an inspection naming another lifecycle, and a
* otherwise one shared Session observation. An absent or transiently failed
* observation is one `unavailable` row retried on the next listing; an observation
* source naming another lifecycle, and a
* settled log the fold cannot identify or that makes any registered unit
* throw are final, so they report `corrupt`.
*/
async function resolveColdIdentity(
persistence: SessionPersistence,
projections: SessionProjectionRegistry,
query: SessionQueryEngine,
cache: SessionProjectionCache | undefined,
header: SessionHeader,
hasChildren: boolean,
@@ -357,7 +356,7 @@ async function resolveColdIdentity(
if (cache !== undefined) {
let cached: SubagentIdentityProjection | null | undefined
try {
cached = cache.cachedSnapshot(header)?.values.subagent
cached = cache.cachedSnapshot(header, ['subagent'])?.values.subagent
} catch {
// Unlike the preparation fold below, a throwing cache read renders no
// verdict: the cache is derived data, so its damage (a poisoned stored
@@ -377,32 +376,35 @@ async function resolveColdIdentity(
}
}
assertListingNotCancelled(signal)
let inspected: { meta: SessionHeader; events: readonly SessionEvent[] }
let observation: SessionObservation
try {
inspected = await persistence.inspect(childId, signal)
} catch {
// Per-child isolation: the child vanished or its backend read failed —
// one diagnostic row, and the listing itself still succeeds.
observation = await query.observeSession(childId, {
...(signal === undefined ? {} : { signal }),
})
} catch (error: unknown) {
// Per-child isolation: durable corruption is stable; absence and backend
// failures remain retryable. Either way, the listing itself still succeeds.
assertListingNotCancelled(signal)
return { kind: 'diagnostic', id: childId, reason: 'unavailable' }
return {
kind: 'diagnostic',
id: childId,
reason: sessionQueryCode(error) === 'SESSION_QUERY_CORRUPT_SESSION'
|| sessionQueryCode(error) === 'SESSION_QUERY_SOURCE_CONFLICT'
? 'corrupt'
: 'unavailable',
}
}
using ownedObservation = observation
assertListingNotCancelled(signal)
// A session id names a slot, not a lifecycle: a child deleted and
// re-published under another owner between the enumeration and this read
// must not leak into the old parent's listing.
if (!sameLifecycle(inspected.meta, header)) {
if (!sameLifecycle(ownedObservation.header, header)) {
return { kind: 'diagnostic', id: childId, reason: 'corrupt' }
}
let identity: SubagentIdentityProjection | null | undefined
try {
identity = projections.restore({}, inspected.events, 0).snapshot.values.subagent
} catch {
// The restore folds EVERY registered unit over this child's log, so any
// unit's fold or schema can reject damaged payloads — deterministic data
// damage in this one child, contained as its own corrupt diagnostic.
return { kind: 'diagnostic', id: childId, reason: 'corrupt' }
}
if (identity === undefined || identity === null) {
const identity = ownedObservation.projections?.values.subagent
if (identity === undefined || identity === null
|| identity.seq < (header.seedLength ?? 0)) {
return { kind: 'diagnostic', id: childId, reason: 'corrupt' }
}
return childRow(childId, identity, 'inactive', hasChildren)
@@ -437,6 +439,7 @@ function childRow(
/** Immutable header fields that distinguish one session lifecycle from another under the same id. */
const LIFECYCLE_WITNESS_KEYS = [
'version', 'id', 'createdAt', 'cwd', 'parentSession', 'seedLength', 'delegationDepth',
'origin', 'agentPreset',
] as const
/** Whether an inspected log still belongs to the enumerated lifecycle. */
@@ -450,3 +453,7 @@ function assertListingNotCancelled(signal: AbortSignal | undefined): void {
throw new SubagentError('subagent listing was cancelled', 'CANCELLED')
}
}
function sessionQueryCode(error: unknown): unknown {
return error instanceof Error && 'code' in error ? error.code : undefined
}
@@ -24,6 +24,7 @@ import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
import ApprovalService, { effectiveApprovalPolicy } from '@deepseek-ai/dsh-user-approval'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import SubagentRuntime from '../src/index.ts'
import { TestSessionQuery } from './test-session-query.ts'
type Script = ConstructorParameters<typeof MockAdapter>[0]
@@ -45,6 +46,7 @@ async function setup(script: Script) {
await ctx.plugin(SandboxPolicyService, { mode: 'workspace-write', workspaceRoot: root })
await ctx.plugin(ApprovalService)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(TestSessionQuery)
await ctx.plugin(SubagentRuntime)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(SubagentFork, { providerName: 'fork' })
@@ -22,6 +22,7 @@ import SubagentRuntime, {
} from '../src/index.ts'
import type { SubagentRunEndInfo, SubagentRunInfo } from '../src/index.ts'
import * as SubagentInvariant from '../src/invariant.ts'
import { TestSessionQuery } from './test-session-query.ts'
type Script = ConstructorParameters<typeof MockAdapter>[0]
@@ -65,7 +66,10 @@ afterEach(async () => {
})
/** Boot the full continuable stack: loop, persistence, providers, and subagents. */
async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean } = {}) {
async function setupWith(
adapter: LlmAdapter,
options: { persistence?: boolean; sessionQuery?: boolean } = {},
) {
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
let disposePersistence: (() => Promise<void>) | undefined
@@ -81,6 +85,7 @@ async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean }
})
}
await ctx.plugin(AgentLoop, { agents: [] })
if (options.sessionQuery !== false) await ctx.plugin(TestSessionQuery)
await ctx.plugin(SubagentRuntime)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(SubagentFork, { providerName: 'fork' })
@@ -455,6 +460,7 @@ describe('SubagentRuntime.startContinuable', () => {
// afterEach closes it before removing the root (even on a failure path).
cleanups.push(async () => { await freshPersistence.dispose() })
await fresh.plugin(AgentLoop, { agents: [] })
await fresh.plugin(TestSessionQuery)
await fresh.plugin(SubagentRuntime)
await fresh.plugin(SubagentSpawn, { providerName: 'spawn' })
const freshParent = fresh.agentLoop.create(SessionId('routeless-resume'), {})
@@ -518,6 +524,16 @@ describe('SubagentRuntime.startContinuable', () => {
})
describe('SubagentRuntime.followup residency routing', () => {
it('fails a cold follow-up when Session query is unavailable', async () => {
const { ctx, parent } = await setupWith(new MockAdapter([]), {
persistence: false,
sessionQuery: false,
})
await expect(followup(ctx, parent, SessionId('cold-without-query'), message('continue')))
.rejects.toMatchObject({ code: 'CONTINUATION_UNAVAILABLE' })
})
it('enqueues in the same Activation while it is running, preserving one inbox FIFO', async () => {
const releaseFirst = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
@@ -669,7 +685,7 @@ describe('SubagentRuntime.followup residency routing', () => {
const started = await ctx.subagents.startContinuable(startSpec(parent))
await waitNoActivation(ctx, started.childId)
const inspectStarted = Promise.withResolvers<undefined>()
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect').mockImplementation((_id, signal) => {
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession').mockImplementation((_id, signal) => {
return new Promise<never>((_resolve, reject) => {
if (signal === undefined) {
reject(new Error('cold inspection must receive the followup signal'))
@@ -9,6 +9,7 @@ import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
@@ -23,6 +24,7 @@ import SubagentRuntime, {
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
import * as SubagentFork from '@deepseek-ai/dsh-subagent-fork-in-process'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import { TestSessionQuery } from './test-session-query.ts'
type Script = ConstructorParameters<typeof MockAdapter>[0]
@@ -51,6 +53,7 @@ async function setup(
ctx.provide('storageDomain', facility)
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
}
await ctx.plugin(TestSessionQuery)
await ctx.plugin(SubagentRuntime)
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
await ctx.plugin(SubagentFork, { providerName: 'fork' })
@@ -152,10 +155,11 @@ const hostileProjectionDefinition = {
} satisfies ProjectionDefinition<'subagentListHostileProbe', { poisoned?: boolean | undefined }>
describe('SubagentRuntime.listChildren', () => {
it('lists live children without persistence, query services, or the continuation runtime', async () => {
it('lists live children without persistence or the continuation runtime', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionProjectionRegistry)
await ctx.plugin(TestSessionQuery)
await ctx.plugin(SubagentRuntime)
expect(ctx.get('jobs')).toBeUndefined()
expect(ctx.get('agents')).toBeUndefined()
@@ -196,6 +200,17 @@ describe('SubagentRuntime.listChildren', () => {
)
})
it('fails loud when the Session query service is not mounted', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionProjectionRegistry)
await ctx.plugin(SubagentRuntime)
await expect(ctx.subagents.listChildren(SessionId('no-query-parent'))).rejects.toThrow(
expect.objectContaining({ code: 'SUBAGENT_CONTROL_QUERY_UNAVAILABLE' }) as Error,
)
})
it('lists a persisted continuable child as inactive with its durable label', async () => {
const { ctx, parent } = await setup([textResponse('done')])
const childId = await startChild(ctx, parent, 'summarize the doc')
@@ -301,6 +316,76 @@ describe('SubagentRuntime.listChildren', () => {
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([])
})
it('contains a live child projection failure as one corrupt diagnostic', async () => {
const { ctx, parent } = await setup([])
const childId = SessionId('live-projection-failure')
const child = ctx.sessions.create(childId, {
meta: { parentSession: parent.id, origin: 'subagent' },
})
child.append('subagent/descriptor', descriptorPayload('broken live child'))
const snapshot = ctx.sessionProjections.snapshot.bind(ctx.sessionProjections)
vi.spyOn(ctx.sessionProjections, 'snapshot').mockImplementation((session, keys) => {
if (session.id === childId) throw new Error('projection failed')
return snapshot(session, keys)
})
await expect(ctx.subagents.listChildren(parent.id)).resolves.toContainEqual({
kind: 'diagnostic', id: childId, reason: 'corrupt',
})
})
it('maps a non-Error cold observation failure to unavailable', async () => {
const { ctx, parent } = await setup([])
const childId = await authorChild(ctx, '00000000-0000-4000-8000-00000000aa01', {
parentSession: parent.id,
origin: 'subagent',
}, childEvents(descriptorPayload('unreadable child')))
const observe = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation((id, options) => {
if (id === childId) {
return Promise.reject('backend unavailable') // oxlint-disable-line typescript/prefer-promise-reject-errors
}
return observe(id, options)
})
await expect(ctx.subagents.listChildren(parent.id)).resolves.toContainEqual({
kind: 'diagnostic', id: childId, reason: 'unavailable',
})
})
it('releases a cold observation when cancellation lands after its read', async () => {
const { ctx, parent } = await setup([])
const controller = new AbortController()
const childId = await authorChild(ctx, '00000000-0000-4000-8000-00000000aa02', {
parentSession: parent.id,
origin: 'subagent',
}, childEvents(descriptorPayload('cancelled child')))
const dispose = vi.fn()
vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation((id) => {
if (id !== childId) throw new Error(`unexpected observation: ${id}`)
controller.abort(new Error('cancelled after observation'))
return Promise.resolve({
source: 'prepared',
header: {
version: SESSION_FORMAT_VERSION,
id: childId,
createdAt: 1,
parentSession: parent.id,
origin: 'subagent',
},
events: [],
cursor: -1,
projections: { asOfSeq: -1, values: {} },
retain: vi.fn(),
[Symbol.dispose]: dispose,
} as unknown as SessionObservation)
})
await expect(ctx.subagents.listChildren(parent.id, controller.signal))
.rejects.toMatchObject({ code: 'CANCELLED' })
expect(dispose).toHaveBeenCalledOnce()
})
it('lists a one-shot child with its durable creation label', async () => {
const { ctx, parent } = await setup([])
const labeled = await authorChild(ctx, '00000000-0000-4000-8000-00000000ab02', {
@@ -428,7 +513,7 @@ describe('SubagentRuntime.listChildren', () => {
asOfSeq: 2,
values: { subagent: { mode: 'continuable', label: 'cached own', seq: 2 } },
})
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: child, label: 'cached own', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -458,7 +543,7 @@ describe('SubagentRuntime.listChildren', () => {
asOfSeq: 2,
values: { subagent: { mode: 'continuable', label: 'ancestor label', seq: 2 } },
})
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: forkChild, label: 'own label', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -481,12 +566,12 @@ describe('SubagentRuntime.listChildren', () => {
parentSession: parent.id,
origin: 'subagent',
}, childEvents(descriptorPayload('reborn child')))
const original = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = async (sessionId, signal) => {
const original = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = async (sessionId, signal) => {
const result = await original(sessionId, signal)
if (sessionId !== reborn) return result
// The id was re-published as a different lifecycle after enumeration.
return { ...result, meta: mutate(result.meta) }
return { ...result, inspection: { ...result.inspection, meta: mutate(result.inspection.meta) } }
}
const entries = await ctx.subagents.listChildren(parent.id)
expect(entries).toContainEqual({ kind: 'diagnostic', id: reborn, reason: 'corrupt' })
@@ -504,7 +589,7 @@ describe('SubagentRuntime.listChildren', () => {
}, childEvents(descriptorPayload('actually valid')))
// A stale cached sentinel must not out-rank the authoritative re-fold.
ctx.sessionProjectionCache.cachedSnapshot = () => ({ asOfSeq: 0, values: { subagent: null } })
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: healthy, label: 'actually valid', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -512,7 +597,7 @@ describe('SubagentRuntime.listChildren', () => {
expect(inspect).toHaveBeenCalledTimes(1)
})
it('maps a child rejected by persistence inspection to unavailable', async () => {
it('maps a child rejected by persistence validation to corrupt', async () => {
const { ctx, parent } = await setup([])
// The surface-eligible user/message lacks its required surfaceOp, so the
// first-party inspection rejects before any projection fold can run.
@@ -530,7 +615,7 @@ describe('SubagentRuntime.listChildren', () => {
{ type: 'subagent/descriptor', seq: 2, time: 3, data: descriptorPayload('broken surface') },
] as SessionEvent[])
const entries = await ctx.subagents.listChildren(parent.id)
expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'unavailable' }])
expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'corrupt' }])
})
it('diagnoses a malformed descriptor payload as corrupt', async () => {
@@ -556,10 +641,10 @@ describe('SubagentRuntime.listChildren', () => {
expect(entries).toEqual([{ kind: 'diagnostic', id: future, reason: 'corrupt' }])
})
it('lists a fork whose seed replays an ancestor descriptor under that identity', async () => {
it('rejects a fork whose only descriptor belongs to its inherited seed', async () => {
const { ctx, parent } = await setup([])
// The last-wins fold serves a seed-replayed ancestor descriptor until the
// child's own descriptor overrides it (known deviation #1 in the design).
// A seed-replayed descriptor predates this child's own suffix and cannot
// identify the fork as a resumable child.
const seed = childEvents(descriptorPayload('ancestor label'))
const forkChild = await authorChild(ctx, '00000000-0000-4000-8000-0000000000f0', {
parentSession: parent.id,
@@ -567,12 +652,7 @@ describe('SubagentRuntime.listChildren', () => {
origin: 'subagent',
}, seed)
const entries = await ctx.subagents.listChildren(parent.id)
expect(entries).toEqual([
{
kind: 'child', id: forkChild, label: 'ancestor label', mode: 'continuable',
activity: 'inactive', hasChildren: false,
},
])
expect(entries).toEqual([{ kind: 'diagnostic', id: forkChild, reason: 'corrupt' }])
})
it('does not filter by provider availability: children of unmounted providers stay listed', async () => {
@@ -613,7 +693,7 @@ describe('SubagentRuntime.listChildren', () => {
})
})
it('contains a foreign unit failure during a live snapshot to that child as corrupt', async () => {
it('does not evaluate an unrelated wire view when exposing a live child identity', async () => {
const { ctx, parent } = await setup([])
ctx.sessionProjections.register(hostileProjectionDefinition)
const poisonedId = SessionId('live-poisoned-child')
@@ -629,7 +709,10 @@ describe('SubagentRuntime.listChildren', () => {
healthy.append('turn/start', { turn: 1 })
healthy.append('subagent/descriptor', descriptorPayload('live healthy'))
const entries = await ctx.subagents.listChildren(parent.id)
expect(entries).toContainEqual({ kind: 'diagnostic', id: poisonedId, reason: 'corrupt' })
expect(entries).toContainEqual({
kind: 'child', id: poisonedId, label: 'poison me', mode: 'continuable',
activity: 'running', hasChildren: false,
})
expect(entries).toContainEqual({
kind: 'child', id: healthyId, label: 'live healthy', mode: 'continuable',
activity: 'running', hasChildren: false,
@@ -652,8 +735,8 @@ describe('SubagentRuntime.listChildren', () => {
parentSession: parent.id,
origin: 'subagent',
}, childEvents(descriptorPayload('flaky storage')))
const original = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = (sessionId, signal) => {
const original = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = (sessionId, signal) => {
if (sessionId === flaky) {
return Promise.reject(new Error('backend read failed'))
}
@@ -669,7 +752,7 @@ describe('SubagentRuntime.listChildren', () => {
})
// Nothing is memoized: with the backend healthy again, the next listing
// folds the same child to its identity.
ctx.sessionPersistence.inspect = original
ctx.sessionPersistence.borrowSession = original
await expect(ctx.subagents.listChildren(parent.id)).resolves.toContainEqual({
kind: 'child', id: flaky, label: 'flaky storage', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -723,8 +806,8 @@ describe('SubagentRuntime.listChildren', () => {
origin: 'subagent',
}, childEvents(descriptorPayload('grandchild')))
const inspected: SessionId[] = []
const original = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = (sessionId, signal) => {
const original = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = (sessionId, signal) => {
inspected.push(sessionId)
return original(sessionId, signal)
}
@@ -755,8 +838,8 @@ describe('SubagentRuntime.listChildren', () => {
live.append('subagent/descriptor', descriptorPayload('live mixed child'))
const inspected: SessionId[] = []
const original = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = (sessionId, signal) => {
const original = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = (sessionId, signal) => {
inspected.push(sessionId)
return original(sessionId, signal)
}
@@ -778,7 +861,7 @@ describe('SubagentRuntime.listChildren', () => {
await vi.waitFor(() => {
expect(ctx.sessionProjectionCache.cachedSnapshot(header!)?.values.subagent).toBeDefined()
}, { timeout: 5_000 })
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: childId, label: 'cached child', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -797,7 +880,7 @@ describe('SubagentRuntime.listChildren', () => {
activity: 'inactive', hasChildren: false,
}]
// No stored row at all for a foreign child this process never ran.
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual(expected)
expect(inspect).toHaveBeenCalledTimes(1)
// A stored row whose cut predates the descriptor: the subagent key is
@@ -814,7 +897,7 @@ describe('SubagentRuntime.listChildren', () => {
parentSession: parent.id,
origin: 'subagent',
}, childEvents(descriptorPayload('uncacheable child')))
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: foreign, label: 'uncacheable child', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -833,7 +916,7 @@ describe('SubagentRuntime.listChildren', () => {
// is derived data, so its failure must not become a verdict.
throw new Error('poisoned cache row')
}
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect')
const inspect = vi.spyOn(ctx.sessionPersistence, 'borrowSession')
await expect(ctx.subagents.listChildren(parent.id)).resolves.toEqual([{
kind: 'child', id: recovered, label: 'recovered child', mode: 'continuable',
activity: 'inactive', hasChildren: false,
@@ -912,7 +995,7 @@ describe('SubagentRuntime.listChildren', () => {
}, childEvents(descriptorPayload('cancelled cold read')))
const controller = new AbortController()
const entered = Promise.withResolvers<undefined>()
ctx.sessionPersistence.inspect = (_sessionId, signal) => {
ctx.sessionPersistence.borrowSession = (_sessionId, signal) => {
entered.resolve(undefined)
return new Promise((_resolve, reject) => {
signal?.addEventListener('abort', () => {
@@ -935,8 +1018,8 @@ describe('SubagentRuntime.listChildren', () => {
origin: 'subagent',
}, childEvents(descriptorPayload('cancelled mid-listing')))
const controller = new AbortController()
const original = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = async (sessionId, signal) => {
const original = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = async (sessionId, signal) => {
const result = await original(sessionId, signal)
controller.abort()
return result
@@ -954,7 +1037,7 @@ describe('SubagentRuntime.listChildren', () => {
origin: 'subagent',
}, childEvents(descriptorPayload('aborted behind a failure')))
const controller = new AbortController()
ctx.sessionPersistence.inspect = () => {
ctx.sessionPersistence.borrowSession = () => {
// The read fails while the caller aborts: cancellation normalization
// must fail the listing rather than return a one-diagnostic success.
controller.abort()
@@ -1188,11 +1271,17 @@ describe('SubagentRuntime.listDescendants', () => {
createdAt: 1,
origin: 'subagent',
}, childEvents(descriptorPayload('lineage checked')))
const realInspect = ctx.sessionPersistence.inspect.bind(ctx.sessionPersistence)
ctx.sessionPersistence.inspect = async (sessionId, signal) => {
const realInspect = ctx.sessionPersistence.borrowSession.bind(ctx.sessionPersistence)
ctx.sessionPersistence.borrowSession = async (sessionId, signal) => {
const inspected = await realInspect(sessionId, signal)
// The exact read reports a different durable parent than enumeration did.
return { ...inspected, meta: { ...inspected.meta, parentSession: SessionId('someone-else') } }
return {
...inspected,
inspection: {
...inspected.inspection,
meta: { ...inspected.inspection.meta, parentSession: SessionId('someone-else') },
},
}
}
await expect(ctx.subagents.listDescendants(parent.id)).resolves.toEqual([
{ kind: 'diagnostic', id: childId, reason: 'corrupt', parentId: parent.id, depth: 1 },
@@ -0,0 +1,14 @@
/** Minimal concrete Session query for tests that exercise only corpus and point reads. */
import SessionQueryEngine from '@deepseek-ai/dsh-session-query'
/** Session query implementation whose search faces are intentionally unavailable. */
export class TestSessionQuery extends SessionQueryEngine {
override searchSessions(): Promise<never> {
return Promise.reject(new Error('session search is not configured in this test'))
}
override searchEvents(): Promise<never> {
return Promise.reject(new Error('event search is not configured in this test'))
}
}
+3
View File
@@ -47,6 +47,9 @@
{
"path": "../../session/session-projection-cache"
},
{
"path": "../../session-query/session-query"
},
{
"path": "../../jobs/jobs"
},
@@ -54,6 +54,7 @@
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-session-projection": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/dsh-subagent-spawn-in-process": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
@@ -17,6 +17,7 @@ import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as tool from '../src/list-agents.ts'
import { parkParent } from './park-parent.ts'
import { TestSessionQuery } from './test-session-query.ts'
/** One scripted response that may wait on a caller-released gate before streaming. */
interface GatedEntry {
@@ -57,6 +58,7 @@ async function setupWith(adapter: MockAdapter | GatedAdapter) {
const root = mkdtempSync(join(tmpdir(), 'dsh-tool-list-agents-'))
roots.push(root)
await ctx.plugin(JsonlSessionPersistence, { root })
await ctx.plugin(TestSessionQuery)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SessionProjectionRegistry)
await ctx.plugin(SubagentRuntime)
@@ -0,0 +1,14 @@
/** Minimal concrete Session query for continuation and catalog integration tests. */
import SessionQueryEngine from '@deepseek-ai/dsh-session-query'
/** Session query implementation whose search faces are outside these tests. */
export class TestSessionQuery extends SessionQueryEngine {
override searchSessions(): Promise<never> {
return Promise.reject(new Error('session search is not configured in this test'))
}
override searchEvents(): Promise<never> {
return Promise.reject(new Error('event search is not configured in this test'))
}
}
@@ -16,6 +16,7 @@ import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
import * as tool from '../src/index.ts'
import { parkParent } from './park-parent.ts'
import { TestSessionQuery } from './test-session-query.ts'
/** One scripted response that may wait on a caller-released gate before streaming. */
interface GatedEntry {
@@ -56,6 +57,7 @@ async function setupWith(adapter: MockAdapter | GatedAdapter) {
const root = mkdtempSync(join(tmpdir(), 'dsh-tool-subagent-control-'))
roots.push(root)
await ctx.plugin(JsonlSessionPersistence, { root })
await ctx.plugin(TestSessionQuery)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SessionProjectionRegistry)
await ctx.plugin(SubagentRuntime)