refactor(agent-loop): unify pending message representation
This commit is contained in:
@@ -12,6 +12,7 @@ import { Agent, AgentMessageId, agentCarrier, agentInterruptReasonOf, assembleCo
|
||||
import { createScope } from '@deepseek-ai/dsh-scope'
|
||||
import type { Scope } from '@deepseek-ai/dsh-scope'
|
||||
import type {
|
||||
AgentMessage,
|
||||
AgentMessageId as AgentMessageIdType,
|
||||
CancelOptions,
|
||||
AgentInterruptReason,
|
||||
@@ -25,28 +26,19 @@ import {
|
||||
BlockAssembler, LlmError, deepFreeze, errorChain, isHarnessError, llmFailureOf, markAgentLoopRequest,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import type {
|
||||
ContentBlock, GenerateOptions, LlmCallConfig, Message, MessageSource,
|
||||
ContentBlock, GenerateOptions, LlmCallConfig, Message,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session'
|
||||
import type { PromptMessageData, Session, SessionId, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionId, TurnEndReason, TurnTrigger, UserMessageData } from '@deepseek-ai/dsh-session'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import { executeToolCalls } from './tool-calls.ts'
|
||||
|
||||
/** One message waiting in the queued or steering inbox. */
|
||||
interface PendingMessage {
|
||||
id: AgentMessageIdType
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
interface PendingMessage extends AgentMessage {
|
||||
wakeup: boolean
|
||||
}
|
||||
|
||||
/** Model-facing input awaiting the next step boundary. */
|
||||
interface OutboxItem extends PromptMessageData {
|
||||
/** Present only when this input is a live inbox item. */
|
||||
steering?: PendingMessage
|
||||
}
|
||||
|
||||
function withoutToolCalls(message: Message): Message {
|
||||
return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
|
||||
}
|
||||
@@ -59,7 +51,7 @@ export class ReactLoopAgent extends Agent {
|
||||
/** Prompts awaiting individual turns. */
|
||||
private queued: PendingMessage[] = []
|
||||
/** Input taken into the session log at step boundaries. */
|
||||
private outbox: OutboxItem[] = []
|
||||
private outbox: (UserMessageData | PendingMessage)[] = []
|
||||
|
||||
/** Whether observers see a running interval; consecutive turns share it. */
|
||||
private busy = false
|
||||
@@ -120,7 +112,7 @@ export class ReactLoopAgent extends Agent {
|
||||
wakeup,
|
||||
}
|
||||
if (steering) {
|
||||
this.outbox.push({ content: message.content, source: message.source, steering: message })
|
||||
this.outbox.push(message)
|
||||
} else {
|
||||
this.queued.push(message)
|
||||
}
|
||||
@@ -145,7 +137,7 @@ export class ReactLoopAgent extends Agent {
|
||||
if (!options.keepInbox) {
|
||||
const discarded = [
|
||||
...this.queued,
|
||||
...this.outbox.map(item => item.steering).filter(steering => steering !== undefined),
|
||||
...this.outbox.filter((item): item is PendingMessage => 'id' in item),
|
||||
]
|
||||
// Clear before abort observers run: replacement work belongs to the next turn.
|
||||
this.queued.length = 0
|
||||
@@ -246,7 +238,7 @@ export class ReactLoopAgent extends Agent {
|
||||
step += 1
|
||||
const { continueTurn, maxTokens } = await this.step(turn, step, signal)
|
||||
if (maxTokens) reason = { kind: 'max-tokens' }
|
||||
if (continueTurn || this.outbox.some(item => item.steering !== undefined)) continue
|
||||
if (continueTurn || this.outbox.some(item => 'id' in item)) continue
|
||||
await this.loopCtx.serial(agentCarrier(this), 'agent/stopping', this, turn, signal)
|
||||
signal.throwIfAborted()
|
||||
if (!this.drainOutbox(turn)) break
|
||||
@@ -438,14 +430,17 @@ export class ReactLoopAgent extends Agent {
|
||||
private drainOutbox(turn: number): boolean {
|
||||
let steered = false
|
||||
for (const item of this.outbox.splice(0)) {
|
||||
const { steering: message, ...data } = item
|
||||
if (message === undefined) {
|
||||
this.session.append('user/message', data, { surfaceOp: 'append' })
|
||||
if (!('id' in item)) {
|
||||
this.session.append('user/message', item, { surfaceOp: 'append' })
|
||||
continue
|
||||
}
|
||||
steered = true
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
|
||||
this.session.append('steering/message', { turn, ...data }, { surfaceOp: 'append' })
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item)
|
||||
this.session.append(
|
||||
'steering/message',
|
||||
{ turn, content: item.content, source: item.source },
|
||||
{ surfaceOp: 'append' },
|
||||
)
|
||||
}
|
||||
return steered
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ import type { Context } from 'cordis'
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { Scoped } from '@deepseek-ai/dsh-scope'
|
||||
import type { ContentBlock, LlmCallConfig, LlmFailure, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionId, UserMessageData } from '@deepseek-ai/dsh-session'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
declare module '@deepseek-ai/dsh-system-prompt' {
|
||||
interface AssembleContext {
|
||||
@@ -85,11 +85,9 @@ export function AgentMessageId(id: string): AgentMessageId {
|
||||
* message's enqueue, dequeue, and discard events. Source defaults are already
|
||||
* applied, so these are the exact values the item was accepted with.
|
||||
*/
|
||||
export interface AgentMessage {
|
||||
export interface AgentMessage extends UserMessageData {
|
||||
/** The id `send` returned for this message. */
|
||||
id: AgentMessageId
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
}
|
||||
|
||||
/** Options for {@link Agent.cancel}. */
|
||||
@@ -111,10 +109,7 @@ export interface CancelOptions {
|
||||
export type AgentStatus = 'idle' | 'running'
|
||||
|
||||
/** Additional model-facing context produced beside a prompt or tool result. */
|
||||
export interface AdditionalContext {
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
}
|
||||
export type AdditionalContext = UserMessageData
|
||||
|
||||
/**
|
||||
* Prompt interception result. `allow.content` replaces the prompt, while
|
||||
|
||||
@@ -185,13 +185,13 @@ export interface EpochHeader {
|
||||
export type RequestHeaderReason = 'initial' | 'resume' | 'change'
|
||||
|
||||
/**
|
||||
* Shared payload for user, injected-context, and steering prompt messages. A
|
||||
* Shared payload for user, injected-context, and steering messages. A
|
||||
* direct human prompt, a synthetic `agent.inject()` context, and mid-turn
|
||||
* steering all project into the model transcript as verbatim user-role content;
|
||||
* they are told apart by `source` (a non-`user` kind marks injected context),
|
||||
* not by event type.
|
||||
*/
|
||||
export interface PromptMessageData {
|
||||
export interface UserMessageData {
|
||||
/** Exact model-facing blocks. */
|
||||
content: ContentBlock[]
|
||||
/** Producer provenance. */
|
||||
@@ -228,7 +228,7 @@ export interface SessionEventMap {
|
||||
* project their `content` verbatim; `source` tells them apart. An idle
|
||||
* injection may append this event between turns without running the model.
|
||||
*/
|
||||
'user/message': PromptMessageData
|
||||
'user/message': UserMessageData
|
||||
/**
|
||||
* Durable record of a prompt veto and its reason. It is log-only: the blocked
|
||||
* prompt never enters the model-visible surface, and its turn runs zero steps.
|
||||
@@ -270,7 +270,7 @@ export interface SessionEventMap {
|
||||
meta?: JsonValue
|
||||
}
|
||||
/** Steering content injected between steps of a running turn. */
|
||||
'steering/message': PromptMessageData & { turn: number }
|
||||
'steering/message': UserMessageData & { turn: number }
|
||||
/** Whole-list snapshot; latest write wins on replay. Log-only UI state; never derived history. */
|
||||
'todo/write': { todos: TodoItem[] }
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user