Prompting 源码分析 — AgentSession 消息处理与排队系统

1

async prompt(text: string, options?: PromptOptions): Promise<void> {

1
2
3
4
5
6

async prompt(text: string, options?: PromptOptions): Promise<void> {
  const expandPromptTemplates = options?.expandPromptTemplates ?? true;
  const preflightResult = options?.preflightResult;
  let messages: AgentMessage[] | undefined;

  try {

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

    // Handle extension commands first (execute immediately, even during streaming)
    // Extension commands manage their own LLM interaction via pi.sendMessage()
    if (expandPromptTemplates && text.startsWith("/")) {
      const handled = await this._tryExecuteExtensionCommand(text);
      if (handled) {
        // Extension command executed, no prompt to send
        preflightResult?.(true);    // ───┐
        return;                     //    │ [Return 1]
      }                             //    │ 扩展命令已处理，无需发送 prompt
    }                               // ───┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

    let currentText = text;
    let currentImages = options?.images;
    if (this._extensionRunner.hasHandlers("input")) {
      const inputResult = await this._extensionRunner.emitInput(
        currentText,
        currentImages,
        options?.source ?? "interactive",   // source 参数：区分交互/扩展来源
        this.isStreaming ? options?.streamingBehavior : undefined,
      );
      if (inputResult.action === "handled") {
        preflightResult?.(true);    // ───┐
        return;                     //    │ [Return 2]
      }                             //    │ 扩展在 input 事件中处理了消息
      if (inputResult.action === "transform") {
        currentText = inputResult.text;
        currentImages = inputResult.images ?? currentImages;
      }
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

	/** Emit input event. Transforms chain, "handled" short-circuits. */
	async emitInput(
		text: string,
		images: ImageContent[] | undefined,
		source: InputSource,
		streamingBehavior?: "steer" | "followUp",
	): Promise<InputEventResult> {
		const ctx = this.createContext();
		let currentText = text;
		let currentImages = images;

		for (const ext of this.extensions) {
			for (const handler of ext.handlers.get("input") ?? []) {
				try {
					const event: InputEvent = {
						type: "input",
						text: currentText,
						images: currentImages,
						source,
						streamingBehavior,
					};
					const result = (await handler(event, ctx)) as InputEventResult | undefined;
					if (result?.action === "handled") return result;
					if (result?.action === "transform") {
						currentText = result.text;
						currentImages = result.images ?? currentImages;
					}
				} catch (err) {
					this.emitError({
						extensionPath: ext.path,
						event: "input",
						error: err instanceof Error ? err.message : String(err),
						stack: err instanceof Error ? err.stack : undefined,
					});
				}
			}
		}
		return currentText !== text || currentImages !== images
			? { action: "transform", text: currentText, images: currentImages }
			: { action: "continue" };
	}

1
2
3
4
5

    let expandedText = currentText;
    if (expandPromptTemplates) {
      expandedText = this._expandSkillCommand(expandedText);
      expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

    if (this.isStreaming) {
      if (!options?.streamingBehavior) {
        throw new Error(
          "Agent is already processing. Specify streamingBehavior " +
          "('steer' or 'followUp') to queue the message.",
        );
      }
      if (options.streamingBehavior === "followUp") {
        await this._queueFollowUp(expandedText, currentImages);
      } else {
        await this._queueSteer(expandedText, currentImages);
      }
      preflightResult?.(true);    // ───┐
      return;                     //    │ [Return 3]
    }                             // ───┘ 消息已排队，无需发送

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

    // Flush any pending bash messages before the new prompt
    this._flushPendingBashMessages();

    if (!this.model) {
      throw new Error(formatNoModelSelectedMessage());
    }

    if (!this._modelRegistry.hasConfiguredAuth(this.model)) {
      const isOAuth = this._modelRegistry.isUsingOAuth(this.model);
      if (isOAuth) {
        throw new Error(
          `Authentication failed for "${this.model.provider}". ` +
          `Credentials may have expired or network is unavailable. ` +
          `Run '/login ${this.model.provider}' to re-authenticate.`,
        );
      }
      throw new Error(formatNoApiKeyFoundMessage(this.model.provider));
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

    const lastAssistant = this._findLastAssistantMessage();
    if (lastAssistant && (await this._checkCompaction(lastAssistant, false))) {
      try {
        await this.agent.continue();
        while (await this._handlePostAgentRun()) {
          await this.agent.continue();
        }
      } finally {
        this._flushPendingBashMessages();
      }
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

    messages = [];

    // 1. user message
    const userContent = [{ type: "text", text: expandedText }];
    if (currentImages) userContent.push(...currentImages);
    messages.push({ role: "user", content: userContent, timestamp: Date.now() });

    // 2. pending "nextTurn" asides
    for (const msg of this._pendingNextTurnMessages) {
      messages.push(msg);
    }
    this._pendingNextTurnMessages = [];

    // 3. before_agent_start 事件
    const result = await this._extensionRunner.emitBeforeAgentStart(
      expandedText, currentImages,
      this._baseSystemPrompt, this._baseSystemPromptOptions,
    );

    // 4. 扩展注入 custom messages
    if (result?.messages) {
      for (const msg of result.messages) {
        messages.push({ role: "custom", customType: msg.customType, content: msg.content, ... });
      }
    }
    // 5. systemPrompt 覆盖或回退
    if (result?.systemPrompt) {
      this.agent.state.systemPrompt = result.systemPrompt;
    } else {
      this.agent.state.systemPrompt = this._baseSystemPrompt;
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  } catch (error) {
    preflightResult?.(false);    // 任何 try 块中的错误
    throw error;
  }

  if (!messages) {
    return;                      // 防呆
  }

  preflightResult?.(true);       // 通过所有检查，确认可发送
  await this._runAgentPrompt(messages);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

export interface PromptOptions {
  /** Whether to expand file-based prompt templates (default: true) */
  expandPromptTemplates?: boolean;
  /** Image attachments */
  images?: ImageContent[];
  /** When streaming, how to queue the message: "steer" (interrupt) or "followUp" (wait) */
  streamingBehavior?: "steer" | "followUp";
  /** Source of input for extension input event handlers. Defaults to "interactive". */
  source?: InputSource;
  /** Internal hook used by RPC mode to observe prompt preflight acceptance or rejection */
  preflightResult?: (success: boolean) => void;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// 场景 1: 模型从未选择
if (!this.model) {
  throw new Error(formatNoModelSelectedMessage());
}

// 场景 2: 模型有但缺乏认证凭据
if (!this._modelRegistry.hasConfiguredAuth(this.model)) {
  const isOAuth = this._modelRegistry.isUsingOAuth(this.model);
  if (isOAuth) {
    throw new Error(
      `Authentication failed for "${this.model.provider}". ` +
      `Credentials may have expired or network is unavailable. ` +
      `Run '/login ${this.model.provider}' to re-authenticate.`,
    );
  }
  throw new Error(formatNoApiKeyFoundMessage(this.model.provider));
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

const lastAssistant = this._findLastAssistantMessage();
if (lastAssistant && (await this._checkCompaction(lastAssistant, false))) {
  try {
    await this.agent.continue();
    while (await this._handlePostAgentRun()) {
      await this.agent.continue();
    }
  } finally {
    this._flushPendingBashMessages();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

private _findLastAssistantMessage(): AssistantMessage | undefined {
  const messages = this.agent.state.messages;
  for (let i = messages.length - 1; i >= 0; i--) {
    const msg = messages[i];
    if (msg.role === "assistant") {
      return msg as AssistantMessage;
    }
  }
  return undefined;
}

1
2
3
4
5
6

const result = await this._extensionRunner.emitBeforeAgentStart(
  expandedText,
  currentImages,
  this._baseSystemPrompt,
  this._baseSystemPromptOptions,
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// 扩展注入的 custom messages 加入消息数组
if (result?.messages) {
  for (const msg of result.messages) {
    messages.push({
      role: "custom",
      customType: msg.customType,
      content: msg.content,
      display: msg.display,
      details: msg.details,
      timestamp: Date.now(),
    });
  }
}
// systemPrompt 回退逻辑
if (result?.systemPrompt) {
  this.agent.state.systemPrompt = result.systemPrompt;
} else {
  this.agent.state.systemPrompt = this._baseSystemPrompt;
}

1
2
3
4
5
6
7
8
9

try {
  // ... 所有检查步骤（命令/input/技能展开/streaming/auth/预压缩/消息构建）...
} catch (error) {
  preflightResult?.(false);  // 任何检查失败 → 通知 RPC 模式
  throw error;
}

preflightResult?.(true);  // 通过所有检查 → 确认可发送
await this._runAgentPrompt(messages);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

private async _queueSteer(text: string, images?: ImageContent[]): Promise<void> {
  this._steeringMessages.push(text);
  this._emitQueueUpdate();
  this.agent.steer({
    role: "user",
    content: [{ type: "text", text }, ...(images ?? [])],
    timestamp: Date.now(),
  });
}

private async _queueFollowUp(text: string, images?: ImageContent[]): Promise<void> {
  this._followUpMessages.push(text);
  this._emitQueueUpdate();
  this.agent.followUp({
    role: "user",
    content: [{ type: "text", text }, ...(images ?? [])],
    timestamp: Date.now(),
  });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

private async _runAgentPrompt(messages: AgentMessage | AgentMessage[]): Promise<void> {
  try {
    await this.agent.prompt(messages);
    while (await this._handlePostAgentRun()) {
      await this.agent.continue();
    }
  } finally {
    this._flushPendingBashMessages();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// agent-session.ts:2636
recordBashResult(command: string, result: BashResult, options?): void {
  const bashMessage: BashExecutionMessage = {
    role: "bashExecution",
    command,
    output: result.output,
    exitCode: result.exitCode,
    // ... 更多字段
  };

  // 流式 → 排队；非流式 → 立即加入
  if (this.isStreaming) {
    this._pendingBashMessages.push(bashMessage);
  } else {
    this.agent.state.messages.push(bashMessage);
    this.sessionManager.appendMessage(bashMessage);
  }
}

1
2
3
4
5
6
7
8

private _flushPendingBashMessages(): void {
  if (this._pendingBashMessages.length === 0) return;
  for (const bashMessage of this._pendingBashMessages) {
    this.agent.state.messages.push(bashMessage);
    this.sessionManager.appendMessage(bashMessage);
  }
  this._pendingBashMessages = [];
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

private async _handlePostAgentRun(): Promise<boolean> {
  const msg = this._lastAssistantMessage;
  this._lastAssistantMessage = undefined;
  if (!msg) return false;

  // Check 1: 重试
  if (this._isRetryableError(msg) && (await this._prepareRetry(msg))) return true;

  // 重试失败时，发送 auto_retry_end 事件并重置计数器
  if (msg.stopReason === "error" && this._retryAttempt > 0) { ... }

  // Check 2: 压缩
  if (await this._checkCompaction(msg)) return true;

  // Check 3: 扩展在 agent_end 中排队了消息
  return this.agent.hasQueuedMessages();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

private async _tryExecuteExtensionCommand(text: string): Promise<boolean> {
  const spaceIndex = text.indexOf(" ");
  const commandName = spaceIndex === -1 ? text.slice(1) : text.slice(1, spaceIndex);
  const args = spaceIndex === -1 ? "" : text.slice(spaceIndex + 1);

  const command = this._extensionRunner.getCommand(commandName);
  if (!command) return false;

  const ctx = this._extensionRunner.createCommandContext();
  try {
    await command.handler(args, ctx);
    return true;
  } catch (err) {
    this._extensionRunner.emitError({
      extensionPath: `command:${commandName}`,
      event: "command",
      error: err instanceof Error ? err.message : String(err),
    });
    return true;
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

private _expandSkillCommand(text: string): string {
  if (!text.startsWith("/skill:")) return text;

  const spaceIndex = text.indexOf(" ");
  const skillName = spaceIndex === -1 ? text.slice(7) : text.slice(7, spaceIndex);
  const args = spaceIndex === -1 ? "" : text.slice(spaceIndex + 1).trim();

  const skill = this.resourceLoader.getSkills().skills.find((s) => s.name === skillName);
  if (!skill) return text;

  const content = readFileSync(skill.filePath, "utf-8");
  const body = stripFrontmatter(content).trim();
  const skillBlock = `<skill name="${skill.name}" location="${skill.filePath}">\nReferences are relative to ${skill.baseDir}.\n\n${body}\n</skill>`;
  return args ? `${skillBlock}\n\n${args}` : skillBlock;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

async steer(text: string, images?: ImageContent[]): Promise<void> {
  // 扩展命令不能排队
  if (text.startsWith("/")) this._throwIfExtensionCommand(text);
  // 展开技能和模板
  let expandedText = this._expandSkillCommand(text);
  expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
  await this._queueSteer(expandedText, images);
}

async followUp(text: string, images?: ImageContent[]): Promise<void> {
  if (text.startsWith("/")) this._throwIfExtensionCommand(text);
  let expandedText = this._expandSkillCommand(text);
  expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]);
  await this._queueFollowUp(expandedText, images);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

private _throwIfExtensionCommand(text: string): void {
  const spaceIndex = text.indexOf(" ");
  const commandName = spaceIndex === -1 ? text.slice(1) : text.slice(1, spaceIndex);
  const command = this._extensionRunner.getCommand(commandName);
  if (command) {
    throw new Error(
      `Extension command "/${commandName}" cannot be queued. ` +
      `Use prompt() or execute the command when not streaming.`,
    );
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

async sendCustomMessage<T = unknown>(
  message: Pick<CustomMessage<T>, "customType" | "content" | "display" | "details">,
  options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" },
): Promise<void> {
  const appMessage = {
    role: "custom" as const,
    customType: message.customType,
    content: message.content,
    display: message.display,
    details: message.details,
    timestamp: Date.now(),
  } satisfies CustomMessage<T>;

  if (options?.deliverAs === "nextTurn") {
    this._pendingNextTurnMessages.push(appMessage);
  } else if (this.isStreaming) {
    if (options?.deliverAs === "followUp") {
      this.agent.followUp(appMessage);
    } else {
      this.agent.steer(appMessage);  // 默认 steering 入队
    }
  } else if (options?.triggerTurn) {
    await this._runAgentPrompt(appMessage);  // 立即触发新 turn
  } else {
    this.agent.state.messages.push(appMessage);
    this.sessionManager.appendCustomMessageEntry(...);
    this._emit({ type: "message_start", message: appMessage });
    this._emit({ type: "message_end", message: appMessage });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

async sendUserMessage(
  content: string | (TextContent | ImageContent)[],
  options?: { deliverAs?: "steer" | "followUp" },
): Promise<void> {
  // 规范化: content 数组 → text + images
  let text: string;
  let images: ImageContent[] | undefined;
  if (typeof content === "string") {
    text = content;
  } else {
    const textParts: string[] = [];
    images = [];
    for (const part of content) {
      if (part.type === "text") textParts.push(part.text);
      else images.push(part);
    }
    text = textParts.join("\n");
    if (images.length === 0) images = undefined;
  }

  await this.prompt(text, {
    expandPromptTemplates: false,   // 跳过命令处理和模板展开
    streamingBehavior: options?.deliverAs,
    images,
    source: "extension",            // 标记来源为扩展
  });
}

1
2
3
4
5
6
7
8
9

clearQueue(): { steering: string[]; followUp: string[] } {
  const steering = [...this._steeringMessages];
  const followUp = [...this._followUpMessages];
  this._steeringMessages = [];
  this._followUpMessages = [];
  this.agent.clearAllQueues();
  this._emitQueueUpdate();
  return { steering, followUp };
}

1
2
3
4
5

async abort(): Promise<void> {
  this.abortRetry();
  this.agent.abort();
  await this.agent.waitForIdle();
}

1
2
3
4
5
6
7

private _getUserMessageText(message: Message): string {
  if (message.role !== "user") return "";
  const content = message.content;
  if (typeof content === "string") return content;
  const textBlocks = content.filter((c) => c.type === "text");
  return textBlocks.map((c) => (c as TextContent).text).join("");
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

if (event.type === "message_start" && event.message.role === "user") {
  this._overflowRecoveryAttempted = false;  // 用户新输入 → 重置 overflow 恢复标志
  const messageText = this._getUserMessageText(event.message);
  if (messageText) {
    // 先检查 steering 队列
    const steeringIndex = this._steeringMessages.indexOf(messageText);
    if (steeringIndex !== -1) {
      this._steeringMessages.splice(steeringIndex, 1);
      this._emitQueueUpdate();
    } else {
      // 再检查 followUp 队列
      const followUpIndex = this._followUpMessages.indexOf(messageText);
      if (followUpIndex !== -1) {
        this._followUpMessages.splice(followUpIndex, 1);
        this._emitQueueUpdate();
      }
    }
  }
}

1
2
3
4
5
6
7

private _emitQueueUpdate(): void {
  this._emit({
    type: "queue_update",
    steering: [...this._steeringMessages],
    followUp: [...this._followUpMessages],
  });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

get pendingMessageCount(): number {
  return this._steeringMessages.length + this._followUpMessages.length;
}

getSteeringMessages(): readonly string[] {
  return this._steeringMessages;
}

getFollowUpMessages(): readonly string[] {
  return this._followUpMessages;
}