mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-09-14 04:01:35 +00:00
Merge remote-tracking branch 'origin/master' into dshw/pr-deepseek-harness-deepseek-harness-2672
# Conflicts: # docs/event-producer-consumer.i18n.yaml # docs/event-producer-consumer.md # docs/event-producer-consumer.zh.md # docs/module-graph.i18n.yaml # docs/module-graph.md # docs/module-graph.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/tests/session-projections.host.spec.ts # packages/bundle/headless/tests/headless.spec.ts # packages/core/agent-loop/README.i18n.yaml # packages/core/agent-loop/README.md # packages/core/agent-loop/README.zh.md # packages/core/agent/src/runtime-types.ts # packages/fs/tool-str-replace-editor/tests/tools.spec.ts # packages/llm/llm-retry/tests/retry.spec.ts # packages/llm/llm-retry/tests/transport-recovery.spec.ts # packages/shell/tool-bash-persistent/tests/loader-composition.spec.ts # packages/shell/tool-bash-persistent/tests/tools.spec.ts # packages/shell/tool-pwsh-persistent/tests/loader-composition.spec.ts # packages/shell/tool-pwsh-persistent/tests/tools.spec.ts # packages/terminal/terminal-bash/tests/index.spec.ts # packages/test-support/agent-loop-testkit/package.json
This commit is contained in:
@@ -0,0 +1,102 @@
|
||||
/** Process-local assistant state retained for reconnecting Web followers. */
|
||||
|
||||
import type { AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
|
||||
import { AssistantStreamAccumulator } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionSeqCursor } from '@deepseek-ai/dsh-session'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import type {
|
||||
SessionAssistantStreamAttempt,
|
||||
SessionAssistantStreamBaseline,
|
||||
} from './types.ts'
|
||||
|
||||
interface MutableAttempt {
|
||||
readonly attemptId: SessionAssistantStreamAttempt['attemptId']
|
||||
readonly startedAfterSeq: SessionSeqCursor
|
||||
readonly turn: number
|
||||
readonly step: number
|
||||
readonly stream: AssistantStreamAccumulator
|
||||
nextIndex: number
|
||||
}
|
||||
|
||||
const EMPTY_BASELINE: SessionAssistantStreamBaseline = { revision: 0 }
|
||||
|
||||
/**
|
||||
* Folds dense Agent frames and materializes one shared immutable reconnect
|
||||
* baseline per accepted revision.
|
||||
*/
|
||||
export class SessionAssistantStreamAccumulator {
|
||||
private activeAttempt: MutableAttempt | undefined
|
||||
private revision = 0
|
||||
private snapshotValue: SessionAssistantStreamBaseline = EMPTY_BASELINE
|
||||
private dirty = false
|
||||
|
||||
/**
|
||||
* Fold one trusted frame from the current attached Agent lifecycle.
|
||||
* @param frame - next dense process-local Assistant frame.
|
||||
* @param durableCursor - last committed Session seq when this frame was observed.
|
||||
*/
|
||||
accept(frame: AssistantStreamFrame, durableCursor: SessionSeqCursor): void {
|
||||
if (frame.type === 'start' && frame.revision === 1 && this.revision !== 0) {
|
||||
this.activeAttempt = undefined
|
||||
this.revision = 0
|
||||
}
|
||||
if (frame.revision !== this.revision + 1) {
|
||||
this.activeAttempt = undefined
|
||||
this.revision = frame.revision
|
||||
this.dirty = true
|
||||
return
|
||||
}
|
||||
this.revision = frame.revision
|
||||
switch (frame.type) {
|
||||
case 'start':
|
||||
this.activeAttempt = {
|
||||
attemptId: frame.attemptId,
|
||||
startedAfterSeq: durableCursor,
|
||||
turn: frame.turn,
|
||||
step: frame.step,
|
||||
stream: new AssistantStreamAccumulator(),
|
||||
nextIndex: 0,
|
||||
}
|
||||
break
|
||||
case 'chunk': {
|
||||
const attempt = this.activeAttempt
|
||||
if (attempt === undefined
|
||||
|| attempt.attemptId !== frame.attemptId
|
||||
|| frame.index !== attempt.nextIndex) {
|
||||
this.activeAttempt = undefined
|
||||
break
|
||||
}
|
||||
attempt.stream.push({ time: frame.time, chunk: frame.chunk })
|
||||
attempt.nextIndex += 1
|
||||
break
|
||||
}
|
||||
case 'end':
|
||||
this.activeAttempt = undefined
|
||||
break
|
||||
}
|
||||
this.dirty = true
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the cached reconnect baseline, materializing it after a state change.
|
||||
* @returns the identity-stable baseline for the latest accepted revision.
|
||||
*/
|
||||
snapshot(): SessionAssistantStreamBaseline {
|
||||
if (!this.dirty) return this.snapshotValue
|
||||
this.snapshotValue = {
|
||||
revision: this.revision,
|
||||
...this.activeAttempt === undefined ? {} : {
|
||||
activeAttempt: {
|
||||
attemptId: this.activeAttempt.attemptId,
|
||||
startedAfterSeq: this.activeAttempt.startedAfterSeq,
|
||||
turn: this.activeAttempt.turn,
|
||||
step: this.activeAttempt.step,
|
||||
nextIndex: this.activeAttempt.nextIndex,
|
||||
stream: this.activeAttempt.stream.snapshot() as unknown as readonly JsonValue[],
|
||||
},
|
||||
},
|
||||
}
|
||||
this.dirty = false
|
||||
return this.snapshotValue
|
||||
}
|
||||
}
|
||||
@@ -1,18 +1,38 @@
|
||||
/** Observable contiguous Session event window consumed by domain assemblers. */
|
||||
import { notifySubscribers, type ObservableSnapshot } from '@deepseek-ai/dsh-client-store'
|
||||
import type { LlmAttemptId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
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-only live chunk presentation; `seq` orders the transient row between durable Session seqs. */
|
||||
export interface AssistantLiveChunkEvent {
|
||||
readonly type: 'assistant/live-chunk'
|
||||
readonly seq: number
|
||||
readonly time: number
|
||||
readonly data: {
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly turn: number
|
||||
readonly step: number
|
||||
readonly chunk: StreamChunk
|
||||
}
|
||||
}
|
||||
|
||||
/** Current durable Session event or one client-only live chunk presentation. */
|
||||
export type SessionEventLike = SessionEvent | AssistantLiveChunkEvent
|
||||
|
||||
/** Client history entry retaining its coarse transport discriminator. */
|
||||
export type SessionEventLikeEntry =
|
||||
| { readonly type: 'event'; readonly event: SessionEvent }
|
||||
| { readonly type: 'chunks'; readonly event: ChunkRowEvent }
|
||||
| { readonly type: 'transient'; readonly event: AssistantLiveChunkEvent }
|
||||
|
||||
/** Scalar live entry accepted by append-only Client paths. */
|
||||
export type SessionLiveEventEntry = Extract<SessionEventLikeEntry, { readonly type: 'event' }>
|
||||
/** Durable Assistant event that atomically supersedes one attempt's transient rows. */
|
||||
export interface SessionAssistantSettlementEntry {
|
||||
readonly type: 'event'
|
||||
readonly event: SessionEvent<'assistant/message'> | SessionEvent<'assistant/attempt'>
|
||||
}
|
||||
/** Client-only Assistant frame admitted outside durable cursor algebra. */
|
||||
export type SessionTransientEventEntry = Extract<SessionEventLikeEntry, { readonly type: 'transient' }>
|
||||
|
||||
interface EventWindowLeaf {
|
||||
readonly kind: 'leaf'
|
||||
@@ -78,7 +98,12 @@ function windowSnapshot(
|
||||
export type SessionEventChange =
|
||||
| { readonly kind: 'replace'; readonly entries: readonly SessionEventLikeEntry[] }
|
||||
| { readonly kind: 'prepend'; readonly entries: readonly SessionEventLikeEntry[] }
|
||||
| { readonly kind: 'append'; readonly entries: readonly SessionLiveEventEntry[] }
|
||||
| { readonly kind: 'append'; readonly entries: readonly SessionEventLikeEntry[] }
|
||||
| {
|
||||
readonly kind: 'settle-assistant'
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly entry?: SessionAssistantSettlementEntry
|
||||
}
|
||||
|
||||
/** Current contiguous event window and its latest synchronous delta. */
|
||||
export interface SessionEventWindow {
|
||||
@@ -139,7 +164,7 @@ export class MutableSessionEventSource implements SessionEventSource {
|
||||
* Append one contiguous live entry.
|
||||
* @param entry - live tail entry.
|
||||
*/
|
||||
append(entry: SessionLiveEventEntry): void {
|
||||
append(entry: SessionEventLikeEntry): void {
|
||||
const entries = [entry]
|
||||
this.window = concat(this.window, leaf(entries))
|
||||
this.publish(this.snapshot.hasMore, {
|
||||
@@ -148,6 +173,28 @@ export class MutableSessionEventSource implements SessionEventSource {
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace one attempt's transient rows with its committed durable settlement.
|
||||
* @param attemptId - process-local attempt whose live rows are now redundant.
|
||||
* @param entry - durable settlement committed for that attempt.
|
||||
*/
|
||||
settleAssistant(attemptId: LlmAttemptId, entry?: SessionAssistantSettlementEntry): void {
|
||||
const entries = materialize(this.window).filter(candidate => (
|
||||
candidate.type !== 'transient' || candidate.event.data.attemptId !== attemptId
|
||||
))
|
||||
if (entry !== undefined) {
|
||||
const index = entries.findIndex(candidate => candidate.event.seq > entry.event.seq)
|
||||
if (index < 0) entries.push(entry)
|
||||
else entries.splice(index, 0, entry)
|
||||
}
|
||||
this.window = leaf(entries)
|
||||
this.publish(this.snapshot.hasMore, {
|
||||
kind: 'settle-assistant',
|
||||
attemptId,
|
||||
...(entry === undefined ? {} : { entry }),
|
||||
})
|
||||
}
|
||||
|
||||
private publish(
|
||||
hasMore: boolean,
|
||||
change: SessionEventChange,
|
||||
|
||||
@@ -49,12 +49,15 @@ export type {
|
||||
export type { ISessions } from './contract/sessions.ts'
|
||||
export { MutableSessionEventSource } from './contract/events.ts'
|
||||
export type {
|
||||
AssistantLiveChunkEvent,
|
||||
SessionAssistantSettlementEntry,
|
||||
SessionEventChange,
|
||||
SessionEventLike,
|
||||
SessionEventLikeEntry,
|
||||
SessionEventSource,
|
||||
SessionEventWindow,
|
||||
SessionLiveEventEntry,
|
||||
SessionTransientEventEntry,
|
||||
} from './contract/events.ts'
|
||||
export type {
|
||||
OpenState,
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
/** Web presentation fold joining transient Assistant frames to one durable v2 settlement. */
|
||||
|
||||
import type {
|
||||
SessionAssistantStreamBaseline,
|
||||
SessionAssistantStreamFrame,
|
||||
} from '../../types.ts'
|
||||
import { expandAssistantStream } from '@deepseek-ai/dsh-llm/assistant-stream'
|
||||
import type { AssistantStreamRecord } from '@deepseek-ai/dsh-llm/assistant-stream'
|
||||
import type { LlmAttemptId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type {
|
||||
SessionAssistantSettlementEntry,
|
||||
SessionEventLikeEntry,
|
||||
SessionLiveEventEntry,
|
||||
SessionTransientEventEntry,
|
||||
} from '../contract/events.ts'
|
||||
|
||||
interface ActiveAttempt {
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly startedAfterSeq: number
|
||||
readonly turn: number
|
||||
readonly step: number
|
||||
nextIndex: number
|
||||
}
|
||||
|
||||
/** One Web publication decision from the assistant stream fold. */
|
||||
export type ClientAssistantStreamResult =
|
||||
| { readonly type: 'publish'; readonly entry: SessionLiveEventEntry }
|
||||
| {
|
||||
readonly type: 'settlement'
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly entry: SessionAssistantSettlementEntry
|
||||
}
|
||||
| { readonly type: 'abandonment'; readonly attemptId: LlmAttemptId }
|
||||
| { readonly type: 'transient'; readonly entry: SessionTransientEventEntry }
|
||||
| { readonly type: 'rebaseline' }
|
||||
| undefined
|
||||
|
||||
/** Keeps transient Assistant presentation behind one settlement-aware interface. */
|
||||
export class ClientAssistantStream {
|
||||
private activeAttempt: ActiveAttempt | undefined
|
||||
private readonly pending = new Map<number, SessionAssistantSettlementEntry>()
|
||||
private publishedSeqs = new Set<number>()
|
||||
private durableCursor = -1
|
||||
private transientInGap = 0
|
||||
|
||||
/**
|
||||
* Replace the durable Web window and adopt an optional reconnect baseline.
|
||||
* @param entries - durable entries in the replacement window.
|
||||
* @param baseline - compact prefix for an Assistant attempt that is still live.
|
||||
* @returns immediately visible durable entries plus reconstructed transient chunks.
|
||||
*/
|
||||
replace(
|
||||
entries: readonly SessionEventLikeEntry[],
|
||||
baseline?: SessionAssistantStreamBaseline,
|
||||
): readonly SessionEventLikeEntry[] {
|
||||
this.pending.clear()
|
||||
this.transientInGap = 0
|
||||
this.activeAttempt = undefined
|
||||
const opening = baseline?.activeAttempt
|
||||
if (opening !== undefined) {
|
||||
this.activeAttempt = {
|
||||
attemptId: opening.attemptId,
|
||||
startedAfterSeq: opening.startedAfterSeq,
|
||||
turn: opening.turn,
|
||||
step: opening.step,
|
||||
nextIndex: opening.nextIndex,
|
||||
}
|
||||
}
|
||||
const visible: SessionEventLikeEntry[] = [...entries]
|
||||
this.publishedSeqs = new Set(visible.map(entry => entry.event.seq))
|
||||
this.durableCursor = visible.reduce((cursor, entry) => Math.max(cursor, entry.event.seq), -1)
|
||||
if (opening !== undefined) {
|
||||
for (const [index, member] of expandAssistantStream(
|
||||
opening.stream as unknown as readonly AssistantStreamRecord[],
|
||||
).entries()) {
|
||||
this.transientInGap += 1
|
||||
visible.push({
|
||||
type: 'transient',
|
||||
event: {
|
||||
type: 'assistant/live-chunk',
|
||||
seq: this.durableCursor + 1 - 1 / (this.transientInGap + 1),
|
||||
time: member.time,
|
||||
data: {
|
||||
attemptId: opening.attemptId,
|
||||
turn: opening.turn,
|
||||
step: opening.step,
|
||||
chunk: member.chunk,
|
||||
},
|
||||
},
|
||||
})
|
||||
if (index + 1 >= opening.nextIndex) break
|
||||
}
|
||||
}
|
||||
return visible
|
||||
}
|
||||
|
||||
/**
|
||||
* Stage one durable v2 settlement while its matching live attempt is open.
|
||||
* @param entry - newly followed durable entry.
|
||||
* @returns a publication decision, or `undefined` when no entry becomes visible.
|
||||
*/
|
||||
acceptDurable(entry: SessionLiveEventEntry): ClientAssistantStreamResult {
|
||||
const event = entry.event
|
||||
this.durableCursor = Math.max(this.durableCursor, event.seq)
|
||||
this.transientInGap = 0
|
||||
const settlement = assistantSettlementEntry(entry)
|
||||
if (settlement !== undefined && this.attemptForSettlement(settlement.event) !== undefined) {
|
||||
if (this.pending.has(event.seq)) return { type: 'rebaseline' }
|
||||
this.pending.set(event.seq, settlement)
|
||||
return undefined
|
||||
}
|
||||
return this.publish(entry)
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold one dense transient frame and release its named durable settlement.
|
||||
* @param frame - next Assistant stream frame received by the follow connection.
|
||||
* @returns a transient, publication, or rebaseline decision, or `undefined` when no entry becomes visible.
|
||||
*/
|
||||
acceptFrame(frame: SessionAssistantStreamFrame): ClientAssistantStreamResult {
|
||||
switch (frame.type) {
|
||||
case 'start':
|
||||
if (this.activeAttempt !== undefined || this.pending.size > 0) return { type: 'rebaseline' }
|
||||
this.pending.clear()
|
||||
this.activeAttempt = {
|
||||
attemptId: frame.attemptId,
|
||||
startedAfterSeq: frame.startedAfterSeq,
|
||||
turn: frame.turn,
|
||||
step: frame.step,
|
||||
nextIndex: 0,
|
||||
}
|
||||
return undefined
|
||||
case 'chunk': {
|
||||
const attempt = this.activeAttempt
|
||||
// A controller mounted after the Host saw this attempt has no start
|
||||
// frame to reconstruct. Its durable settlement publishes directly;
|
||||
// ignore the transient suffix until the next known start.
|
||||
if (attempt === undefined || attempt.attemptId !== frame.attemptId) return undefined
|
||||
if (frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
|
||||
attempt.nextIndex += 1
|
||||
this.transientInGap += 1
|
||||
return {
|
||||
type: 'transient',
|
||||
entry: {
|
||||
type: 'transient',
|
||||
event: {
|
||||
type: 'assistant/live-chunk',
|
||||
seq: this.durableCursor + 1 - 1 / (this.transientInGap + 1),
|
||||
time: frame.time,
|
||||
data: {
|
||||
attemptId: frame.attemptId,
|
||||
turn: attempt.turn,
|
||||
step: attempt.step,
|
||||
chunk: frame.chunk as never,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
case 'end': {
|
||||
const attempt = this.activeAttempt
|
||||
if (attempt === undefined || attempt.attemptId !== frame.attemptId) {
|
||||
return undefined
|
||||
}
|
||||
this.activeAttempt = undefined
|
||||
if (frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
|
||||
if (frame.outcome.kind === 'abandoned') {
|
||||
return this.pending.size === 0
|
||||
? { type: 'abandonment', attemptId: attempt.attemptId }
|
||||
: { type: 'rebaseline' }
|
||||
}
|
||||
if (this.publishedSeqs.has(frame.outcome.seq)) return undefined
|
||||
const entry = this.pending.get(frame.outcome.seq)
|
||||
if (entry === undefined
|
||||
|| entry.event.type !== frame.outcome.eventType) {
|
||||
return { type: 'rebaseline' }
|
||||
}
|
||||
this.pending.delete(frame.outcome.seq)
|
||||
this.publishedSeqs.add(entry.event.seq)
|
||||
return { type: 'settlement', attemptId: attempt.attemptId, entry }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private attemptForSettlement(
|
||||
event: SessionAssistantSettlementEntry['event'],
|
||||
): ActiveAttempt | undefined {
|
||||
const attempt = this.activeAttempt
|
||||
if (attempt === undefined
|
||||
|| (event.type === 'assistant/message' && event.surfaceOp !== 'append')
|
||||
|| event.seq <= attempt.startedAfterSeq
|
||||
|| attempt.turn !== event.data.turn
|
||||
|| attempt.step !== event.data.step) return undefined
|
||||
return attempt
|
||||
}
|
||||
|
||||
private publish(entry: SessionLiveEventEntry): ClientAssistantStreamResult {
|
||||
this.publishedSeqs.add(entry.event.seq)
|
||||
return { type: 'publish', entry }
|
||||
}
|
||||
}
|
||||
|
||||
function assistantSettlementEntry(
|
||||
entry: SessionLiveEventEntry,
|
||||
): SessionAssistantSettlementEntry | undefined {
|
||||
return entry.event.type === 'assistant/message' || entry.event.type === 'assistant/attempt'
|
||||
? entry as SessionAssistantSettlementEntry
|
||||
: undefined
|
||||
}
|
||||
@@ -18,7 +18,7 @@ export function historyEntries(
|
||||
|
||||
/**
|
||||
* Read the first logical sequence represented by one wire record.
|
||||
* @param record - validated scalar event or packed Assistant delta run.
|
||||
* @param record - validated Session event.
|
||||
* @returns inclusive first Session sequence.
|
||||
*/
|
||||
export function historyRecordFirstSeq(record: SessionHistoryRecord): number {
|
||||
@@ -27,13 +27,9 @@ export function historyRecordFirstSeq(record: SessionHistoryRecord): number {
|
||||
|
||||
/**
|
||||
* Read the final logical sequence represented by one wire record.
|
||||
* @param record - validated scalar event or packed Assistant delta run.
|
||||
* @param record - validated Session event.
|
||||
* @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
|
||||
return record.event.seq
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import type {
|
||||
PromptContentPart,
|
||||
QueueAction,
|
||||
SessionAddress,
|
||||
SessionAssistantStreamBaseline,
|
||||
SessionControlFrame,
|
||||
SessionProjectionBaseline,
|
||||
SessionQueuedItem,
|
||||
@@ -35,6 +36,10 @@ import { ProjectionValueStore } from './projection-store.ts'
|
||||
import type { ProjectionsBaseline } from './projection-store.ts'
|
||||
import { resolvedClientTimeZone } from '../time-zone.ts'
|
||||
import { SessionQueueMirror } from './queue-mirror.ts'
|
||||
import {
|
||||
ClientAssistantStream,
|
||||
type ClientAssistantStreamResult,
|
||||
} from './assistant-stream.ts'
|
||||
|
||||
function projectionsBaseline(value: SessionProjectionBaseline): ProjectionsBaseline {
|
||||
return {
|
||||
@@ -95,6 +100,7 @@ export class Session implements SessionFace {
|
||||
private jumpPromise: Promise<void> | null = null
|
||||
/** Authoritative stream-only inbox snapshot; pending work never hits history. */
|
||||
private readonly queueMirror = new SessionQueueMirror()
|
||||
private readonly assistantStream = new ClientAssistantStream()
|
||||
private running = false
|
||||
private address: SubagentAddress | undefined
|
||||
private parentAvailable: boolean | undefined
|
||||
@@ -620,27 +626,67 @@ export class Session implements SessionFace {
|
||||
change.entries,
|
||||
change.hasMore,
|
||||
change.page.projections === undefined ? undefined : projectionsBaseline(change.page.projections),
|
||||
change.page.assistantStream,
|
||||
)
|
||||
return
|
||||
case 'prepend':
|
||||
this.prependWindow(change.entries, change.hasMore)
|
||||
return
|
||||
case 'append':
|
||||
if (this.appendLive(change.entry)) this.notifier.markDirty()
|
||||
this.publishAssistantEntry(this.assistantStream.acceptDurable(change.entry))
|
||||
return
|
||||
case 'assistant-stream':
|
||||
this.publishAssistantEntry(this.assistantStream.acceptFrame(change.frame))
|
||||
}
|
||||
}
|
||||
|
||||
/** Replace the complete contiguous window and apply page-owned projection metadata. */
|
||||
private installWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void {
|
||||
private installWindow(
|
||||
entries: readonly SessionEventLikeEntry[],
|
||||
hasMore: boolean,
|
||||
projections?: ProjectionsBaseline,
|
||||
assistantStream?: SessionAssistantStreamBaseline,
|
||||
): void {
|
||||
// A durable gap-repair page has no assistant baseline. Clearing transient
|
||||
// attempts makes a held notification reopen follow once for an atomic
|
||||
// page/baseline pair instead of applying it to an unrelated repair cut.
|
||||
const visible = this.assistantStream.replace(entries, assistantStream)
|
||||
this.baseSeq = SessionLogOffset(entries[0]?.event.seq ?? 0)
|
||||
this.hasMore = hasMore
|
||||
if (entries.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
|
||||
if (visible.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false
|
||||
if (projections !== undefined) this.projections.seed(projections)
|
||||
this.eventSource.replace(entries, hasMore)
|
||||
for (const entry of entries) this.observeSubmissionEvent(entry.event)
|
||||
this.eventSource.replace(visible, hasMore)
|
||||
for (const entry of visible) this.observeSubmissionEvent(entry.event)
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
|
||||
private publishAssistantEntry(result: ClientAssistantStreamResult): void {
|
||||
if (result?.type === 'rebaseline') {
|
||||
const events = this.events
|
||||
queueMicrotask(() => {
|
||||
if (events !== undefined && this.events === events) events.restart()
|
||||
})
|
||||
return
|
||||
}
|
||||
if (result?.type === 'settlement') {
|
||||
this.eventSource.settleAssistant(result.attemptId, result.entry)
|
||||
this.observeSubmissionEvent(result.entry.event)
|
||||
this.notifier.markDirty()
|
||||
return
|
||||
}
|
||||
if (result?.type === 'abandonment') {
|
||||
this.eventSource.settleAssistant(result.attemptId)
|
||||
this.notifier.markDirty()
|
||||
return
|
||||
}
|
||||
if (result?.type === 'publish' && this.appendLive(result.entry)) {
|
||||
this.notifier.markDirty()
|
||||
} else if (result?.type === 'transient') {
|
||||
this.eventSource.append(result.entry)
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
}
|
||||
|
||||
/** Prepend one stream-validated history page. */
|
||||
private prependWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void {
|
||||
this.baseSeq = entries[0] === undefined ? this.baseSeq : SessionLogOffset(entries[0].event.seq)
|
||||
|
||||
@@ -12,6 +12,8 @@ import {
|
||||
} from '@deepseek-ai/dsh-api-gateway/client'
|
||||
import type {
|
||||
SessionAddress,
|
||||
SessionAssistantStreamBaseline,
|
||||
SessionAssistantStreamFrame,
|
||||
SessionControlFrame,
|
||||
SessionHistoryRecord,
|
||||
SessionPage,
|
||||
@@ -40,6 +42,7 @@ export type SessionRemote = ClientRemote['session']
|
||||
/** Opening metadata carried only by a follow snapshot, never by loadOlder pages. */
|
||||
interface SessionJournalPage extends SessionPage {
|
||||
readonly projections?: SessionProjectionBaseline
|
||||
readonly assistantStream?: SessionAssistantStreamBaseline
|
||||
}
|
||||
|
||||
/** One complete publication from the Session journal stream. */
|
||||
@@ -51,27 +54,25 @@ export type SessionJournalChange =
|
||||
readonly hasMore: boolean
|
||||
}
|
||||
| { readonly type: 'append'; readonly entry: SessionLiveEventEntry }
|
||||
| { readonly type: 'assistant-stream'; readonly frame: SessionAssistantStreamFrame }
|
||||
|
||||
function toSessionJournalChange(
|
||||
change: RemoteJournalChange<SessionJournalPage, SessionHistoryRecord>,
|
||||
change: RemoteJournalChange<
|
||||
SessionJournalPage, SessionHistoryRecord, SessionAssistantStreamFrame
|
||||
>,
|
||||
): SessionJournalChange {
|
||||
switch (change.type) {
|
||||
case 'replace':
|
||||
case 'prepend':
|
||||
return { ...change, entries: historyEntries(change.entries) }
|
||||
case 'append': {
|
||||
if (change.entry.type !== 'event') {
|
||||
throw new RemoteError(
|
||||
'gateway/internal',
|
||||
'session live stream emitted a packed history record',
|
||||
{},
|
||||
)
|
||||
}
|
||||
return {
|
||||
type: 'append',
|
||||
entry: change.entry as unknown as SessionLiveEventEntry,
|
||||
}
|
||||
}
|
||||
case 'notification':
|
||||
return { type: 'assistant-stream', frame: change.notification }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,7 +137,8 @@ export class SessionEventStream extends RemoteJournalStream<
|
||||
SessionJournalPage,
|
||||
SessionHistoryRecord,
|
||||
number,
|
||||
ClientSessionPageRequest
|
||||
ClientSessionPageRequest,
|
||||
SessionAssistantStreamFrame
|
||||
> {
|
||||
/**
|
||||
* @param remote - generated Session namespace and Gateway stream factory.
|
||||
@@ -169,12 +171,24 @@ export class SessionEventStream extends RemoteJournalStream<
|
||||
protected override async * follow(
|
||||
request: ClientSessionPageRequest,
|
||||
signal: AbortSignal,
|
||||
): AsyncIterable<RemoteJournalFrame<SessionHistoryRecord, number, SessionJournalPage>> {
|
||||
): AsyncIterable<RemoteJournalFrame<
|
||||
SessionHistoryRecord, number, SessionJournalPage, SessionAssistantStreamFrame
|
||||
>> {
|
||||
let assistantRevision: number | undefined
|
||||
for await (const frame of this.remote.session.follow({
|
||||
address: this.address,
|
||||
assistantStream: true,
|
||||
...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }),
|
||||
}, signal)) {
|
||||
if (frame.type === 'snapshot') {
|
||||
if (frame.assistantStream === undefined) {
|
||||
throw new RemoteError(
|
||||
'gateway/internal',
|
||||
'session assistant stream omitted its opted-in opening baseline',
|
||||
{},
|
||||
)
|
||||
}
|
||||
assistantRevision = frame.assistantStream.revision
|
||||
yield {
|
||||
type: 'opened',
|
||||
cursor: frame.cursor,
|
||||
@@ -182,10 +196,22 @@ export class SessionEventStream extends RemoteJournalStream<
|
||||
records: frame.records,
|
||||
hasMore: frame.hasMore,
|
||||
projections: frame.projections,
|
||||
assistantStream: frame.assistantStream,
|
||||
},
|
||||
}
|
||||
continue
|
||||
}
|
||||
if (frame.type === 'assistant-stream') {
|
||||
const expected = (assistantRevision ?? 0) + 1
|
||||
if (frame.frame.revision !== expected) {
|
||||
throw new RemoteStreamCarrierError(
|
||||
`session assistant stream skipped revision ${String(expected)}`,
|
||||
)
|
||||
}
|
||||
assistantRevision = frame.frame.revision
|
||||
yield { type: 'notification', notification: frame.frame }
|
||||
continue
|
||||
}
|
||||
yield { type: 'entry', entry: frame }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import type { Agent, ModelSelection as AgentModelSelection } from '@deepseek-ai/
|
||||
import { AttachmentError, admitPromptContent } from '@deepseek-ai/dsh-attachment'
|
||||
import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
|
||||
import {
|
||||
ReasoningEffortId, createUserMessage, freezeMessage,
|
||||
ReasoningEffortId, createUserMessage, expandAssistantStream, freezeMessage,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import type { MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
|
||||
@@ -525,7 +525,6 @@ function imageInEvent(
|
||||
readonly content?: unknown
|
||||
readonly message?: { readonly content?: unknown }
|
||||
readonly inserted?: readonly { readonly content?: unknown }[]
|
||||
readonly chunk?: { readonly type?: unknown; readonly block?: unknown }
|
||||
}
|
||||
const direct = imageBlockIn(data.content, match)
|
||||
if (direct !== undefined) return direct
|
||||
@@ -535,9 +534,14 @@ function imageInEvent(
|
||||
const found = imageBlockIn(inserted.content, match)
|
||||
if (found !== undefined) return found
|
||||
}
|
||||
return event.type === 'assistant/chunk' && data.chunk?.type === 'block-end'
|
||||
? imageBlockIn([data.chunk.block], match)
|
||||
: undefined
|
||||
if (event.type === 'assistant/message' || event.type === 'assistant/attempt') {
|
||||
for (const { chunk } of expandAssistantStream(event.data.stream)) {
|
||||
if (chunk.type !== 'block-end') continue
|
||||
const found = imageBlockIn([chunk.block], match)
|
||||
if (found !== undefined) return found
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
function referencedImage(
|
||||
|
||||
@@ -2,12 +2,12 @@
|
||||
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import { Deque } from '@deepseek-ai/dsh-deque'
|
||||
import type { AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
|
||||
import {
|
||||
isAppendSurfaceEvent,
|
||||
SessionLogOffset,
|
||||
SessionSeq,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import { isChunkRow, packChunkRuns, type ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type {
|
||||
SessionEvent,
|
||||
SessionHeader,
|
||||
@@ -18,9 +18,10 @@ import type {
|
||||
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
|
||||
import type {} from '@deepseek-ai/dsh-subagent'
|
||||
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
import type {
|
||||
SessionAddress,
|
||||
SessionChunkRun,
|
||||
SessionAssistantStreamFrame,
|
||||
SessionEventEntry,
|
||||
SessionFollowRequest,
|
||||
SessionFollowFrame,
|
||||
@@ -32,6 +33,7 @@ import type {
|
||||
SessionWireHeader,
|
||||
SessionWireEvent,
|
||||
} from './types.ts'
|
||||
import { SessionAssistantStreamAccumulator } from './assistant-stream.ts'
|
||||
|
||||
const DEFAULT_MAX_MESSAGES = 50
|
||||
const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
|
||||
@@ -39,6 +41,7 @@ const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
|
||||
/** Implements cold-safe history operations delegated by the Session Controller. */
|
||||
export class SessionHistoryController {
|
||||
private readonly closeFollowers = new Set<() => void>()
|
||||
private readonly assistantStreams = new Map<SessionId, SessionAssistantStreamAccumulator>()
|
||||
|
||||
/**
|
||||
* @param ctx - Host context carrying Session query and projection services.
|
||||
@@ -48,6 +51,17 @@ export class SessionHistoryController {
|
||||
private readonly ctx: Context,
|
||||
private readonly promote: (observation: SessionObservation) => void,
|
||||
) {
|
||||
ctx.on('agent/assistant-stream', ({ agent, frame }) => {
|
||||
let stream = this.assistantStreams.get(agent.session.id)
|
||||
if (stream === undefined) {
|
||||
stream = new SessionAssistantStreamAccumulator()
|
||||
this.assistantStreams.set(agent.session.id, stream)
|
||||
}
|
||||
stream.accept(frame, cursorBeforeNext(agent.session.seq))
|
||||
}, { global: true })
|
||||
ctx.on('agent/disposed', ({ agent }) => {
|
||||
this.assistantStreams.delete(agent.session.id)
|
||||
}, { global: true })
|
||||
ctx.effect(() => () => {
|
||||
for (const close of this.closeFollowers) close()
|
||||
this.closeFollowers.clear()
|
||||
@@ -100,14 +114,22 @@ export class SessionHistoryController {
|
||||
* Follow events appended after an initial cursor on one durable address.
|
||||
* @param request - durable address and last committed sequence already held by the caller.
|
||||
* @param signal - stream cancellation owned by the Remote carrier.
|
||||
* @returns a complete opening snapshot followed by gap-free event frames.
|
||||
* @returns a complete opening snapshot followed by gap-free durable events and opted-in assistant frames.
|
||||
*/
|
||||
async *follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
|
||||
validateFollowRequest(request)
|
||||
const { address } = request
|
||||
const target = addressId(address)
|
||||
const buffered = new Deque<SessionEvent>()
|
||||
const buffered = new Deque<
|
||||
| { readonly type: 'event'; readonly event: SessionEvent }
|
||||
| {
|
||||
readonly type: 'assistant-stream'
|
||||
readonly frame: SessionAssistantStreamFrame
|
||||
readonly ordinal: number
|
||||
}
|
||||
>()
|
||||
let snapshotCursor: SessionSeqCursor | undefined
|
||||
let assistantStreamOrdinal = 0
|
||||
let wake: (() => void) | undefined
|
||||
const notify = (): void => {
|
||||
const resume = wake
|
||||
@@ -122,7 +144,7 @@ export class SessionHistoryController {
|
||||
this.closeFollowers.add(close)
|
||||
const disposeEvent = this.ctx.on('session/event', (session, event) => {
|
||||
if (session.id !== target) return
|
||||
buffered.pushBack(event)
|
||||
buffered.pushBack({ type: 'event', event })
|
||||
notify()
|
||||
}, { global: true })
|
||||
const disposeCreated = this.ctx.on('session/created', (session) => {
|
||||
@@ -134,10 +156,21 @@ export class SessionHistoryController {
|
||||
? session.firstLiveSeq
|
||||
: SessionLogOffset(snapshotCursor + 1))
|
||||
for (let index = suffix.length - 1; index >= 0; index -= 1) {
|
||||
buffered.pushFront(suffix[index] as SessionEvent)
|
||||
buffered.pushFront({ type: 'event', event: suffix[index] as SessionEvent })
|
||||
}
|
||||
notify()
|
||||
}, { global: true })
|
||||
const disposeAssistantStream = request.assistantStream !== true
|
||||
? undefined
|
||||
: this.ctx.on('agent/assistant-stream', ({ agent, frame }) => {
|
||||
if (agent.session.id !== target) return
|
||||
buffered.pushBack({
|
||||
type: 'assistant-stream',
|
||||
frame: wireAssistantStreamFrame(frame, cursorBeforeNext(agent.session.seq)),
|
||||
ordinal: ++assistantStreamOrdinal,
|
||||
})
|
||||
notify()
|
||||
}, { global: true })
|
||||
const onAbort = (): void => { notify() }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
try {
|
||||
@@ -147,15 +180,24 @@ export class SessionHistoryController {
|
||||
const cursor = source.cursor
|
||||
snapshotCursor = cursor
|
||||
const page = paginate(events, undefined, request.maxMessages ?? DEFAULT_MAX_MESSAGES)
|
||||
const assistantStream = request.assistantStream === true
|
||||
? this.assistantStreams.get(target)?.snapshot() ?? { revision: 0 }
|
||||
: undefined
|
||||
// The accumulator snapshot and this watermark are synchronous. Frames
|
||||
// through the cut are represented or superseded by that baseline,
|
||||
// including larger revisions from a retired Agent; later revision
|
||||
// resets reach Client continuity validation.
|
||||
const assistantStreamOrdinalCut = assistantStreamOrdinal
|
||||
yield {
|
||||
type: 'snapshot',
|
||||
header: wireHeader(source.header, source.inheritedEventCount),
|
||||
header: wireHeader(source.header),
|
||||
cursor,
|
||||
records: pageRecords(page.events),
|
||||
hasMore: page.hasMore,
|
||||
projections: source.projections === undefined
|
||||
? { asOfSeq: cursor, values: {} }
|
||||
: projectionBlock(source.projections),
|
||||
...assistantStream === undefined ? {} : { assistantStream },
|
||||
}
|
||||
if (address.kind === 'session' && source.source === 'prepared') {
|
||||
const promotion = source.retain()
|
||||
@@ -173,19 +215,26 @@ export class SessionHistoryController {
|
||||
await new Promise<void>((resolve) => { wake = resolve })
|
||||
continue
|
||||
}
|
||||
if (item.type === 'assistant-stream') {
|
||||
if (item.ordinal > assistantStreamOrdinalCut) {
|
||||
yield { type: 'assistant-stream', frame: item.frame }
|
||||
}
|
||||
continue
|
||||
}
|
||||
const expectedSeq = SessionSeq(nextOffset)
|
||||
if (item.seq < expectedSeq) continue
|
||||
if (item.seq !== expectedSeq) {
|
||||
if (item.event.seq < expectedSeq) continue
|
||||
if (item.event.seq !== expectedSeq) {
|
||||
throw new RemoteError('gateway/internal', `session event stream skipped seq ${String(expectedSeq)}`, {})
|
||||
}
|
||||
nextOffset = SessionLogOffset(nextOffset + 1)
|
||||
yield entryFor(item)
|
||||
yield entryFor(item.event)
|
||||
}
|
||||
} finally {
|
||||
this.closeFollowers.delete(close)
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
disposeCreated()
|
||||
disposeEvent()
|
||||
disposeAssistantStream?.()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -225,6 +274,22 @@ export class SessionHistoryController {
|
||||
|
||||
}
|
||||
|
||||
function cursorBeforeNext(nextSeq: SessionLogOffsetType): SessionSeqCursor {
|
||||
return nextSeq === 0 ? -1 : SessionSeq(nextSeq - 1)
|
||||
}
|
||||
|
||||
function wireAssistantStreamFrame(
|
||||
frame: AssistantStreamFrame,
|
||||
durableCursor: SessionSeqCursor,
|
||||
): SessionAssistantStreamFrame {
|
||||
if (frame.type === 'start') return { ...frame, startedAfterSeq: durableCursor }
|
||||
if (frame.type === 'end') return frame
|
||||
return {
|
||||
...frame,
|
||||
chunk: frame.chunk as JsonValue,
|
||||
}
|
||||
}
|
||||
|
||||
function projectionBlock(
|
||||
snapshot: NonNullable<SessionObservation['projections']>,
|
||||
): SessionProjectionBaseline {
|
||||
@@ -343,16 +408,9 @@ function paginate(
|
||||
return { events: events.slice(cut, end), hasMore: cut > 0 }
|
||||
}
|
||||
|
||||
/** Translate logical Session metadata to the unchanged v0 browser wire. */
|
||||
function wireHeader(
|
||||
header: SessionHeader,
|
||||
inheritedEventCount: SessionLogOffsetType,
|
||||
): SessionWireHeader {
|
||||
const { isSeeded, ...wire } = header
|
||||
return {
|
||||
...wire,
|
||||
...isSeeded ? { seedLength: inheritedEventCount } : {},
|
||||
}
|
||||
/** Translate current logical Session metadata to the browser wire. */
|
||||
function wireHeader(header: SessionHeader): SessionWireHeader {
|
||||
return { ...header }
|
||||
}
|
||||
|
||||
function entryFor(event: SessionEvent): SessionEventEntry {
|
||||
@@ -363,29 +421,7 @@ function entryFor(event: SessionEvent): SessionEventEntry {
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
? chunkEntryFor(record)
|
||||
: entryFor(record))
|
||||
return events.map(entryFor)
|
||||
}
|
||||
|
||||
@@ -17,11 +17,7 @@ import { SessionCommandController } from './commands.ts'
|
||||
import { SessionControlController } from './control.ts'
|
||||
import { SessionHistoryController } from './history.ts'
|
||||
import { SessionFileReferences } from './file-references.ts'
|
||||
import {
|
||||
ApiSessionList,
|
||||
DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
|
||||
DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS,
|
||||
} from './list.ts'
|
||||
import { ApiSessionList } from './list.ts'
|
||||
import { buildModelCatalog } from './catalog.ts'
|
||||
import { installModelSelectionProjection } from './model-selection-projection.ts'
|
||||
import { SessionSkillCatalog } from './skill-catalog.ts'
|
||||
@@ -70,10 +66,6 @@ declare module '@deepseek-ai/cordis' {
|
||||
|
||||
/** Session Controller deployment policy. */
|
||||
export interface Config {
|
||||
/** Maximum stat-reported event count eligible for one full cold projection observation; `0` disables the event-count gate. */
|
||||
readonly coldBlankProbeMaxEvents?: number
|
||||
/** Maximum stat-reported artifact byte size eligible for one full cold projection observation; `0` disables the byte-size gate. */
|
||||
readonly coldBlankProbeMaxBytes?: number
|
||||
/** Override platform desktop-opener detection. */
|
||||
readonly nativeOpen?: boolean
|
||||
}
|
||||
@@ -101,8 +93,6 @@ export class SessionController extends TypertRemoteService {
|
||||
]
|
||||
|
||||
static Config: z<Config> = z.object({
|
||||
coldBlankProbeMaxEvents: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS),
|
||||
coldBlankProbeMaxBytes: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_BYTES),
|
||||
nativeOpen: z.boolean(),
|
||||
})
|
||||
|
||||
@@ -117,7 +107,7 @@ export class SessionController extends TypertRemoteService {
|
||||
|
||||
/**
|
||||
* @param ctx - Host context containing the Session capability assembly.
|
||||
* @param config - cold-list observation and native-opener deployment policy.
|
||||
* @param config - native-opener deployment policy.
|
||||
* @param internals - host integrations replaceable by direct unit tests.
|
||||
*/
|
||||
constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) {
|
||||
@@ -132,10 +122,7 @@ export class SessionController extends TypertRemoteService {
|
||||
await Promise.allSettled([...this.promotions])
|
||||
}, 'session-controller.promotions')
|
||||
this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
|
||||
this.listState = new ApiSessionList(ctx, {
|
||||
coldBlankProbeMaxEvents: config.coldBlankProbeMaxEvents ?? DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS,
|
||||
coldBlankProbeMaxBytes: config.coldBlankProbeMaxBytes ?? DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
|
||||
})
|
||||
this.listState = new ApiSessionList(ctx)
|
||||
this.openPath = internals.openPath ?? openNativePath
|
||||
this.canOpenPath = internals.canOpenPath
|
||||
?? (() => config.nativeOpen ?? (internals.openPath !== undefined || canOpenNativePath()))
|
||||
@@ -382,7 +369,8 @@ export class SessionController extends TypertRemoteService {
|
||||
* Follow one Session log from its opening or resume cursor.
|
||||
* @param request - durable address and last committed sequence already held by the caller.
|
||||
* @param signal - cancellation owned by the Remote stream carrier.
|
||||
* @returns a complete opening snapshot followed by gap-free event frames.
|
||||
* @returns a complete opening snapshot followed by gap-free durable event
|
||||
* frames and optional cursorless assistant-stream frames.
|
||||
*/
|
||||
@Remote({ mode: 'stream' })
|
||||
follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
|
||||
|
||||
@@ -19,21 +19,6 @@ import type {
|
||||
SessionSearchValue, SessionSummary,
|
||||
} from './types.ts'
|
||||
|
||||
/** Default maximum stat-reported event count eligible for one cold projection observation. */
|
||||
export const DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS = 16
|
||||
|
||||
/** Default maximum stat-reported artifact size eligible for one cold projection observation. */
|
||||
export const DEFAULT_COLD_BLANK_PROBE_MAX_BYTES = 1024
|
||||
|
||||
/** Resolved cold-blank probe policy: each threshold gates its stat metric; `0` disables that gate. */
|
||||
export interface ColdBlankProbePolicy {
|
||||
/** Maximum stat-reported `eventCount` eligible for a full observation. */
|
||||
readonly coldBlankProbeMaxEvents: number
|
||||
/** Maximum stat-reported `sizeBytes` eligible for a full observation. */
|
||||
readonly coldBlankProbeMaxBytes: number
|
||||
}
|
||||
|
||||
const COLD_SUMMARY_BATCH_SIZE = 16
|
||||
const SEARCH_PROVIDER_CALL_LIMIT = 100
|
||||
const SESSION_SEARCH_QUERY_MAX_CHARS = 500
|
||||
const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
|
||||
@@ -90,14 +75,8 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
|
||||
|
||||
/** Owns list projection registration, bounded cold summaries, and authorized search. */
|
||||
export class ApiSessionList {
|
||||
/**
|
||||
* @param ctx - Host context carrying Session, query, persistence, and projection services.
|
||||
* @param probe - stat-metadata thresholds gating a full cold observation.
|
||||
*/
|
||||
constructor(
|
||||
private readonly ctx: Context,
|
||||
private readonly probe: ColdBlankProbePolicy,
|
||||
) {
|
||||
/** @param ctx - Host context carrying Session, query, persistence, and projection services. */
|
||||
constructor(private readonly ctx: Context) {
|
||||
ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
|
||||
key: 'sessionListMetadata',
|
||||
stateSchema: sessionListMetadataSchema,
|
||||
@@ -159,28 +138,13 @@ export class ApiSessionList {
|
||||
if (record.header.cwd === undefined) continue
|
||||
cold.push(record.header)
|
||||
}
|
||||
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
|
||||
const settled = await Promise.allSettled(cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
|
||||
.map(header => this.summarizeCold(header, signal)))
|
||||
for (const result of settled) {
|
||||
if (result.status === 'rejected') throw result.reason
|
||||
items.push(result.value)
|
||||
}
|
||||
}
|
||||
for (const header of cold) items.push(this.summarizeCold(header))
|
||||
items.sort((left, right) => right.updatedAt - left.updatedAt)
|
||||
return items
|
||||
}
|
||||
|
||||
private async summarizeCold(
|
||||
header: SessionHeader,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<SessionSummary> {
|
||||
const cached = this.projectionsFor(header, undefined)
|
||||
const projections = cached?.values.sessionListMetadata?.blank === false
|
||||
? cached
|
||||
: await this.probeSmallCold(header, signal) ?? cached
|
||||
const raced = this.ctx.sessions.get(header.id)
|
||||
if (raced !== undefined) return this.summaryFor(raced)
|
||||
private summarizeCold(header: SessionHeader): SessionSummary {
|
||||
const projections = this.projectionsFor(header, undefined)
|
||||
const metadata = projections?.values.sessionListMetadata
|
||||
return {
|
||||
sessionId: header.id,
|
||||
@@ -193,54 +157,6 @@ export class ApiSessionList {
|
||||
}
|
||||
}
|
||||
|
||||
private async probeSmallCold(
|
||||
header: SessionHeader,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<SessionProjectionHints | undefined> {
|
||||
const { coldBlankProbeMaxEvents, coldBlankProbeMaxBytes } = this.probe
|
||||
if (coldBlankProbeMaxEvents === 0 && coldBlankProbeMaxBytes === 0) return undefined
|
||||
const persistence = this.ctx.get('sessionPersistence')
|
||||
if (persistence === undefined) return undefined
|
||||
signal?.throwIfAborted()
|
||||
let snapshot: Awaited<ReturnType<typeof persistence.stat>>
|
||||
try {
|
||||
snapshot = await persistence.stat(header.id, signal === undefined ? {} : { signal })
|
||||
} catch (error: unknown) {
|
||||
// An unreadable single session degrades to unknown state instead of
|
||||
// failing the whole list request.
|
||||
signal?.throwIfAborted()
|
||||
this.ctx.logger.warn(
|
||||
`api-session.list: cold stat for "${header.id}" failed; serving it as visible: ${String(error)}`,
|
||||
)
|
||||
return undefined
|
||||
}
|
||||
if (snapshot === undefined) return undefined
|
||||
if (snapshot.eventCount !== undefined) {
|
||||
if (coldBlankProbeMaxEvents === 0 || snapshot.eventCount > coldBlankProbeMaxEvents) return undefined
|
||||
} else if (snapshot.sizeBytes !== undefined) {
|
||||
if (coldBlankProbeMaxBytes === 0 || snapshot.sizeBytes > coldBlankProbeMaxBytes) return undefined
|
||||
} else {
|
||||
// The backend offers no cheap size hint, so a full observation is unbounded work.
|
||||
return undefined
|
||||
}
|
||||
try {
|
||||
using observation = await this.ctx.sessionQuery.observeSession(header.id, {
|
||||
...(signal === undefined ? {} : { signal }),
|
||||
projectionMode: 'all',
|
||||
})
|
||||
const block = observation.projections
|
||||
return block === undefined
|
||||
? undefined
|
||||
: { asOfSeq: block.asOfSeq, values: block.values as SessionProjectionValues }
|
||||
} catch (error: unknown) {
|
||||
signal?.throwIfAborted()
|
||||
this.ctx.logger.warn(
|
||||
`api-session.list: small cold observation for "${header.id}" failed; serving it as visible: ${String(error)}`,
|
||||
)
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Search current visible message content without activating any matching Session.
|
||||
* @param query - literal message-content query.
|
||||
@@ -354,10 +270,12 @@ export class ApiSessionList {
|
||||
session: Session | undefined,
|
||||
): SessionProjectionHints | undefined {
|
||||
try {
|
||||
const cache = this.ctx.get('sessionProjectionCache')
|
||||
const block = session === undefined
|
||||
? header.isSeeded
|
||||
? undefined
|
||||
: this.ctx.get('sessionProjectionCache')?.cachedSnapshot(header, SessionLogOffset(0))
|
||||
: cache?.cachedSnapshot(header, SessionLogOffset(0))
|
||||
?? cache?.cachedPredecessorTitle(header, SessionLogOffset(0))
|
||||
: this.ctx.sessionProjections.cachedSnapshot(session)
|
||||
return block !== undefined && Object.keys(block.values).length > 0
|
||||
? {
|
||||
|
||||
@@ -4,10 +4,9 @@ import type {
|
||||
AttachmentIdType, ImageAttachmentLimits, ImageAttachmentRef, ImageMediaType,
|
||||
} from '@deepseek-ai/dsh-attachment'
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
|
||||
import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { SessionId } from '@deepseek-ai/dsh-session/types'
|
||||
import type { LlmAttemptId, MessageId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionId, SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
|
||||
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
|
||||
import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
|
||||
@@ -384,15 +383,15 @@ export interface SessionEventEntry {
|
||||
readonly event: SessionWireEvent
|
||||
}
|
||||
|
||||
/** v0-compatible Session metadata carried on the browser wire. */
|
||||
/** Current logical Session metadata carried on the browser wire. */
|
||||
export interface SessionWireHeader {
|
||||
readonly version: number
|
||||
readonly id: SessionId
|
||||
readonly createdAt: number
|
||||
readonly cwd?: string
|
||||
readonly parentSession?: SessionId
|
||||
/** Exact inherited prefix length; absent for an unseeded Session. */
|
||||
readonly seedLength?: number
|
||||
/** Whether the Session contains a fork-inherited prefix. */
|
||||
readonly isSeeded: boolean
|
||||
readonly origin?: 'subagent'
|
||||
readonly delegationDepth?: number
|
||||
readonly agentPreset?: string
|
||||
@@ -403,24 +402,8 @@ export type SessionWireSurfaceOp =
|
||||
| 'append'
|
||||
| { readonly op: 'replace'; readonly start: number; readonly end: number }
|
||||
|
||||
/** 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 type: 'chunks'
|
||||
readonly event: ChunkRowEvent
|
||||
}
|
||||
|
||||
/** One history-page record: a raw event or a packed Assistant delta run. */
|
||||
export type SessionHistoryRecord = SessionEventEntry | SessionChunkRun
|
||||
/** One history-page record. V2 embeds compact Assistant streams inside events. */
|
||||
export type SessionHistoryRecord = SessionEventEntry
|
||||
|
||||
/** Session event wire form; durable readers own recognition of merge-extensible event names. */
|
||||
export interface SessionWireEvent {
|
||||
@@ -446,15 +429,69 @@ export interface SessionPageRequest {
|
||||
export interface SessionFollowRequest {
|
||||
readonly address: SessionAddress
|
||||
readonly maxMessages?: number
|
||||
/** Include process-local assistant presentation frames for the Web client. */
|
||||
readonly assistantStream?: true
|
||||
}
|
||||
|
||||
/** One active assistant attempt in a reconnect opening snapshot. */
|
||||
export interface SessionAssistantStreamAttempt {
|
||||
readonly attemptId: LlmAttemptId
|
||||
/** Last durable Session seq observed when this attempt started. */
|
||||
readonly startedAfterSeq: SessionSeqCursor
|
||||
readonly turn: number
|
||||
readonly step: number
|
||||
/** Dense position expected for the next live chunk frame. */
|
||||
readonly nextIndex: number
|
||||
/** Compact detached stream accumulated at this opening revision. */
|
||||
readonly stream: readonly JsonValue[]
|
||||
}
|
||||
|
||||
/** Complete process-local assistant state at one follow opening. */
|
||||
export interface SessionAssistantStreamBaseline {
|
||||
readonly revision: number
|
||||
readonly activeAttempt?: SessionAssistantStreamAttempt
|
||||
}
|
||||
|
||||
/** Browser wire form of one process-local assistant frame. */
|
||||
export type SessionAssistantStreamFrame =
|
||||
| {
|
||||
readonly type: 'start'
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly revision: number
|
||||
readonly startedAfterSeq: SessionSeqCursor
|
||||
readonly turn: number
|
||||
readonly step: number
|
||||
}
|
||||
| {
|
||||
readonly type: 'chunk'
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly revision: number
|
||||
readonly index: number
|
||||
readonly time: number
|
||||
readonly chunk: JsonValue
|
||||
}
|
||||
| {
|
||||
readonly type: 'end'
|
||||
readonly attemptId: LlmAttemptId
|
||||
readonly revision: number
|
||||
/** Number of chunk frames represented by this terminal marker. */
|
||||
readonly index: number
|
||||
readonly outcome:
|
||||
| {
|
||||
readonly kind: 'committed'
|
||||
readonly eventType: 'assistant/message' | 'assistant/attempt'
|
||||
readonly seq: number
|
||||
}
|
||||
| { readonly kind: 'abandoned' }
|
||||
}
|
||||
|
||||
/** One contiguous backwards page of a Session log. */
|
||||
export interface SessionPage {
|
||||
readonly records: readonly SessionHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
}
|
||||
|
||||
/** Complete opening window followed by ordered events appended after its cursor. */
|
||||
/** Complete opening window followed by ordered durable events and opted-in assistant frames. */
|
||||
export type SessionFollowFrame =
|
||||
| {
|
||||
readonly type: 'snapshot'
|
||||
@@ -463,8 +500,10 @@ export type SessionFollowFrame =
|
||||
readonly records: readonly SessionHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
readonly projections: SessionProjectionBaseline
|
||||
readonly assistantStream?: SessionAssistantStreamBaseline
|
||||
}
|
||||
| SessionEventEntry
|
||||
| { readonly type: 'assistant-stream'; readonly frame: SessionAssistantStreamFrame }
|
||||
|
||||
/** One pending inbox occurrence in the authoritative queue snapshot. */
|
||||
export interface SessionQueuedItem {
|
||||
|
||||
Reference in New Issue
Block a user