mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-09-11 04:00:38 +00:00
Merge remote-tracking branch 'origin/worktree/session-format-04-live-assistant-stream' into worktree/session-format-05-v1-v2-chunk-migration
# Conflicts: # .agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.i18n.yaml # .agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md # .agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.zh.md # .agents/notes/implemented/simplification/2026-08-06-buffer-free-feedback-telemetry.i18n.yaml # .agents/notes/implemented/simplification/2026-08-06-buffer-free-feedback-telemetry.md # .agents/notes/implemented/simplification/2026-08-06-buffer-free-feedback-telemetry.zh.md # apps/web/tests/scaffold.ts # docs/config-catalog.i18n.yaml # docs/config-catalog.md # docs/config-catalog.zh.md # packages/api/session-controller/README.i18n.yaml # packages/api/session-controller/README.md # packages/api/session-controller/README.zh.md # packages/api/session-controller/src/client/sessions/assistant-stream.ts # packages/core/agent-loop/src/agent.ts # packages/core/agent-loop/src/assistant-stream.ts # packages/core/agent-loop/tests/loop.spec.ts # packages/core/agent/src/runtime-types.ts # packages/session/session-telemetry/tests/telemetry.spec.ts # packages/test-support/llm-replay/README.i18n.yaml # packages/test-support/llm-replay/README.md # packages/test-support/llm-replay/README.zh.md # packages/test-support/llm-replay/src/alpha-refusal-fixtures.ts # packages/test-support/llm-replay/src/index.ts # packages/test-support/llm-replay/tests/llm-replay.spec.ts # packages/test-support/session-snapshot/README.i18n.yaml # packages/test-support/session-snapshot/README.md # packages/test-support/session-snapshot/README.zh.md # packages/test-support/session-snapshot/src/suite.ts # packages/test-support/session-snapshot/tests/suite.spec.ts # snapshots/sdk/sdk.snapshot.ts # snapshots/session/headless.snapshot.ts
This commit is contained in:
Tianyi Cui
@@ -0,0 +1,211 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { LlmAttemptId, createAssistantMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { SessionSeq, type SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type {
|
||||
SessionAssistantStreamBaseline,
|
||||
SessionAssistantStreamFrame,
|
||||
} from '../src/types.ts'
|
||||
import { ClientAssistantStream } from '../src/client/sessions/assistant-stream.ts'
|
||||
import type { SessionLiveEventEntry } from '../src/client/contract/events.ts'
|
||||
|
||||
const ATTEMPT = LlmAttemptId('session:1')
|
||||
|
||||
function entry(event: SessionEvent): SessionLiveEventEntry {
|
||||
return { type: 'event', event }
|
||||
}
|
||||
|
||||
function ordinary(seq: number): SessionLiveEventEntry {
|
||||
return entry({ type: 'turn/start', seq: SessionSeq(seq), time: seq, data: { turn: 1 } })
|
||||
}
|
||||
|
||||
function attemptEvent(seq: number, turn = 1, step = 1): SessionLiveEventEntry {
|
||||
return entry({
|
||||
type: 'assistant/attempt',
|
||||
seq: SessionSeq(seq),
|
||||
time: seq,
|
||||
data: { turn, step, stream: [] },
|
||||
})
|
||||
}
|
||||
|
||||
function messageEvent(
|
||||
seq: number,
|
||||
turn = 1,
|
||||
step = 1,
|
||||
surfaceOp: 'append' | { readonly op: 'replace'; readonly start: number; readonly end: number } = 'append',
|
||||
): SessionLiveEventEntry {
|
||||
return entry({
|
||||
type: 'assistant/message',
|
||||
seq: SessionSeq(seq),
|
||||
time: seq,
|
||||
data: {
|
||||
turn,
|
||||
step,
|
||||
message: createAssistantMessage({
|
||||
content: [{ type: 'text', text: 'done' }],
|
||||
source: { provider: 'mock', model: 'mock' },
|
||||
}),
|
||||
stream: [],
|
||||
},
|
||||
surfaceOp: surfaceOp === 'append'
|
||||
? surfaceOp
|
||||
: { ...surfaceOp, start: SessionSeq(surfaceOp.start), end: SessionSeq(surfaceOp.end) },
|
||||
})
|
||||
}
|
||||
|
||||
function start(
|
||||
attemptId = ATTEMPT,
|
||||
startedAfterSeq = -1,
|
||||
): SessionAssistantStreamFrame {
|
||||
return {
|
||||
type: 'start', attemptId, revision: 1, startedTime: 10,
|
||||
startedAfterSeq: startedAfterSeq === -1 ? -1 : SessionSeq(startedAfterSeq),
|
||||
turn: 1, step: 1,
|
||||
}
|
||||
}
|
||||
|
||||
function chunkFrame(
|
||||
index: number,
|
||||
attemptId = ATTEMPT,
|
||||
): SessionAssistantStreamFrame {
|
||||
return {
|
||||
type: 'chunk', attemptId, revision: index + 2, index, time: 20 + index,
|
||||
chunk: { type: 'text-delta', index: 0, text: `chunk-${index}` },
|
||||
}
|
||||
}
|
||||
|
||||
function end(
|
||||
index: number,
|
||||
outcome: Extract<SessionAssistantStreamFrame, { type: 'end' }>['outcome'],
|
||||
attemptId = ATTEMPT,
|
||||
): SessionAssistantStreamFrame {
|
||||
return { type: 'end', attemptId, revision: index + 2, index, outcome }
|
||||
}
|
||||
|
||||
function baseline(nextIndex = 1): SessionAssistantStreamBaseline {
|
||||
return {
|
||||
revision: nextIndex + 1,
|
||||
activeAttempt: {
|
||||
attemptId: ATTEMPT,
|
||||
startedTime: 10,
|
||||
startedAfterSeq: -1,
|
||||
turn: 1,
|
||||
step: 1,
|
||||
nextIndex,
|
||||
stream: [
|
||||
{ type: 'chunk', time: 20, chunk: { type: 'text-delta', index: 0, text: 'first' } },
|
||||
{ type: 'chunk', time: 21, chunk: { type: 'text-delta', index: 0, text: 'second' } },
|
||||
],
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function opened(): ClientAssistantStream {
|
||||
const stream = new ClientAssistantStream()
|
||||
expect(stream.acceptFrame(start())).toBeUndefined()
|
||||
return stream
|
||||
}
|
||||
|
||||
describe('ClientAssistantStream', () => {
|
||||
it('replaces the durable window and reconstructs only the baseline prefix', () => {
|
||||
const stream = new ClientAssistantStream()
|
||||
const durable = ordinary(4)
|
||||
const visible = stream.replace([durable], baseline(1))
|
||||
|
||||
expect(visible[0]).toBe(durable)
|
||||
expect(visible.slice(1)).toEqual([expect.objectContaining({
|
||||
type: 'transient',
|
||||
event: expect.objectContaining({
|
||||
type: 'assistant/live-chunk',
|
||||
seq: 4.5,
|
||||
time: 20,
|
||||
}),
|
||||
})])
|
||||
|
||||
expect(stream.replace([], baseline(3))).toHaveLength(2)
|
||||
expect(stream.replace([])).toEqual([])
|
||||
})
|
||||
|
||||
it('passes through durable events not owned by the active attempt', () => {
|
||||
const stream = new ClientAssistantStream()
|
||||
stream.acceptFrame(start(ATTEMPT, 1))
|
||||
for (const durable of [
|
||||
ordinary(1),
|
||||
messageEvent(2, 1, 1, { op: 'replace', start: 0, end: 0 }),
|
||||
attemptEvent(0),
|
||||
attemptEvent(3, 2, 1),
|
||||
attemptEvent(4, 1, 2),
|
||||
]) {
|
||||
expect(stream.acceptDurable(durable)).toEqual({ type: 'publish', entry: durable })
|
||||
}
|
||||
})
|
||||
|
||||
it('stages one owned settlement and releases it from the matching end frame', () => {
|
||||
const stream = opened()
|
||||
const durable = messageEvent(2)
|
||||
expect(stream.acceptDurable(durable)).toBeUndefined()
|
||||
expect(stream.acceptFrame(chunkFrame(0))).toEqual(expect.objectContaining({ type: 'transient' }))
|
||||
expect(stream.acceptFrame(end(1, {
|
||||
kind: 'committed', eventType: 'assistant/message', seq: 2,
|
||||
}))).toEqual({ type: 'settlement', attemptId: String(ATTEMPT), entry: durable })
|
||||
})
|
||||
|
||||
it('rebaselines duplicate durable settlements or starts', () => {
|
||||
const duplicate = opened()
|
||||
const durable = attemptEvent(2)
|
||||
expect(duplicate.acceptDurable(durable)).toBeUndefined()
|
||||
expect(duplicate.acceptDurable(durable)).toEqual({ type: 'rebaseline' })
|
||||
expect(duplicate.acceptFrame(start(LlmAttemptId('session:2')))).toEqual({ type: 'rebaseline' })
|
||||
|
||||
const clean = new ClientAssistantStream()
|
||||
expect(clean.acceptFrame(start())).toBeUndefined()
|
||||
})
|
||||
|
||||
it('falls back to durable settlement for frames from an unknown attempt', () => {
|
||||
const stream = new ClientAssistantStream()
|
||||
const unknown = LlmAttemptId('session:unknown')
|
||||
expect(stream.acceptFrame(chunkFrame(0, unknown))).toBeUndefined()
|
||||
expect(stream.acceptFrame(end(0, { kind: 'abandoned' }, unknown))).toBeUndefined()
|
||||
const durable = attemptEvent(2)
|
||||
expect(stream.acceptDurable(durable)).toEqual({ type: 'publish', entry: durable })
|
||||
|
||||
const known = opened()
|
||||
expect(known.acceptFrame(chunkFrame(0, unknown))).toBeUndefined()
|
||||
expect(known.acceptFrame(end(0, { kind: 'abandoned' }, unknown))).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rebaselines known attempts on chunk or terminal index mismatch', () => {
|
||||
const chunkMismatch = opened()
|
||||
expect(chunkMismatch.acceptFrame(chunkFrame(1))).toEqual({ type: 'rebaseline' })
|
||||
|
||||
const endMismatch = opened()
|
||||
expect(endMismatch.acceptFrame(end(1, { kind: 'abandoned' }))).toEqual({ type: 'rebaseline' })
|
||||
})
|
||||
|
||||
it('settles abandonment only when no durable settlement remains pending', () => {
|
||||
const empty = opened()
|
||||
expect(empty.acceptFrame(end(0, { kind: 'abandoned' }))).toBeUndefined()
|
||||
|
||||
const pending = opened()
|
||||
expect(pending.acceptDurable(attemptEvent(2))).toBeUndefined()
|
||||
expect(pending.acceptFrame(end(0, { kind: 'abandoned' }))).toEqual({ type: 'rebaseline' })
|
||||
})
|
||||
|
||||
it('rebaselines committed outcomes without one exact staged settlement', () => {
|
||||
const published = new ClientAssistantStream()
|
||||
published.replace([attemptEvent(2)], baseline(0))
|
||||
expect(published.acceptFrame(end(0, {
|
||||
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
|
||||
}))).toBeUndefined()
|
||||
|
||||
const missing = opened()
|
||||
expect(missing.acceptFrame(end(0, {
|
||||
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
|
||||
}))).toEqual({ type: 'rebaseline' })
|
||||
|
||||
const wrongType = opened()
|
||||
expect(wrongType.acceptDurable(messageEvent(2))).toBeUndefined()
|
||||
expect(wrongType.acceptFrame(end(0, {
|
||||
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
|
||||
}))).toEqual({ type: 'rebaseline' })
|
||||
})
|
||||
})
|
||||
@@ -90,7 +90,7 @@ describe('sessions.list cold merge', () => {
|
||||
ctx.provide('sessionProjectionCache', {
|
||||
cachedSnapshot: () => undefined,
|
||||
cachedPredecessorTitle: (meta: SessionHeader) => meta.id === sid('legacy-title')
|
||||
? { asOfSeq: 3, values: { title: 'Cached predecessor title' } }
|
||||
? { asOfSeq: -1, values: { title: 'Cached predecessor title' } }
|
||||
: undefined,
|
||||
} as never)
|
||||
const remote = createSessionTestRemote(ctx, {
|
||||
@@ -112,7 +112,7 @@ describe('sessions.list cold merge', () => {
|
||||
sessionId: sid('legacy-title'),
|
||||
blank: false,
|
||||
updatedAt: 100,
|
||||
projections: { asOfSeq: 3, values: { title: 'Cached predecessor title' } },
|
||||
projections: { asOfSeq: -1, values: { title: 'Cached predecessor title' } },
|
||||
}),
|
||||
])
|
||||
expect(stat).not.toHaveBeenCalled()
|
||||
@@ -172,51 +172,6 @@ describe('sessions.list cold merge', () => {
|
||||
expect(inspect).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('prefers a live row attached during cache lookup without folding its seed', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const meta = header('attached-during-list', 100)
|
||||
providePersistence(ctx, {
|
||||
list: () => Promise.resolve([meta]),
|
||||
})
|
||||
const cacheLookup = vi.fn(() => {
|
||||
const session = ctx.sessions.create(meta.id, {
|
||||
seed: [
|
||||
{ type: 'turn/start', seq: SessionSeq(0), time: 200, data: { turn: 1 } },
|
||||
{
|
||||
type: 'user/message', seq: SessionSeq(1), time: 300,
|
||||
data: createUserMessage({ content: [{ type: 'text', text: 'live' }], source: { kind: 'user' } }),
|
||||
surfaceOp: 'append',
|
||||
},
|
||||
],
|
||||
meta: {
|
||||
...meta.cwd === undefined ? {} : { cwd: meta.cwd },
|
||||
createdAt: meta.createdAt,
|
||||
},
|
||||
})
|
||||
ctx.agents.register({ id: session.id, session, status: 'running', ctx } as Agent)
|
||||
return undefined
|
||||
})
|
||||
ctx.provide('sessionProjectionCache', {
|
||||
cachedSnapshot: cacheLookup,
|
||||
cachedPredecessorTitle: () => undefined,
|
||||
} as never)
|
||||
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
|
||||
|
||||
const response = await remote.list(request({}))
|
||||
if (!response.ok) throw new Error('list failed')
|
||||
expect(response.value.items).toEqual([
|
||||
expect.objectContaining({
|
||||
sessionId: meta.id,
|
||||
blank: false,
|
||||
running: true,
|
||||
updatedAt: 100,
|
||||
}),
|
||||
])
|
||||
expect(cacheLookup).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
describe('attached updatedAt tracks human prompts', () => {
|
||||
|
||||
Reference in New Issue
Block a user