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

refactor(agent): complete inbox lifecycle migration

Browse Source
This commit is contained in:
_Kerman
2026-08-03 12:25:33 +08:00
parent dc1d542092
commit 49e90695cc
214 changed files with 6019 additions and 4235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
packages/client/runtime/src/client/sessions/conversation.ts
View File

@@ -4,11 +4,12 @@
// string here (narrow to real brands when convenient).
import type { CommandId } from '@deepseek-ai/dsh-commands/brand'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { LlmRetryEventData } from '@deepseek-ai/dsh-llm-retry/types'
import type { TodoItem } from '@deepseek-ai/dsh-session/types'
import type {
MessageId, RpcError, SessionId, ToolCallView, ToolResultView,
RpcError, SessionId, SubagentAddress, ToolCallView, ToolResultView,
} from '@deepseek-ai/dsh-client-connection/client'
import type { PendingInteraction } from './pending.ts'
export type { TodoItem }
@@ -104,6 +105,8 @@ export interface AssistantMessageNode {
/** A steering message injected mid-turn. */
export interface SteeringMessageNode {
kind: 'steering'
/** Stable identity shared with its pre-admission inbox occurrence. */
messageId: MessageId
seq: number
/** Unix epoch ms from the source session event. */
time: number
@@ -122,15 +125,15 @@ export interface ContextMessageNode {
source: unknown
}
/** Durable notice that a failed model request is waiting for another attempt. */
/** Durable notice that a closed failed step is waiting for a model-request retry. */
export type ModelRetryNode = LlmRetryEventData & {
kind: 'model-retry'
seq: number
/** Unix epoch ms from the llm/retry session event. */
time: number
/**
* Client-derived lifecycle: scheduled until another attempt emits retry or
* chunk evidence, started once it does, or cancelled if the turn aborts first.
* Client-derived lifecycle: scheduled until a retry turn starts, started
* once it does, or cancelled when the failed turn aborts first.
*/
retryState: 'scheduled' | 'started' | 'cancelled'
}
@@ -271,9 +274,15 @@ export interface RunningToolCall {
}
/** One independently addressable row from the transient queue snapshot. */
/** One transient inbox occurrence from the authoritative `session/queue` snapshot. */
export interface QueuedMessage {
readonly id: MessageId
/** Stable message identity used for transient-to-durable steering handoff. */
readonly messageId: MessageId
/** Agent-resolved placement; only queued rows accept queue mutations. */
readonly placement: 'queued' | 'steering'
/** Complete content used to render pending steering before it becomes durable. */
readonly content: readonly ContentBlock[]
readonly preview: string
/** Complete editable text; null when the message contains non-text blocks. */
readonly text: string | null
@@ -332,9 +341,14 @@ export interface ConversationSnapshot {
*/
codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
pending: readonly PendingInteraction[]
/** Authoritative transient inbox snapshot, replaced after every host-side change. */
/** Authoritative transient inbox snapshot, including queued and steering placements. */
queue: readonly QueuedMessage[]
running: boolean
/**
* Catalog-discovered continuation address. Its parent availability controls
* human input; null means ordinary session transport.
*/
subagent: { address: SubagentAddress; parentAvailable: boolean } | null
/** Input-area shape (see {@link ComposerPhase}); derived here, switched on by consumers. */
composerPhase: ComposerPhase
/** Set after host/session-removed; the UI grays out and disables input. */

126
packages/client/runtime/src/client/sessions/session.ts
View File

@@ -6,7 +6,7 @@ import type { LlmRetryEventData } from '@deepseek-ai/dsh-llm-retry/types'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type {
HistoryEntry, IApiClient, MessageId, MuxFrame, QueueAction, RpcError,
RpcId, RpcResult, SessionId, ToolEventView,
RpcId, RpcResponse, RpcResult, SessionId, SubagentAddress, ToolEventView,
} from '@deepseek-ai/dsh-client-connection/client'
// Value import from the inline-safe wire layer (not the connection plugin):
// plugin-to-plugin value imports are a bundle purity error.
@@ -34,6 +34,10 @@ const MAX_RETRY_DELAY_MS = 2_147_483_647
/** Manager-owned observers of a Session object's local state edges. */
export interface SessionOptions {
/** Catalog-discovered address selecting non-activating subagent transport. */
address?: SubagentAddress
/** Whether the exact direct parent Agent was live at the latest catalog read. */
parentAvailable?: boolean
/**
* First ACCEPTED prompt on a blank session (fires at most once, on the
* prompt RPC's success response): the manager mirrors the blank→false flip
@@ -119,6 +123,8 @@ export class Session implements SessionFace {
private dispatchesRev = 0
private dispatchesCache: { rev: number; value: ReadonlyMap<string, readonly CodeSubCall[]> } | null = null
private running = false
private address: SubagentAddress | undefined
private parentAvailable = false
/**
* Sticky send marker, private input of the composerPhase derivation: set
* synchronously before prompt()'s first await, never reset — the blank →
@@ -174,6 +180,8 @@ export class Session implements SessionFace {
private readonly options: SessionOptions = {},
) {
this.projections = options.projections ?? new ProjectionValueStore()
this.address = options.address
this.parentAvailable = options.parentAvailable ?? false
this.snapshotCache = this.buildSnapshot()
}
@@ -213,7 +221,21 @@ export class Session implements SessionFace {
this.notifier.markDirty()
let result: RpcResult<{ accepted: true }>
try {
result = (await this.api.sessions.prompt({ sessionId: this.sessionId, mode, content })).result
if (this.address === undefined) {
result = (await this.api.sessions.prompt({ sessionId: this.sessionId, mode, content })).result
} else if (this.address.mode === 'one-shot') {
result = {
ok: false,
error: {
code: 'subagent-not-resumable',
message: 'one-shot subagent conversations are read-only',
details: { childSessionId: this.address.childSessionId },
},
}
} else {
const routed = (await this.api.subagents.prompt({ ...this.address, content })).result
result = routed.ok ? { ok: true, value: { accepted: true } } : routed
}
} catch (error) {
result = transportError(error)
}
@@ -253,6 +275,19 @@ export class Session implements SessionFace {
* @returns the cancel result.
*/
async cancel(): Promise<RpcResult<{ accepted: true }>> {
if (this.address !== undefined) {
const result: RpcResult<{ accepted: true }> = {
ok: false,
error: {
code: 'subagent-delivery-unavailable',
message: 'subagent activation cancellation is unavailable',
details: { childSessionId: this.address.childSessionId },
},
}
this.promptError = { op: 'stop', error: result.error }
this.notifier.markDirty()
return result
}
let result: RpcResult<{ accepted: true }>
try {
result = (await this.api.sessions.cancel({ sessionId: this.sessionId })).result
@@ -318,9 +353,7 @@ export class Session implements SessionFace {
this.loadingOlder = true
this.notifier.markDirty()
try {
const { result } = await this.api.sessions.history({
sessionId: this.sessionId, beforeSeq: this.baseSeq, maxMessages: PAGE_MESSAGES,
})
const { result } = await this.history({ beforeSeq: this.baseSeq, maxMessages: PAGE_MESSAGES })
if (!result.ok) return // keep the window as-is; do not overwrite openError (open already succeeded)
const older = result.value.events
if (older.length === 0) {
@@ -413,8 +446,11 @@ export class Session implements SessionFace {
case 'session/queue': {
this.queued = frame.items.map(item => ({
id: item.id,
preview: queuePreviewOf(item.content),
text: queueTextOf(item.content),
messageId: item.message.id,
placement: item.placement,
content: item.message.content,
preview: queuePreviewOf(item.message.content),
text: queueTextOf(item.message.content),
}))
this.queueRev++
this.notifier.markDirty()
@@ -479,6 +515,32 @@ export class Session implements SessionFace {
this.notifier.markDirty()
}
/**
* Install or clear the catalog-discovered transport address. A changed
* address rebuilds an already-open window through its new history route.
* @param address - direct parent/child address, or undefined for ordinary transport.
* @param parentAvailable - latest exact-parent availability hint.
*/
configureSubagent(address: SubagentAddress | undefined, parentAvailable = false): void {
const same = this.address?.parentSessionId === address?.parentSessionId
&& this.address?.childSessionId === address?.childSessionId
&& this.address?.mode === address?.mode
this.address = address
this.parentAvailable = parentAvailable
if (!same && this.openState !== 'cold') void this.resync()
else this.notifier.markDirty()
}
/**
* Update only the parent availability hint from a catalog refresh.
* @param available - whether the exact direct parent is live.
*/
handleSubagentParentAvailable(available: boolean): void {
if (this.parentAvailable === available) return
this.parentAvailable = available
this.notifier.markDirty()
}
/**
* Blank-bit relay from the authoritative summary source (list baseline and
* the session-added frame). Monotone: once any signal (local first send,
@@ -533,7 +595,7 @@ export class Session implements SessionFace {
this.openError = null
this.notifier.markDirty()
try {
let { result } = await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })
let { result } = await this.history({ maxMessages: PAGE_MESSAGES })
if (generation !== this.openGeneration) return
if (!result.ok) {
this.openState = 'error'
@@ -544,7 +606,7 @@ export class Session implements SessionFace {
// Gap detection (§D.3-4): baseline past the window tail and liveBuffer did not cover it -> pull the tail page once more.
const tailSeq = this.windowTailSeq()
if (this.subscribedLastSeq !== null && tailSeq !== null && this.subscribedLastSeq > tailSeq) {
result = (await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })).result
result = (await this.history({ maxMessages: PAGE_MESSAGES })).result
if (generation !== this.openGeneration) return
if (result.ok) this.installWindow(result.value.events, result.value.hasMore, result.value.projections)
}
@@ -588,9 +650,20 @@ export class Session implements SessionFace {
this.events.push(event)
this.views.push(view)
this.transcript.append(event, view)
this.handoffPendingSteering(event)
this.applyEventSideEffects(event, view)
}
/** Retire the first matching live steering occurrence when its durable event takes over. */
private handoffPendingSteering(event: SessionEvent): void {
if (event.type !== 'steering/message') return
const index = this.queued.findIndex(item =>
item.placement === 'steering' && item.messageId === event.data.message.id)
if (index === -1) return
this.queued = this.queued.filter((_item, candidate) => candidate !== index)
this.queueRev++
}
/** Land a live session/event (open/repair in flight -> buffer; overlapping seq -> drop;
* a seq gap -> buffer + tail-page repull instead of appending a hole (audit S3: a gap is an
* expected reconnect-window artifact, repaired by refetch). The window stays one contiguous
@@ -621,7 +694,7 @@ export class Session implements SessionFace {
this.stitching = true
const generation = this.openGeneration
try {
const { result } = await this.api.sessions.history({ sessionId: this.sessionId, maxMessages: PAGE_MESSAGES })
const { result } = await this.history({ maxMessages: PAGE_MESSAGES })
// Failure or superseded by a full resync: drop — the resync path rebuilds and clears the buffer itself.
if (result.ok && generation === this.openGeneration && this.openState === 'open') {
this.installWindow(result.value.events, result.value.hasMore, result.value.projections)
@@ -646,7 +719,6 @@ export class Session implements SessionFace {
if (this.partial !== null && this.partial.turn === data.turn && this.partial.step === data.step) {
this.partial = null
}
this.settleScheduledRetry('started', data.turn)
this.derivedNodes.push({
kind: 'model-retry',
seq: event.seq,
@@ -720,8 +792,8 @@ export class Session implements SessionFace {
case 'turn/start':
return
case 'assistant/chunk': {
this.settleScheduledRetry('started', event.data.turn)
const { turn, step, chunk } = event.data
this.settleScheduledRetry('started', turn)
if (this.partial === null || this.partial.turn !== turn || this.partial.step !== step) {
this.partial = new PartialAccumulator(turn, step)
}
@@ -748,9 +820,7 @@ export class Session implements SessionFace {
return
}
case 'turn/end': {
if (event.data.reason.kind === 'error') {
this.settleScheduledRetry('started', event.data.turn)
} else if (event.data.reason.kind === 'aborted' || event.data.reason.kind === 'interrupted') {
if (event.data.reason.kind === 'aborted') {
this.settleScheduledRetry('cancelled', event.data.turn)
}
if (
@@ -758,22 +828,20 @@ export class Session implements SessionFace {
&& !this.derivedNodes.some(node => node.kind === 'model-retry' && node.turn === event.data.turn)
) {
const failure = event.data.reason.error
const failedTurn = event.data.turn
const code = failure !== null && typeof failure === 'object'
&& typeof (failure as { code?: unknown }).code === 'string'
? (failure as { code: string }).code
: undefined
const code = failure !== null && typeof failure === 'object' && 'code' in failure
&& typeof failure.code === 'string' ? failure.code : undefined
this.derivedNodes.push({
kind: 'turn-error',
seq: event.seq,
time: event.time,
turn: failedTurn,
turn: event.data.turn,
step: event.data.step,
message: displayFailureMessage(failure),
...code === undefined ? {} : { code },
...(code === undefined ? {} : { code }),
})
this.derivedRev++
}
if (event.data.reason.kind === 'error') this.settleScheduledRetry('started', event.data.turn)
// Aborted turns never finalize. The accumulated partial is VALUE, not residue: freeze it
// into an interrupted terminal node (pulse stops, text survives) instead of deleting it.
// Shared by live and window-replay paths, so a refresh reconstructs the same frozen node
@@ -895,6 +963,9 @@ export class Session implements SessionFace {
codeDispatches: this.dispatchesCache.value,
queue: this.queueCache.value,
running: this.running,
subagent: this.address === undefined
? null
: { address: this.address, parentAvailable: this.parentAvailable },
composerPhase: derivePhase(
// Command lifecycle nodes are not conversation: running /permission
// or /plan on a fresh session keeps the hero (the client mirror of
@@ -912,6 +983,17 @@ export class Session implements SessionFace {
lastAgentError: this.lastAgentError,
}
}
/** Select ordinary or addressed history transport from the stored browser fact. */
private history(payload: { beforeSeq?: number; maxMessages?: number }): Promise<RpcResponse<{
events: HistoryEntry[]
hasMore: boolean
projections?: ProjectionsBaseline
}>> {
return this.address === undefined
? this.api.sessions.history({ sessionId: this.sessionId, ...payload })
: this.api.subagents.history({ ...this.address, ...payload })
}
}
/** Validate the plugin-owned payload at the session-event wire boundary. */

120
packages/client/runtime/tests/queue-store.spec.ts
View File

@@ -4,11 +4,10 @@
* projection, and snapshot reference stability.
*/
import { describe, expect, it } from 'vitest'
import { createUserMessage, freezeMessage } from '@deepseek-ai/dsh-llm'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type {
MessageId, MuxFrame, RpcId, SessionId,
} from '@deepseek-ai/dsh-client-connection/client'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, UserMessage } from '@deepseek-ai/dsh-llm/types'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { MessageId, MuxFrame, RpcId, SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { Session } from '../src/client/sessions/session.ts'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient } from './fake-api.ts'
@@ -16,12 +15,14 @@ import { FakeApiClient } from './fake-api.ts'
const SID = 'fk-q1' as SessionId
const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }]
const rid = (id: string): RpcId => id as RpcId
const mid = (id: string): MessageId => id as MessageId
const iid = (id: string): MessageId => id as MessageId
interface QueueFixture {
id: string
body: string
content?: ContentBlock[]
placement?: 'queued' | 'steering'
message?: UserMessage
}
/** Build one authoritative queue snapshot. */
@@ -29,12 +30,13 @@ function queueFrame(items: QueueFixture[]): MuxFrame {
return {
type: 'session/queue',
sessionId: SID,
items: items.map(item => freezeMessage({
...createUserMessage({
items: items.map(item => ({
id: iid(item.id),
placement: item.placement ?? 'queued',
message: item.message ?? createUserMessage({
content: item.content ?? text(item.body),
source: { kind: 'user', rpcId: rid(`rpc-${item.id}`) } as never,
}),
id: mid(item.id),
})),
}
}
@@ -49,8 +51,14 @@ describe('queue snapshot intake', () => {
session.handleMuxEnvelope(rid('env-1'), queueFrame([
{ id: 'q-1', body: '第一条 排队\n消息' },
]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-1', preview: '第一条 排队 消息', text: '第一条 排队\n消息' },
const queue = session.getSnapshot().queue
expect(typeof queue[0]?.messageId).toBe('string')
expect(queue).toMatchObject([
{
id: 'q-1', placement: 'queued',
content: [{ type: 'text', text: '第一条 排队\n消息' }],
preview: '第一条 排队 消息', text: '第一条 排队\n消息',
},
])
})
@@ -61,8 +69,14 @@ describe('queue snapshot intake', () => {
body: '',
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
}]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-image', preview: 'hi [image]', text: null },
const queue = session.getSnapshot().queue
expect(typeof queue[0]?.messageId).toBe('string')
expect(queue).toMatchObject([
{
id: 'q-image', placement: 'queued',
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' }],
preview: 'hi [image]', text: null,
},
])
})
@@ -85,8 +99,14 @@ describe('queue snapshot intake', () => {
session.handleMuxEnvelope(rid('env-5'), queueFrame([
{ id: 'q-2', body: 'two edited' },
]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-2', preview: 'two edited', text: 'two edited' },
const queue = session.getSnapshot().queue
expect(typeof queue[0]?.messageId).toBe('string')
expect(queue).toMatchObject([
{
id: 'q-2', placement: 'queued',
content: [{ type: 'text', text: 'two edited' }],
preview: 'two edited', text: 'two edited',
},
])
session.handleMuxEnvelope(rid('env-6'), queueFrame([]))
expect(session.getSnapshot().queue).toEqual([])
@@ -99,6 +119,55 @@ describe('queue snapshot intake', () => {
session.handleAgentError('unrelated')
expect(session.getSnapshot().queue).toBe(before)
})
it('retains steering placement and complete content in the same authoritative snapshot', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-steering'), queueFrame([
{ id: 'q-next', body: 'later' },
{ id: 's-now', body: 'interrupt now', placement: 'steering' },
]))
expect(session.getSnapshot().queue.map(item => ({
id: item.id, placement: item.placement, content: item.content,
}))).toEqual([
{ id: 'q-next', placement: 'queued', content: text('later') },
{ id: 's-now', placement: 'steering', content: text('interrupt now') },
])
})
it('hands off exactly one current occurrence when live steering becomes durable', async () => {
const session = makeSession()
await session.open()
const message = createUserMessage({
content: text('same message'),
source: { kind: 'user' },
})
session.handleMuxEnvelope(rid('env-same-id'), queueFrame([
{ id: 's-first', body: '', placement: 'steering', message },
{ id: 's-second', body: '', placement: 'steering', message },
]))
const durable = {
seq: 0,
time: 1_700_000_000_000,
type: 'steering/message',
surfaceOp: 'append',
data: { turn: 1, message },
} as SessionEvent
session.handleMuxEnvelope(rid('env-durable'), {
type: 'session/event', sessionId: SID, event: durable,
})
expect(session.getSnapshot().queue.map(item => item.id)).toEqual(['s-second'])
expect(session.getSnapshot().nodes.filter(node => node.kind === 'steering')).toHaveLength(1)
session.handleMuxEnvelope(rid('env-reused-id'), queueFrame([
{ id: 's-later', body: '', placement: 'steering', message },
]))
session.handleMuxEnvelope(rid('env-replayed-durable'), {
type: 'session/event', sessionId: SID, event: durable,
})
expect(session.getSnapshot().queue.map(item => item.id)).toEqual(['s-later'])
})
})
describe('queue operation transport', () => {
@@ -108,13 +177,22 @@ describe('queue operation transport', () => {
session.handleMuxEnvelope(rid('env-op'), queueFrame([{ id: 'q-op', body: 'pending' }]))
const before = session.getSnapshot().queue
await expect(session.updateQueue(mid('q-op'), { kind: 'edit', content: text('next') }))
await expect(session.updateQueue(iid('q-op'), { kind: 'edit', content: text('next') }))
.resolves.toEqual({ ok: true, value: { accepted: true } })
expect(api.callsOf('session.updateQueue')).toEqual([{
sessionId: SID,
itemId: 'q-op',
action: { kind: 'edit', content: text('next') },
}])
await expect(session.updateQueue(iid('q-op'), { kind: 'steer' }))
.resolves.toEqual({ ok: true, value: { accepted: true } })
expect(api.callsOf('session.updateQueue')).toEqual([
{
sessionId: SID,
itemId: 'q-op',
action: { kind: 'edit', content: text('next') },
},
{
sessionId: SID,
itemId: 'q-op',
action: { kind: 'steer' },
},
])
expect(session.getSnapshot().queue).toBe(before)
})
})