mengzhx/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness.git synced 2026-08-29 04:26:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(history): retain packed records in client

Browse Source
This commit is contained in:
imccyu
2026-08-25 20:13:54 +08:00
parent 20d55b2c41
commit 1ec75c9082
28 changed files with 442 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/cli/tests/github-webhook-real.e2e.ts
+17 -2
View File
@@ -60,7 +60,10 @@ interface WorkspaceBaseline {
}
interface HistoryPage {
records: Array<{ event: HistoryEvent } | { chunks: unknown }>
records: Array<
| { type: 'event'; event: HistoryEvent }
| { type: 'chunks'; event: HistoryChunkEvent }
>
hasMore: boolean
}
@@ -69,6 +72,11 @@ interface HistoryEvent {
data: unknown
}
interface HistoryChunkEvent extends HistoryEvent {
seq: number
time: number
}
interface ProcessObservation {
readonly ready: Promise<string>
readonly text: () => string
@@ -308,7 +316,14 @@ function assistantText(page: HistoryPage): string {
/** Expand lossless history records for assertions over the public event stream. */
function historyEvents(page: HistoryPage): HistoryEvent[] {
return page.records.flatMap(record => 'event' in record ? [record.event] : decodeStorageRecord(record.chunks))
return page.records.flatMap(record => record.type === 'event'
? [record.event]
: decodeStorageRecord({
type: record.event.type.replace(/^chunkrow\//u, ''),
seq0: record.event.seq,
time0: record.event.time,
data: record.event.data,
}))
}
/** Stop the spawned CLI through its normal signal path, escalating only on a stuck teardown. */
apps/web/tests/smoke-real.e2e.ts
+10 -5
View File
@@ -180,7 +180,10 @@ async function sessionCursor(baseUrl: string, sessionId: string): Promise<number
}
interface HistoryPage {
records: ({ event: HistoryEvent } | { chunks: unknown })[]
records: (
| { type: 'event'; event: HistoryEvent }
| { type: 'chunks'; event: HistoryEvent }
)[]
hasMore: boolean
}
@@ -194,7 +197,7 @@ function isRecord(value: unknown): value is Record<string, unknown> {
}
function providerTitle(page: HistoryPage): string | undefined {
const events = page.records.flatMap(record => 'event' in record ? [record.event] : [])
const events = page.records.flatMap(record => record.type === 'event' ? [record.event] : [])
for (let index = events.length - 1; index >= 0; index--) {
const event = events[index] as HistoryEvent
if (event.type !== 'session/title' || !isRecord(event.data)) continue
@@ -208,7 +211,7 @@ function providerTitle(page: HistoryPage): string | undefined {
function hasAssistantMarker(page: HistoryPage, marker: string): boolean {
return page.records.some((record) => {
if (!('event' in record)) return false
if (record.type !== 'event') return false
const { event } = record
if (event.type !== 'assistant/message' || !isRecord(event.data) || !isRecord(event.data.message)) return false
const content = event.data.message.content
@@ -564,9 +567,11 @@ describe('dsh web keyless CLI smoke', () => {
return hasAssistantMarker(page, recoveredMarker)
}, { timeout: 20_000 }).toBe(true)
if (page === undefined) throw new Error('retry history was not observed')
const retry = page.records.find(record => 'event' in record && record.event.type === 'llm/retry')
const retry = page.records.find(record => (
record.type === 'event' && record.event.type === 'llm/retry'
))
expect(mainAttempts).toBe(2)
expect(retry !== undefined && 'event' in retry ? retry.event.data : undefined).toMatchObject({
expect(retry?.type === 'event' ? retry.event.data : undefined).toMatchObject({
turn: 1,
step: 1,
retry: 1,
docs/event-producer-consumer.i18n.yaml
+2 -2
View File
@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write docs/event-producer-consumer.md
event-producer-consumer.md: ce914f90821adc09c59dca15f202831a51655565
event-producer-consumer.zh.md: ca375c98b6c0552f255e5d2604c9ad0c1d968aad
event-producer-consumer.md: b38d919b0586a116e163d8be3eb2c33fcfc6d67d
event-producer-consumer.zh.md: fd36b3ebf40d5467f4b0849d972a7925838dee65
docs/event-producer-consumer.md
+5 -5
View File
@@ -21,11 +21,11 @@ This matrix shows which packages dispatch each harness-owned event and which pac
| `agent/session-start` | `emit` | [`packages/core/agent/src/runtime-types.ts:224`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `agent-team`, [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
| `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:185`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), [`schedule`](../packages/schedule/schedule), `server`, `session-controller` |
| `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:285`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:491`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:471`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:498`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:477`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:484`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:503`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:483`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:510`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:489`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:496`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:85`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes` |
| `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | [`authorization`](../packages/credentials/authorization) |
| `commands/change` | `emit` | [`packages/interaction/commands/src/types.ts:80`](../packages/interaction/commands/src/types.ts) | [`commands`](../packages/interaction/commands) (`events.dispatch`) | `remotes` |
docs/event-producer-consumer.zh.md
+5 -5
View File
@@ -23,11 +23,11 @@
| `agent/session-start` | `emit` | [`packages/core/agent/src/runtime-types.ts:224`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `agent-team`, [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
| `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:185`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`agent`](../packages/core/agent), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), [`schedule`](../packages/schedule/schedule), `server`, `session-controller` |
| `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:285`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex) |
| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:482`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:462`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:489`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:468`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:475`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:503`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:483`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:510`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:489`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:496`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` |
| `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:85`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes` |
| `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | [`authorization`](../packages/credentials/authorization) |
| `commands/change` | `emit` | [`packages/interaction/commands/src/types.ts:80`](../packages/interaction/commands/src/types.ts) | [`commands`](../packages/interaction/commands) (`events.dispatch`) | `remotes` |
packages/api/session-controller/README.i18n.yaml
+2 -2
View File
@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/api/session-controller/README.md
README.md: 24df6e1e367eb76e942087065cf9a29a34e94127
README.zh.md: 710912be75195567b557c7dc103cabb7eed1e80f
README.md: 510f336c76dd80ed01bdd2bd4a364106f418831f
README.zh.md: 4503a8f9d0bcfc7f669f00cbc4db69e0d3efd3cd
packages/api/session-controller/README.md
+2 -2
View File
@@ -4,11 +4,11 @@ English | [中文](README.zh.md)
`@deepseek-ai/dsh-api-session-controller` owns the Host `ctx.sessionController` service and the generated Client `ctx.remote.session` namespace. It serves Session list, search, creation, model selection, rename, fork, prompt, attachment, queue, cancellation, message-aligned history, live log following, and Host-wide control state.
History pages carry `records`: ordinary records contain a raw `SessionWireEvent`, while consecutive same-block `assistant/chunk` deltas use the Session package's lossless packed-row encoding. `SessionEventStream` expands packed rows member-for-member before publishing a page to the Client Session object layer, so replay still observes every original event and sequence number. Follow frames remain raw individual events. Tool arguments, result content, failures, and `tool/result.data.meta` pass through unchanged; the controller does not resolve a Tool definition, run a presenter, or attach UI data.
History pages and follow opening snapshots carry a discriminated `SessionHistoryRecord`. Both variants use `{ type, event }`: `type: 'event'` carries one raw `SessionWireEvent`, while `type: 'chunks'` carries one lossless `ChunkRowEvent` for consecutive same-block `assistant/chunk` deltas. Both inner values expose `type`, `seq`, `time`, and `data`, so the Client retains each accepted record as one `SessionEventLikeEntry` without record-by-record conversion. A packed event's `seq` and `time` identify its first member, and `data` retains the fragment and timestamp-gap arrays. Live follow frames remain individual `event` records. Tool arguments, result content, failures, and `tool/result.data.meta` pass through unchanged; the controller does not resolve a Tool definition, run a presenter, or attach UI data.
Each endpoint states its activation policy. List, search, attachment, history pages, and log following can inspect persistence without activating an Agent; queue mutation and cancellation require the corresponding live state; model, rename, and prompt commands may explicitly resume an ordinary Session. Create and fork are the only operations that create a new Agent. The service applies one preset-aware resume policy and subagent ownership fence to its own methods and to the Typert Agent and Session lookups used by other Remote namespaces.
The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events.
The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events.
## Model Experience
packages/api/session-controller/README.zh.md
+2 -2
View File
@@ -4,11 +4,11 @@
`@deepseek-ai/dsh-api-session-controller` 拥有 Host 的 `ctx.sessionController` 服务和生成的 Client `ctx.remote.session` namespace。它提供 Session 列表、搜索、创建、模型选择、重命名、fork、prompt、附件、queue、取消、按消息对齐的历史、live 日志跟随和 Host 范围 control 状态。
历史页携带 `records`：普通记录包含原始 `SessionWireEvent`，连续且属于同一 block 的 `assistant/chunk` delta 使用 Session 包的无损打包行编码。`SessionEventStream` 会在向 Client Session 对象层发布页面前逐成员展开打包行，因此回放仍能观察到每个原始事件和序号。Follow frame 继续携带单个原始事件。工具参数、结果内容、失败信息和 `tool/result.data.meta` 原样通过；controller 不解析 Tool definition、不运行 presenter，也不附加 UI 数据。
历史页与 follow opening snapshot 携带带判别字段的 `SessionHistoryRecord`。两个分支都使用 `{ type, event }`：`type: 'event'` 携带一个原始 `SessionWireEvent`，`type: 'chunks'` 则携带一个由连续且属于同一 block 的 `assistant/chunk` delta 组成的无损 `ChunkRowEvent`。两种内部值都公开 `type`、`seq`、`time` 与 `data`，因此 Client 无需逐 record 转换，就能把每条已接受 record 保留为一个 `SessionEventLikeEntry`。packed event 的 `seq` 与 `time` 表示首成员，`data` 保留 fragment 与 timestamp-gap 数组。实时 follow frame 继续携带单个 `event` record。工具参数、结果内容、失败信息和 `tool/result.data.meta` 原样通过；controller 不解析 Tool definition、不运行 presenter，也不附加 UI 数据。
每个 endpoint 都声明自己的激活策略。列表、搜索、附件、历史页和日志跟随可以在不激活 Agent 的情况下检查 persistence；queue 变更和取消要求对应 live 状态仍然存在；模型、重命名和 prompt 命令可以显式恢复普通 Session。只有 create 和 fork 会创建新 Agent。该服务把同一套感知 preset 的恢复策略和 subagent ownership fence 同时用于自身方法，以及其他 Remote namespace 使用的 Typert Agent 与 Session lookup。
Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。
Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。
## 模型体验
packages/api/session-controller/src/client/contract/events.ts
+25 -13
View File
@@ -1,10 +1,22 @@
/** Observable contiguous Session event window consumed by domain assemblers. */
import { notifySubscribers, type ObservableSnapshot } from '@deepseek-ai/dsh-client-store'
import type { SessionEventEntry } from '../../types.ts'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { ChunkRowEvent } from '../../types.ts'
/** Standard Session event or compact historical Assistant run. */
export type SessionEventLike = SessionEvent | ChunkRowEvent
/** Client history entry retaining its coarse transport discriminator. */
export type SessionEventLikeEntry =
| { readonly type: 'event'; readonly event: SessionEvent }
| { readonly type: 'chunks'; readonly event: ChunkRowEvent }
/** Scalar live entry accepted by append-only Client paths. */
export type SessionLiveEventEntry = Extract<SessionEventLikeEntry, { readonly type: 'event' }>
interface EventWindowLeaf {
readonly kind: 'leaf'
readonly entries: readonly SessionEventEntry[]
readonly entries: readonly SessionEventLikeEntry[]
readonly length: number
}
@@ -17,7 +29,7 @@ interface EventWindowConcat {
type EventWindowNode = EventWindowLeaf | EventWindowConcat
function leaf(entries: readonly SessionEventEntry[]): EventWindowLeaf {
function leaf(entries: readonly SessionEventLikeEntry[]): EventWindowLeaf {
return { kind: 'leaf', entries, length: entries.length }
}
@@ -25,9 +37,9 @@ function concat(left: EventWindowNode, right: EventWindowNode): EventWindowConca
return { kind: 'concat', left, right, length: left.length + right.length }
}
function materialize(node: EventWindowNode): readonly SessionEventEntry[] {
function materialize(node: EventWindowNode): readonly SessionEventLikeEntry[] {
if (node.kind === 'leaf') return node.entries
const entries = new Array<SessionEventEntry>(node.length)
const entries = new Array<SessionEventLikeEntry>(node.length)
const pending: EventWindowNode[] = [node]
let index = 0
while (pending.length > 0) {
@@ -50,7 +62,7 @@ function windowSnapshot(
revision: number,
change: SessionEventChange,
): SessionEventWindow {
let entries: readonly SessionEventEntry[] | undefined
let entries: readonly SessionEventLikeEntry[] | undefined
return {
get entries() {
entries ??= materialize(node)
@@ -64,13 +76,13 @@ function windowSnapshot(
/** Exact delta that produced the latest event-window revision. */
export type SessionEventChange =
| { readonly kind: 'replace'; readonly entries: readonly SessionEventEntry[] }
| { readonly kind: 'prepend'; readonly entries: readonly SessionEventEntry[] }
| { readonly kind: 'append'; readonly entries: readonly SessionEventEntry[] }
| { readonly kind: 'replace'; readonly entries: readonly SessionEventLikeEntry[] }
| { readonly kind: 'prepend'; readonly entries: readonly SessionEventLikeEntry[] }
| { readonly kind: 'append'; readonly entries: readonly SessionLiveEventEntry[] }
/** Current contiguous event window and its latest synchronous delta. */
export interface SessionEventWindow {
readonly entries: readonly SessionEventEntry[]
readonly entries: readonly SessionEventLikeEntry[]
readonly hasMore: boolean
readonly revision: number
readonly change: SessionEventChange
@@ -108,7 +120,7 @@ export class MutableSessionEventSource implements SessionEventSource {
* @param entries - complete window.
* @param hasMore - whether older history remains.
*/
replace(entries: readonly SessionEventEntry[], hasMore: boolean): void {
replace(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void {
this.window = leaf(entries)
this.publish(hasMore, { kind: 'replace', entries })
}
@@ -118,7 +130,7 @@ export class MutableSessionEventSource implements SessionEventSource {
* @param entries - newly loaded older entries.
* @param hasMore - whether still older history remains.
*/
prepend(entries: readonly SessionEventEntry[], hasMore: boolean): void {
prepend(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void {
this.window = concat(leaf(entries), this.window)
this.publish(hasMore, { kind: 'prepend', entries })
}
@@ -127,7 +139,7 @@ export class MutableSessionEventSource implements SessionEventSource {
* Append one contiguous live entry.
* @param entry - live tail entry.
*/
append(entry: SessionEventEntry): void {
append(entry: SessionLiveEventEntry): void {
const entries = [entry]
this.window = concat(this.window, leaf(entries))
this.publish(this.snapshot.hasMore, {
packages/api/session-controller/src/client/index.ts
+8 -1
View File
@@ -43,7 +43,14 @@ export type {
export type { ISession, ProjectionsFace, SessionFace } from './contract/session.ts'
export type { ISessions } from './contract/sessions.ts'
export { MutableSessionEventSource } from './contract/events.ts'
export type { SessionEventChange, SessionEventSource, SessionEventWindow } from './contract/events.ts'
export type {
SessionEventChange,
SessionEventLike,
SessionEventLikeEntry,
SessionEventSource,
SessionEventWindow,
SessionLiveEventEntry,
} from './contract/events.ts'
export type {
OpenState,
PromptError,
packages/api/session-controller/src/client/sessions/history-records.ts
+31 -13
View File
@@ -1,21 +1,39 @@
/** Lossless client decoding for packed Assistant delta runs in history responses. */
/** Client range access and type narrowing for aligned Session history records. */
import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
import type {
SessionEventEntry,
SessionHistoryRecord,
SessionWireEvent,
} from '../../types.ts'
import type { SessionEventLikeEntry } from '../contract/events.ts'
/**
* Convert history wire records into exact event inputs for Conversation Definitions.
* @param records - validated lossless history transport records.
* @returns Ordinary entries unchanged and packed runs expanded member-for-member.
* Narrow aligned wire records to their Client event types without allocation.
* @param records - validated history transport records.
* @returns the same record array with typed inner events.
*/
export function historyEntries(records: readonly SessionHistoryRecord[]): SessionEventEntry[] {
return records.flatMap(record => 'event' in record
? [record]
: decodeStorageRecord(record.chunks).map(event => ({
event: event as unknown as SessionWireEvent,
})))
export function historyEntries(
records: readonly SessionHistoryRecord[],
): readonly SessionEventLikeEntry[] {
return records as unknown as readonly SessionEventLikeEntry[]
}
/**
* Read the first logical sequence represented by one wire record.
* @param record - validated scalar event or packed Assistant delta run.
* @returns inclusive first Session sequence.
*/
export function historyRecordFirstSeq(record: SessionHistoryRecord): number {
return record.event.seq
}
/**
* Read the final logical sequence represented by one wire record.
* @param record - validated scalar event or packed Assistant delta run.
* @returns inclusive final Session sequence.
*/
export function historyRecordLastSeq(record: SessionHistoryRecord): number {
if (record.type === 'event') return record.event.seq
const length = record.event.type === 'chunkrow/tool-call-chunks'
? record.event.data.args.length
: record.event.data.texts.length
return record.event.seq + length - 1
}
packages/api/session-controller/src/client/sessions/session.ts
+11 -14
View File
@@ -7,7 +7,7 @@ import type {
IApiClient, SubagentAddress,
} from '@deepseek-ai/dsh-client-connection/client'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import {
SessionEventStream,
sessionStreamFailure,
@@ -18,7 +18,6 @@ import type {
QueueAction,
SessionAddress,
SessionControlFrame,
SessionEventEntry,
SessionQueuedItem,
SessionRequestId,
SessionError,
@@ -30,6 +29,9 @@ import type {
OpenState, PromptError, SessionSnapshot,
} from '../contract/snapshot.ts'
import { MutableSessionEventSource } from '../contract/events.ts'
import type {
SessionEventLikeEntry, SessionLiveEventEntry,
} from '../contract/events.ts'
import { Notifier } from './notifier.ts'
import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
import type { SessionRemotes } from './remotes.ts'
@@ -72,7 +74,6 @@ export interface SessionOptions {
*/
export class Session implements SessionFace {
// ---- Window and derived state (all private; the snapshot is the only read API) ----
private eventWindow: SessionEvent[] = []
private baseSeq = 0
private hasMore = false
private openState: OpenState = 'cold'
@@ -400,7 +401,6 @@ export class Session implements SessionFace {
this.openPromise = null
this.openState = 'cold'
this.openError = null
this.eventWindow = []
this.baseSeq = 0
this.notifier.markDirty()
await this.open()
@@ -575,28 +575,25 @@ export class Session implements SessionFace {
}
/** Replace the complete contiguous window and apply page-owned projection metadata. */
private installWindow(entries: readonly SessionEventEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void {
this.eventWindow = entries.map(entry => entry.event as SessionEvent)
this.baseSeq = this.eventWindow[0]?.seq ?? 0
private installWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void {
this.baseSeq = entries[0]?.event.seq ?? 0
this.hasMore = hasMore
if (this.eventWindow.some(event => event.type === 'turn/start')) this.firstPromptPendingTurn = false
if (entries.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
if (projections !== undefined) this.projections.seed(projections)
this.eventSource.replace(entries, hasMore)
this.notifier.markDirty()
}
/** Prepend one stream-validated history page. */
private prependWindow(entries: readonly SessionEventEntry[], hasMore: boolean): void {
this.eventWindow = [...entries.map(entry => entry.event as SessionEvent), ...this.eventWindow]
this.baseSeq = this.eventWindow[0]?.seq ?? 0
private prependWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void {
this.baseSeq = entries[0]?.event.seq ?? this.baseSeq
this.hasMore = hasMore
this.eventSource.prepend(entries, hasMore)
}
/** Append one stream-validated live event. */
private appendLive(entry: SessionEventEntry): boolean {
const event = entry.event as SessionEvent
this.eventWindow.push(event)
private appendLive(entry: SessionLiveEventEntry): boolean {
const event = entry.event
const awaitingFirstTurn = this.firstPromptPendingTurn
if (event.type === 'turn/start') this.firstPromptPendingTurn = false
const queueChanged = this.queueMirror.acceptDurable(event)
packages/api/session-controller/src/client/transport.ts
+41 -10
View File
@@ -14,12 +14,17 @@ import {
import type {
SessionAddress,
SessionControlFrame,
SessionEventEntry,
SessionHistoryRecord,
SessionPage,
SessionPageRequest,
SessionProjectionBaseline,
} from '../types.ts'
import { historyEntries } from './sessions/history-records.ts'
import {
historyEntries,
historyRecordFirstSeq,
historyRecordLastSeq,
} from './sessions/history-records.ts'
import type { SessionEventLikeEntry, SessionLiveEventEntry } from './contract/events.ts'
export {
SESSION_SEARCH_RESULT_LIMIT,
@@ -38,7 +43,33 @@ interface SessionJournalPage extends SessionPage {
}
/** One complete publication from the Session journal stream. */
export type SessionJournalChange = RemoteJournalChange<SessionJournalPage, SessionEventEntry>
export type SessionJournalChange =
| {
readonly type: 'replace' | 'prepend'
readonly page: SessionJournalPage
readonly entries: readonly SessionEventLikeEntry[]
readonly hasMore: boolean
}
| { readonly type: 'append'; readonly entry: SessionLiveEventEntry }
function toSessionJournalChange(
change: RemoteJournalChange<SessionJournalPage, SessionHistoryRecord>,
): SessionJournalChange {
switch (change.type) {
case 'replace':
case 'prepend':
return { ...change, entries: historyEntries(change.entries) }
case 'append': {
if (change.entry.type !== 'event') {
throw new Error('session live stream emitted a packed history record')
}
return {
type: 'append',
entry: change.entry as unknown as SessionLiveEventEntry,
}
}
}
}
type SessionControlBaselineFrame = Extract<SessionControlFrame, { type: 'baseline' }>
type SessionControlDeltaFrame = Exclude<SessionControlFrame, SessionControlBaselineFrame>
@@ -101,7 +132,7 @@ export function createSessionControlStream(
/** Gateway-owned event journal bound to one ordinary or direct-subagent Session address. */
export class SessionEventStream extends RemoteJournalStream<
SessionJournalPage,
SessionEventEntry,
SessionHistoryRecord,
number,
ClientSessionPageRequest
> {
@@ -118,12 +149,13 @@ export class SessionEventStream extends RemoteJournalStream<
super(remote, {
name: 'session event stream',
emptyCursor: -1,
entries: page => historyEntries(page.records),
entries: page => page.records,
hasMore: page => page.hasMore,
cursor: entry => entry.event.seq,
first: historyRecordFirstSeq,
last: historyRecordLastSeq,
compare: (left, right) => left - right,
follows: (left, right) => right === left + 1,
publish: options.publish,
publish: (change) => { options.publish(toSessionJournalChange(change)) },
...(options.carrierFailed === undefined
? {}
: { carrierFailed: options.carrierFailed }),
@@ -135,7 +167,7 @@ export class SessionEventStream extends RemoteJournalStream<
protected override async * follow(
request: ClientSessionPageRequest,
signal: AbortSignal,
): AsyncIterable<RemoteJournalFrame<SessionEventEntry, number, SessionJournalPage>> {
): AsyncIterable<RemoteJournalFrame<SessionHistoryRecord, number, SessionJournalPage>> {
for await (const frame of this.remote.session.follow({
address: this.address,
...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }),
@@ -152,8 +184,7 @@ export class SessionEventStream extends RemoteJournalStream<
}
continue
}
const { type: _type, ...entry } = frame
yield { type: 'entry', entry }
yield { type: 'entry', entry: frame }
}
}
packages/api/session-controller/src/history.ts
+25 -3
View File
@@ -2,13 +2,14 @@
import type { Context } from '@deepseek-ai/cordis'
import { isAppendSurfaceEvent } from '@deepseek-ai/dsh-session'
import { isChunkRow, packChunkRuns } from '@deepseek-ai/dsh-session/chunk-rows'
import { isChunkRow, packChunkRuns, type ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type {} from '@deepseek-ai/dsh-subagent'
import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
import type {
SessionAddress,
SessionChunkRun,
SessionEventEntry,
SessionFollowRequest,
SessionFollowFrame,
@@ -157,7 +158,7 @@ export class SessionHistoryController {
reject('internal', `session event stream skipped seq ${String(nextSeq)}`, {})
}
nextSeq++
yield { type: 'event', ...entryFor(item) }
yield entryFor(item)
}
} finally {
this.closeFollowers.delete(close)
@@ -315,14 +316,35 @@ function paginate(
function entryFor(event: SessionEvent): SessionEventEntry {
return {
type: 'event',
// Session.append validates and freezes event data as JSON before publication.
event: event as unknown as SessionWireEvent,
}
}
function chunkEntryFor(row: ChunkRow): SessionChunkRun {
switch (row.type) {
case 'text-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/text-chunks', seq: row.seq0, time: row.time0, data: row.data },
}
case 'reasoning-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/reasoning-chunks', seq: row.seq0, time: row.time0, data: row.data },
}
case 'tool-call-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/tool-call-chunks', seq: row.seq0, time: row.time0, data: row.data },
}
}
}
/** Encode one bounded logical page without changing its pagination cut. */
function pageRecords(events: readonly SessionEvent[]): SessionHistoryRecord[] {
return packChunkRuns(events).map(record => isChunkRow(record)
? { chunks: record }
? chunkEntryFor(record)
: entryFor(record))
}
packages/api/session-controller/src/types.ts
+14 -2
View File
@@ -362,12 +362,24 @@ export type SessionAddress =
/** One raw Session event in the Remote journal. */
export interface SessionEventEntry {
readonly type: 'event'
readonly event: SessionWireEvent
}
/** Event-shaped wire representation of one packed chunk row. */
export type ChunkRowEvent = {
[Kind in ChunkRow['type']]: {
readonly type: `chunkrow/${Kind}`
readonly seq: number
readonly time: number
readonly data: Extract<ChunkRow, { readonly type: Kind }>['data']
}
}[ChunkRow['type']]
/** One lossless run of consecutive Assistant delta events in a history page. */
export interface SessionChunkRun {
readonly chunks: ChunkRow
readonly type: 'chunks'
readonly event: ChunkRowEvent
}
/** One history-page record: a raw event or a packed Assistant delta run. */
@@ -415,7 +427,7 @@ export type SessionFollowFrame =
readonly hasMore: boolean
readonly projections: SessionProjectionBaseline
}
| ({ readonly type: 'event' } & SessionEventEntry)
| SessionEventEntry
/** One pending inbox occurrence in the authoritative queue snapshot. */
export interface SessionQueuedItem {
packages/api/session-controller/tests/client-contract.client.spec.ts
+7 -6
View File
@@ -1,16 +1,17 @@
import type { SessionEventEntry } from '@deepseek-ai/dsh-api-session-controller/types'
import { describe, expect, it, vi } from 'vitest'
import { MutableSessionEventSource } from '../src/client/contract/events.ts'
import {
MutableSessionEventSource, type SessionLiveEventEntry,
} from '../src/client/contract/events.ts'
import { transportResult } from '../src/client/contract/result.ts'
function entry(seq: number): SessionEventEntry {
function entry(seq: number): SessionLiveEventEntry {
return {
type: 'event',
event: {
type: 'fixture/event',
type: 'turn/start',
seq,
time: seq,
data: { seq },
ignorable: true,
data: { turn: seq },
},
}
}
packages/api/session-controller/tests/event-script.client.ts
+1 -1
View File
@@ -149,7 +149,7 @@ export function plainTurn(startSeq: number, turn: number, ask: string, answer: s
/** Wrap raw events in the journal envelope returned by history. */
export function entries(events: readonly SessionEvent[]): SessionEventEntry[] {
return events.map(event => ({ event: event as unknown as SessionWireEvent }))
return events.map(event => ({ type: 'event', event: event as unknown as SessionWireEvent }))
}
/** Build one view-less history response value. */
packages/api/session-controller/tests/fake-api.client.ts
+6 -6
View File
@@ -28,7 +28,7 @@ import {
} from '@deepseek-ai/dsh-api-gateway/client'
import { RpcId } from '@deepseek-ai/dsh-client-connection/client'
import type { SessionRemotes } from '../src/client/sessions/remotes.ts'
import { historyEntries } from '../src/client/sessions/history-records.ts'
import { historyRecordLastSeq } from '../src/client/sessions/history-records.ts'
const AVAILABLE_STREAM_CONNECTION = {
hostDescription: {
@@ -440,8 +440,8 @@ export class FakeApiClient implements IApiClient {
ok: true,
value: {
...result.value,
records: historyEntries(result.value.records)
.filter(entry => entry.event.seq <= request.throughSeq),
records: result.value.records
.filter(record => historyRecordLastSeq(record) <= request.throughSeq),
},
}
}
@@ -469,8 +469,8 @@ export class FakeApiClient implements IApiClient {
)
}
const page = response.result.value
const entries = historyEntries(page.records)
const cursor = this.followCursor ?? entries.at(-1)?.event.seq ?? -1
const tail = page.records.at(-1)
const cursor = this.followCursor ?? (tail === undefined ? -1 : historyRecordLastSeq(tail))
yield {
type: 'snapshot',
header: {
@@ -482,7 +482,7 @@ export class FakeApiClient implements IApiClient {
: {}),
},
cursor,
records: entries.filter(entry => entry.event.seq <= cursor),
records: page.records.filter(record => historyRecordLastSeq(record) <= cursor),
hasMore: page.hasMore,
projections: page.projections ?? { asOfSeq: cursor, values: {} },
}
packages/api/session-controller/tests/history-records.client.spec.ts
+51 -42
View File
@@ -1,50 +1,60 @@
/** Packed history records decode to the exact Session event stream. */
/** Packed history records become one event-shaped Client value per wire record. */
import { describe, expect, it } from 'vitest'
import { CallId } from '@deepseek-ai/dsh-llm/brand'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { SessionHistoryRecord } from '../src/types.ts'
import { historyEntries } from '../src/client/sessions/history-records.ts'
import {
historyEntries,
historyRecordFirstSeq,
historyRecordLastSeq,
} from '../src/client/sessions/history-records.ts'
describe('historyEntries', () => {
it('keeps ordinary entries and expands every packed text member with its exact boundary', () => {
const ordinary = {
event: { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
} as SessionHistoryRecord
describe('Session history record projection', () => {
it('retains an ordinary event and its point cursor', () => {
const ordinary: SessionHistoryRecord = {
type: 'event',
event: { type: 'turn/start', seq: 7, time: 1, data: { turn: 1 } },
}
const records = [ordinary]
const [entry] = historyEntries(records)
expect(historyEntries(records)).toBe(records)
expect(entry).toBe(ordinary)
expect(historyRecordFirstSeq(ordinary)).toBe(7)
expect(entry?.event.time).toBe(1)
expect(historyRecordLastSeq(ordinary)).toBe(7)
})
it('retains one packed text row without copying or reshaping it', () => {
const packed: SessionHistoryRecord = {
chunks: {
type: 'text-chunks',
seq0: 1,
time0: 2,
data: { turn: 1, step: 1, index: 0, dt: [1, 1, 1], texts: ['a', 'b', 'c', 'd'] },
type: 'chunks',
event: {
type: 'chunkrow/text-chunks',
seq: 11,
time: 20,
data: { turn: 1, step: 2, index: 0, dt: [1, 2, 3], texts: ['a', 'b', 'c', 'd'] },
},
}
const entries = historyEntries([ordinary, packed])
const [entry] = historyEntries([packed])
if (entry?.type !== 'chunks') throw new Error('expected packed history entry')
const { event } = entry
expect(entries).toHaveLength(5)
expect(entries[0]).toBe(ordinary)
expect(entries.slice(1).map((entry) => {
const event = entry.event as unknown as SessionEvent<'assistant/chunk'>
return {
seq: event.seq,
time: event.time,
chunk: event.data.chunk,
}
})).toEqual([
{ seq: 1, time: 2, chunk: { type: 'text-delta', index: 0, text: 'a' } },
{ seq: 2, time: 3, chunk: { type: 'text-delta', index: 0, text: 'b' } },
{ seq: 3, time: 4, chunk: { type: 'text-delta', index: 0, text: 'c' } },
{ seq: 4, time: 5, chunk: { type: 'text-delta', index: 0, text: 'd' } },
])
expect(entry).toBe(packed)
expect(event).toBe(packed.event)
expect(historyRecordFirstSeq(packed)).toBe(11)
expect(event.time).toBe(20)
expect(historyRecordLastSeq(packed)).toBe(14)
})
it('preserves every tool-call fragment and optional-name presence', () => {
it('preserves a packed tool-call row and optional-name absence', () => {
const packed: SessionHistoryRecord = {
chunks: {
type: 'tool-call-chunks',
seq0: 20,
time0: 200,
type: 'chunks',
event: {
type: 'chunkrow/tool-call-chunks',
seq: 20,
time: 200,
data: {
turn: 2,
step: 4,
@@ -56,14 +66,13 @@ describe('historyEntries', () => {
},
}
const events = historyEntries([packed])
.map(entry => entry.event as unknown as SessionEvent<'assistant/chunk'>)
const [entry] = historyEntries([packed])
if (entry?.type !== 'chunks') throw new Error('expected packed history entry')
const { event } = entry
expect(events).toMatchObject([
{ seq: 20, time: 200, data: { chunk: { argumentsDelta: '' } } },
{ seq: 21, time: 202, data: { chunk: { argumentsDelta: '{"x":' } } },
{ seq: 22, time: 205, data: { chunk: { argumentsDelta: '1}' } } },
])
expect(events.every(event => !Object.hasOwn(event.data.chunk, 'name'))).toBe(true)
if (event.type !== 'chunkrow/tool-call-chunks') throw new Error('expected packed history event')
expect(event).toBe(packed.event)
expect(Object.hasOwn(event.data, 'name')).toBe(false)
expect(historyRecordLastSeq(packed)).toBe(22)
})
})
packages/api/session-controller/tests/session-cold.host.spec.ts
+2 -2
View File
@@ -390,7 +390,7 @@ describe('cold history recovery view', () => {
maxMessages: 10,
})
if (!history.ok) throw new Error('history failed')
expect(history.value.records.map(record => 'event' in record ? record.event : record.chunks)).toMatchInlineSnapshot(`
expect(history.value.records.map(record => record.event)).toMatchInlineSnapshot(`
[
{
"data": {
@@ -564,7 +564,7 @@ describe('subagent ownership fence', () => {
},
throughSeq: 3,
}, new AbortController().signal)
expect(history.records.map(record => 'event' in record ? record.event.type : record.chunks.type))
expect(history.records.map(record => record.event.type))
.toEqual(events.map(event => event.type))
expect(ctx.agents.get(sessionId)).toBeUndefined()
packages/api/session-controller/tests/session-history-journal.host.spec.ts
+28 -10
View File
@@ -4,11 +4,12 @@ import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
import { decodeStorageRecord, type ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
import { CallId, createMessage, createToolResultMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
import type {
ChunkRowEvent,
SessionFollowFrame,
SessionPage,
SessionWireEvent,
@@ -84,9 +85,20 @@ async function openFollow(
/** Expand packed page records for assertions over the logical journal. */
function pageEvents(page: SessionPage): SessionWireEvent[] {
return page.records.flatMap(record => 'event' in record
return page.records.flatMap(record => record.type === 'event'
? [record.event]
: decodeStorageRecord(record.chunks).map(event => event as unknown as SessionWireEvent))
: decodeStorageRecord(chunkRow(record.event)).map(event => event as unknown as SessionWireEvent))
}
function chunkRow(event: ChunkRowEvent): ChunkRow {
switch (event.type) {
case 'chunkrow/text-chunks':
return { type: 'text-chunks', seq0: event.seq, time0: event.time, data: event.data }
case 'chunkrow/reasoning-chunks':
return { type: 'reasoning-chunks', seq0: event.seq, time0: event.time, data: event.data }
case 'chunkrow/tool-call-chunks':
return { type: 'tool-call-chunks', seq0: event.seq, time0: event.time, data: event.data }
}
}
describe('Session history raw journal', () => {
@@ -182,9 +194,9 @@ describe('Session history raw journal', () => {
expect(response.ok).toBe(true)
if (!response.ok) throw new Error('unreachable')
expect(response.value.records).toEqual([
{ event: start },
{ event: call },
{ event: result },
{ type: 'event', event: start },
{ type: 'event', event: call },
{ type: 'event', event: result },
])
})
@@ -270,7 +282,7 @@ describe('Session history raw journal', () => {
})
if (!response.ok) throw new Error('unreachable')
expect(pageEvents(response.value).map(event => event.seq)).toEqual([...sources, message.seq])
expect(response.value.records.filter(record => 'chunks' in record)).toHaveLength(1)
expect(response.value.records.filter(record => record.type === 'chunks')).toHaveLength(1)
expect(response.value.hasMore).toBe(true)
} finally {
min.mockRestore()
@@ -286,11 +298,17 @@ describe('Session history raw journal', () => {
const iterator = stream[Symbol.asyncIterator]()
session.append('turn/start', { turn: 1 })
await expect(iterator.next()).resolves.toMatchObject({ value: { event: { type: 'turn/start' } } })
await expect(iterator.next()).resolves.toMatchObject({
value: { type: 'event', event: { type: 'turn/start' } },
})
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-late'), name: 'term', arguments: '{"cmd":"tail"}' })
await expect(iterator.next()).resolves.toMatchObject({ value: { event: { type: 'tool/call' } } })
await expect(iterator.next()).resolves.toMatchObject({
value: { type: 'event', event: { type: 'tool/call' } },
})
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await expect(iterator.next()).resolves.toMatchObject({ value: { event: { type: 'turn/end' } } })
await expect(iterator.next()).resolves.toMatchObject({
value: { type: 'event', event: { type: 'turn/end' } },
})
const events = vi.spyOn(session, 'events', 'get').mockImplementation(() => {
throw new Error('live result rescanned Session history')
})
packages/api/session-controller/tests/session-projections.host.spec.ts
+2 -2
View File
@@ -153,7 +153,7 @@ describe('session.history projections block', () => {
expect(projections.values['test/last-user']).toEqual({ text: 'm2' })
// asOfSeq IS the window tail: the last served event carries it.
const last = records.at(-1)
expect(last !== undefined && 'event' in last ? last.event.seq : last?.chunks.seq0).toBe(projections.asOfSeq)
expect(last?.event.seq).toBe(projections.asOfSeq)
})
it('returns a complete current replacement cut on each follow generation', async () => {
@@ -163,7 +163,7 @@ describe('session.history projections block', () => {
const snapshot = await opening(remote(ctx), session.id)
expect(snapshot.records.map(record => 'event' in record ? record.event.seq : record.chunks.seq0)).toEqual([0, 1])
expect(snapshot.records.map(record => record.event.seq)).toEqual([0, 1])
expect(snapshot.projections.asOfSeq).toBe(1)
expect(snapshot.projections.values).toEqual(
expect.objectContaining({ 'test/last-user': { text: 'm1' } }),
packages/api/session-controller/tests/session.client.spec.ts
+6 -6
View File
@@ -618,23 +618,23 @@ describe('remaining branches', () => {
api.onHistory = () => Promise.resolve(ok({
records: [
...entries(plainTurn(0, 0, 'a', 'b')),
{ event: historyCall },
{ event: historyResult },
{ type: 'event', event: historyCall },
{ type: 'event', event: historyResult },
] as never[],
hasMore: false,
modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
}))
await session.open()
expect(windowEntries(session).slice(-2)).toEqual([
{ event: historyCall },
{ event: historyResult },
{ type: 'event', event: historyCall },
{ type: 'event', event: historyResult },
])
const liveCall = ev.toolCall(8, 2, 'l1', 'write', '{"file_path":"a.ts"}')
await follow(api, liveCall)
expect(windowEntries(session).at(-1)).toEqual({ event: liveCall })
expect(windowEntries(session).at(-1)).toEqual({ type: 'event', event: liveCall })
const liveResult = ev.toolResult(9, 2, 'l1', 'ok')
await follow(api, liveResult)
expect(windowEntries(session).at(-1)).toEqual({ event: liveResult })
expect(windowEntries(session).at(-1)).toEqual({ type: 'event', event: liveResult })
})
})
packages/api/session-controller/tests/transport.client.spec.ts
+50 -9
View File
@@ -19,6 +19,7 @@ import type {
SessionEventEntry,
SessionFollowFrame,
SessionFollowRequest,
SessionHistoryRecord,
SessionPage,
SessionPageRequest,
} from '../src/types.ts'
@@ -36,16 +37,28 @@ const AVAILABLE_CONNECTION = {
}
function entry(seq: number): SessionEventEntry {
return { event: { type: 'turn/start', seq, time: seq, data: { turn: seq } } }
return { type: 'event', event: { type: 'turn/start', seq, time: seq, data: { turn: seq } } }
}
function page(events: readonly SessionEventEntry[], hasMore = false): SessionPage {
return { records: events, hasMore }
function chunks(seq0: number): SessionHistoryRecord {
return {
type: 'chunks',
event: {
type: 'chunkrow/text-chunks',
seq: seq0,
time: seq0,
data: { turn: 1, step: 1, index: 0, texts: ['a', 'b', 'c'], dt: [1, 1] },
},
}
}
function page(records: readonly SessionHistoryRecord[], hasMore = false): SessionPage {
return { records, hasMore }
}
function snapshot(
cursor: number,
events: readonly SessionEventEntry[],
records: readonly SessionHistoryRecord[],
hasMore = false,
): SessionFollowFrame {
return {
@@ -56,7 +69,7 @@ function snapshot(
createdAt: 0,
},
cursor,
records: events,
records,
hasMore,
projections: { asOfSeq: cursor, values: {} },
}
@@ -123,13 +136,41 @@ class ScriptedSessionRemote implements SessionTransportRemote {
}
describe('Session Client stream adapters', () => {
it('validates a packed logical range before publishing one compact Client entry', async () => {
const row = chunks(1)
const remote = new ScriptedSessionRemote(
[{ frames: [snapshot(4, [entry(0), row, entry(4)]), entry(5)], hold: true }],
[],
)
const changes: SessionJournalChange[] = []
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: (change) => { changes.push(change) },
failed: vi.fn(),
})
await stream.open({})
await vi.waitFor(() => { expect(changes).toHaveLength(2) })
expect(changes[0]).toMatchObject({
type: 'replace',
entries: [
entry(0),
row,
entry(4),
],
})
expect(changes[0]?.type === 'replace' ? changes[0].entries[1] : undefined).toBe(row)
expect(changes[1]).toEqual({ type: 'append', entry: entry(5) })
await stream.dispose()
})
it('binds an event journal to one address and publishes replace, append, and prepend changes', async () => {
const remote = new ScriptedSessionRemote(
[{
frames: [
snapshot(3, [entry(2), entry(3)], true),
{ type: 'event', ...entry(3) },
{ type: 'event', ...entry(4) },
entry(3),
entry(4),
],
hold: true,
}],
@@ -165,7 +206,7 @@ describe('Session Client stream adapters', () => {
const remote = new ScriptedSessionRemote(
[
{
frames: [snapshot(1, [entry(0), entry(1)]), { type: 'event', ...entry(2) }],
frames: [snapshot(1, [entry(0), entry(1)]), entry(2)],
terminal: lost,
},
{ frames: [snapshot(4, [entry(0), entry(1), entry(2), entry(3), entry(4)])], hold: true },
@@ -221,7 +262,7 @@ describe('Session Client stream adapters', () => {
it('repairs a live gap without adding an absent message limit', async () => {
const remote = new ScriptedSessionRemote(
[{ frames: [snapshot(0, [entry(0)]), { type: 'event', ...entry(2) }], hold: true }],
[{ frames: [snapshot(0, [entry(0)]), entry(2)], hold: true }],
[{ ok: true, value: page([entry(0), entry(1), entry(2)]) }],
)
const changes: SessionJournalChange[] = []
packages/api/session-controller/tests/transport.host.spec.ts
+21 -8
View File
@@ -87,7 +87,7 @@ describe('SessionHistoryController', () => {
{ address: { kind: 'session', sessionId: session.id }, throughSeq: 1 },
new AbortController().signal,
)
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([0, 1])
expect(page.records.map(entry => entry.event.seq)).toEqual([0, 1])
abort.abort()
expect(await iterator.next()).toMatchObject({ done: true })
@@ -137,7 +137,11 @@ describe('SessionHistoryController', () => {
value: {
type: 'snapshot',
cursor: 2,
records: [{ event: { seq: 0 } }, { event: { seq: 1 } }, { event: { seq: 2 } }],
records: [
{ type: 'event', event: { seq: 0 } },
{ type: 'event', event: { seq: 1 } },
{ type: 'event', event: { seq: 2 } },
],
},
})
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
@@ -200,7 +204,12 @@ describe('SessionHistoryController', () => {
await expect(opening).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot', cursor: 1, records: [{ event: { seq: 0 } }, { event: { seq: 1 } }],
type: 'snapshot',
cursor: 1,
records: [
{ type: 'event', event: { seq: 0 } },
{ type: 'event', event: { seq: 1 } },
],
},
})
expect(attached.id).toBe(sessionId)
@@ -377,7 +386,9 @@ describe('SessionHistoryController', () => {
await expect(transport.page({
address: { kind: 'subagent', parentSessionId, childSessionId, mode: 'continuable' },
throughSeq: 0,
}, signal)).resolves.toMatchObject({ records: [{ event: { type: 'subagent/descriptor' } }] })
}, signal)).resolves.toMatchObject({
records: [{ type: 'event', event: { type: 'subagent/descriptor' } }],
})
await expect(transport.page({
address: {
kind: 'subagent',
@@ -505,7 +516,9 @@ describe('SessionHistoryController', () => {
await expect(ordinaryBench.transport.page({
address: { kind: 'session', sessionId: ordinaryId },
throughSeq: 0,
}, signal())).resolves.toMatchObject({ records: [{ event: { seq: 0 } }] })
}, signal())).resolves.toMatchObject({
records: [{ type: 'event', event: { seq: 0 } }],
})
const parentSessionId = SessionId('cold-parent')
const childSessionId = SessionId('cold-child')
@@ -618,13 +631,13 @@ describe('SessionHistoryController', () => {
const page = await transport.page({
address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, maxMessages: 2,
}, signal())
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0))
expect(page.records.map(entry => entry.event.seq))
.toEqual([3, 4, 5, replacement.seq])
expect(page.hasMore).toBe(true)
const before = await transport.page({
address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, beforeSeq: 3, maxMessages: 1,
}, signal())
expect(before.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([2])
expect(before.records.map(entry => entry.event.seq)).toEqual([2])
})
it('keeps cited source events in the page that owns their appended message', async () => {
@@ -638,7 +651,7 @@ describe('SessionHistoryController', () => {
const page = await transport.page({
address: { kind: 'session', sessionId: session.id }, throughSeq: 1, maxMessages: 1,
}, signal())
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([0, 1])
expect(page.records.map(entry => entry.event.seq)).toEqual([0, 1])
expect(page.hasMore).toBe(false)
})
packages/client/connection/src/client/fixture.ts
+35 -7
View File
@@ -75,11 +75,22 @@ interface FixtureProjectionsBlock {
}
interface FixtureHistoryEntry {
readonly type: 'event'
readonly event: SessionEvent
}
type FixtureChunkRowEvent = {
[Kind in ChunkRow['type']]: {
readonly type: `chunkrow/${Kind}`
readonly seq: number
readonly time: number
readonly data: Extract<ChunkRow, { readonly type: Kind }>['data']
}
}[ChunkRow['type']]
interface FixtureHistoryChunkRun {
readonly chunks: ChunkRow
readonly type: 'chunks'
readonly event: FixtureChunkRowEvent
}
type FixtureHistoryRecord = FixtureHistoryEntry | FixtureHistoryChunkRun
@@ -114,7 +125,7 @@ type FixtureFollowFrame =
readonly hasMore: boolean
readonly projections: FixtureProjectionsBlock
}
| ({ readonly type: 'event' } & FixtureHistoryEntry)
| FixtureHistoryEntry
type FixtureFollowEventFrame = Extract<FixtureFollowFrame, { type: 'event' }>
@@ -1414,9 +1425,26 @@ function pageOf(
break
}
}
const records = packChunkRuns(log.slice(start, end)).map((record): FixtureHistoryRecord => (
isChunkRow(record) ? { chunks: record } : { event: record }
))
const records = packChunkRuns(log.slice(start, end)).map((record): FixtureHistoryRecord => {
if (!isChunkRow(record)) return { type: 'event', event: record }
switch (record.type) {
case 'text-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/text-chunks', seq: record.seq0, time: record.time0, data: record.data },
}
case 'reasoning-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/reasoning-chunks', seq: record.seq0, time: record.time0, data: record.data },
}
case 'tool-call-chunks':
return {
type: 'chunks',
event: { type: 'chunkrow/tool-call-chunks', seq: record.seq0, time: record.time0, data: record.data },
}
}
})
return { records, hasMore: start > 0 }
}
@@ -1884,7 +1912,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
for (const conn of remoteEventConns.values()) conn.push(frame)
}
const emitFollow = (sessionId: SessionId, entry: FixtureHistoryEntry): void => {
for (const conn of followConns.get(sessionId) ?? []) conn.push({ type: 'event', ...entry })
for (const conn of followConns.get(sessionId) ?? []) conn.push(entry)
}
/** OK response echoing the caller's rpcId (contract: responses always backfill, never mint). */
@@ -1942,7 +1970,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
const log = logOf(id)
const event = { seq: log.length, time: Date.now(), ...e } as unknown as SessionEvent
log.push(event)
emitFollow(id, { event })
emitFollow(id, { type: 'event', event })
// Host eager-drive parallel: a unit-advancing event pushes its finished value.
for (const frame of projectionFramesOf(id, log, event)) emitControl(frame)
if (event.type === 'user/message' && event.data.source.kind === 'user') {
packages/client/connection/tests/fixture.client.spec.ts
+26 -4
View File
@@ -37,11 +37,22 @@ interface FixtureSessionSummary {
}
interface FixtureHistoryEntry {
readonly type: 'event'
readonly event: SessionEvent
}
type FixtureChunkRowEvent = {
[Kind in ChunkRow['type']]: {
readonly type: `chunkrow/${Kind}`
readonly seq: number
readonly time: number
readonly data: Extract<ChunkRow, { readonly type: Kind }>['data']
}
}[ChunkRow['type']]
interface FixtureHistoryChunkRun {
readonly chunks: ChunkRow
readonly type: 'chunks'
readonly event: FixtureChunkRowEvent
}
type FixtureHistoryRecord = FixtureHistoryEntry | FixtureHistoryChunkRun
@@ -52,9 +63,20 @@ interface FixturePage {
}
function historyEvents(records: readonly FixtureHistoryRecord[]): SessionEvent[] {
return records.flatMap(record => 'event' in record
return records.flatMap(record => record.type === 'event'
? [record.event]
: decodeStorageRecord(record.chunks))
: decodeStorageRecord(chunkRow(record.event)))
}
function chunkRow(event: FixtureChunkRowEvent): ChunkRow {
switch (event.type) {
case 'chunkrow/text-chunks':
return { type: 'text-chunks', seq0: event.seq, time0: event.time, data: event.data }
case 'chunkrow/reasoning-chunks':
return { type: 'reasoning-chunks', seq0: event.seq, time0: event.time, data: event.data }
case 'chunkrow/tool-call-chunks':
return { type: 'tool-call-chunks', seq0: event.seq, time0: event.time, data: event.data }
}
}
type FixtureFollowFrame =
@@ -68,7 +90,7 @@ type FixtureFollowFrame =
readonly values: Readonly<Record<string, unknown>>
}
}
| ({ readonly type: 'event' } & FixtureHistoryEntry)
| FixtureHistoryEntry
type FixtureControlFrame =
| {
packages/extensions/tool-cordis/src/api-catalog.ts
+7 -3
View File
@@ -3343,6 +3343,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'Branded',
declaration: 'export type Branded<B extends string> = string & {\n readonly [BRAND]: B;\n};',
},
{
name: 'ChunkRowEvent',
declaration: 'export type ChunkRowEvent = {\n [Kind in ChunkRow[\'type\']]: {\n readonly type: `chunkrow/${Kind}`;\n readonly seq: number;\n readonly time: number;\n readonly data: Extract<ChunkRow, {\n readonly type: Kind;\n }>[\'data\'];\n };\n}[ChunkRow[\'type\']];',
},
{
name: 'ClientArtifactBaseline',
declaration: 'export interface ClientArtifactBaseline {\n readonly path: string;\n readonly mtimeMs: number;\n readonly size: number;\n readonly mapMtimeMs: number | null;\n readonly mapSize: number | null;\n}',
@@ -4445,7 +4449,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SessionChunkRun',
declaration: 'export interface SessionChunkRun {\n readonly chunks: ChunkRow;\n}',
declaration: 'export interface SessionChunkRun {\n readonly type: \'chunks\';\n readonly event: ChunkRowEvent;\n}',
},
{
name: 'SessionControlBaseline',
@@ -4477,7 +4481,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SessionEventEntry',
declaration: 'export interface SessionEventEntry {\n readonly event: SessionWireEvent;\n}',
declaration: 'export interface SessionEventEntry {\n readonly type: \'event\';\n readonly event: SessionWireEvent;\n}',
},
{
name: 'SessionEventMap',
@@ -4541,7 +4545,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SessionFollowFrame',
declaration: 'export type SessionFollowFrame = {\n readonly type: \'snapshot\';\n readonly header: SessionHeader;\n readonly cursor: number;\n readonly records: readonly SessionHistoryRecord[];\n readonly hasMore: boolean;\n readonly projections: SessionProjectionBaseline;\n} | ({\n readonly type: \'event\';\n} & SessionEventEntry);',
declaration: 'export type SessionFollowFrame = {\n readonly type: \'snapshot\';\n readonly header: SessionHeader;\n readonly cursor: number;\n readonly records: readonly SessionHistoryRecord[];\n readonly hasMore: boolean;\n readonly projections: SessionProjectionBaseline;\n} | SessionEventEntry;',
},
{
name: 'SessionFollowRequest',