第 8 讲pi-coding-agent 架构：AgentSession 与核心模块

/**
 * AgentSession - Core abstraction for agent lifecycle and session management.
 *
 * This class is shared between all run modes (interactive, print, rpc).
 * It encapsulates:
 * - Agent state access
 * - Event subscription with automatic session persistence
 * - Model and thinking level management
 * - Compaction (manual and auto)
 * - Bash execution
 * - Session switching and branching
 *
 * Modes use this class and add their own I/O layer on top.
 */

export class AgentSession {
  // ……大量私有状态：队列、重试、工具注册表、扩展绑定……
  constructor(config: AgentSessionConfig) {
    this.agent = config.agent;
    this.sessionManager = config.sessionManager;
    this.settingsManager = config.settingsManager;
    this._resourceLoader = config.resourceLoader;
    this._cwd = config.cwd;
    this._modelRuntime = config.modelRuntime;
    this._cacheWarmer = config.cacheWarmer;
    // ……工具白名单 / 黑名单 / 扩展引用等初始化……

    // Always subscribe to agent events for internal handling
    // (session persistence, extensions, auto-compaction, retry logic)
    this._unsubscribeAgent = this.agent.subscribe(this._handleAgentEvent);
    this._installAgentToolHooks();          // 工具调用前后的扩展钩子
    this._installAgentNextTurnRefresh();
    this._installAgentRequestProjection();
    this._installAgentBoundaryHooks();     // turn_end / before_settle 边界
    this._installHiddenDeclarationsProjection();
    this._installAgentForcedPromptProjection();
    this._buildRuntime({
      activeToolNames: this._initialActiveToolNames,
      includeAllExtensionTools: true,
    });
  }
}

/**
 * Owns the current AgentSession plus its cwd-bound services.
 *
 * Session replacement methods tear down the current runtime first, then create
 * and apply the next runtime. If creation fails, the error is propagated to the
 * caller. The caller is responsible for user-facing error handling.
 */
export class AgentSessionRuntime {
  private rebindSession?: (session: AgentSession) => Promise<void>;
  private beforeSessionInvalidate?: () => void;
  private _session: AgentSession;
  private _services: AgentSessionServices;
  private readonly createRuntime: CreateAgentSessionRuntimeFactory;
  // ……
}

export type CreateAgentSessionRuntimeFactory = (options: {
  cwd: string;
  agentDir: string;
  sessionManager: SessionManager;
  sessionStartEvent?: SessionStartEvent;
  projectTrustContext?: ProjectTrustContext;
}) => Promise<CreateAgentSessionRuntimeResult>;

export interface AgentSessionServices {
  cwd: string;
  agentDir: string;
  modelRuntime: ModelRuntime;
  settingsManager: SettingsManager;
  resourceLoader: ResourceLoader;
  diagnostics: AgentSessionRuntimeDiagnostic[];
}

export async function createAgentSessionServices(
  options: CreateAgentSessionServicesOptions,
): Promise<AgentSessionServices> {
  const cwd = resolvePath(options.cwd);
  const agentDir = options.agentDir ? resolvePath(options.agentDir) : getAgentDir();
  const modelRuntime =
    options.modelRuntime ??
    (await ModelRuntime.create({
      authPath: join(agentDir, "auth.json"),
      modelsPath: join(agentDir, "models.json"),
      signal: options.modelRuntimeSignal,
    }));
  const settingsManager = options.settingsManager ?? SettingsManager.create(cwd, agentDir);
  const resourceLoader = new DefaultResourceLoader({ cwd, agentDir, settingsManager });
  await resourceLoader.reload(options.resourceLoaderReloadOptions);

  const diagnostics: AgentSessionRuntimeDiagnostic[] = [];
  const extensionsResult = resourceLoader.getExtensions();
  for (const { name, config, extensionPath } of extensionsResult.runtime.pendingProviderRegistrations) {
    try { modelRuntime.registerProvider(name, config); }
    catch (error) { /* 记入 diagnostics 而非直接退出 */ }
  }
  // ……native provider、virtual model 注册同理……
  await modelRuntime.refresh({ allowNetwork: false });
  diagnostics.push(...applyExtensionFlagValues(resourceLoader, options.extensionFlagValues));
  return { cwd, agentDir, modelRuntime, settingsManager, resourceLoader, diagnostics };
}

import { EventEmitter } from "node:events";

export interface EventBus {
  emit(channel: string, data: unknown): void;
  on(channel: string, handler: (data: unknown) => void): () => void;
}

export interface EventBusController extends EventBus {
  clear(): void;
}

export function createEventBus(): EventBusController {
  const emitter = new EventEmitter();
  return {
    emit: (channel, data) => { emitter.emit(channel, data); },
    on: (channel, handler) => {
      const safeHandler = async (data: unknown) => {
        try { await handler(data); }
        catch (err) { console.error(`Event handler error (${channel}):`, err); }
      };
      emitter.on(channel, safeHandler);
      return () => emitter.off(channel, safeHandler);
    },
    clear: () => { emitter.removeAllListeners(); },
  };
}

/** Emit extension events based on agent events */
private async function _emitExtensionEvent(event: AgentEvent): Promise<void> {
  if (event.type === "agent_start") {
    this._turnIndex = 0;
    await this._extensionRunner.emit({ type: "agent_start" });
  } else if (event.type === "turn_start") {
    const extensionEvent: TurnStartEvent = {
      type: "turn_start",
      turnIndex: this._turnIndex,        // 内核没有的字段，桥接层补上
      timestamp: Date.now(),
    };
    await this._extensionRunner.emit(extensionEvent);
  } else if (event.type === "message_end") {
    const extensionEvent: MessageEndEvent = { type: "message_end", message: event.message };
    const replacement = await this._extensionRunner.emitMessageEnd(extensionEvent);
    if (replacement) {                 // 扩展可以请求替换消息
      /* 规范化后就地替换 agent state 与会话历史中的消息 */
      this._replaceMessageInPlace(event.message, normalized);
    }
  } else if (event.type === "tool_execution_end") {
    const extensionEvent: ToolExecutionEndEvent = {
      type: "tool_execution_end",
      toolCallId: event.toolCallId, toolName: event.toolName,
      args: event.args, result: event.result, isError: event.isError,
      ...(event.durationMs === undefined ? {} : { durationMs: event.durationMs }),
    };
    await this._extensionRunner.emit(extensionEvent);
  }
  // ……message_start / message_update / tool_execution_start / update / turn_end 同理
}

private _handleAgentEvent = async (event: AgentEvent): Promise<void> => {
  // 嵌套工具调用（ctx.executeTool）的记录与 usage 归并……
  // steering / followUp 队列消息出队……

  // Emit to extensions first, then notify public listeners.
  await this._emitExtensionEvent(event);
  this._emit(event.type === "agent_end"
    ? { ...event, willRetry: this._willRetryAfterAgentEnd(event) } : event);

  // Handle session persistence
  if (event.type === "message_end") {
    // system / user / assistant / toolResult → SessionMessageEntry
    // custom → CustomMessageEntry
    entryId = this.sessionManager.appendMessage(event.message);
    if (entryId) this._entryIdsByMessage.set(event.message, entryId);
  }
  if (event.type === "turn_end") {
    this._flushPendingCustomMessages();  // 避开 toolCall/toolResult 中缝
  }
};

static async create(options: CreateModelRuntimeOptions = {}): Promise<ModelRuntime> {
  const credentials = new RuntimeCredentials(options.credentials ?? DefaultAuthStorage.create(options.authPath));
  const modelsPath = options.modelsPath ?? join(getAgentDir(), "models.json");
  const config = await ModelConfig.load(modelsPath);
  const modelsStore = options.modelsStore ??
    (modelsPath ? new FileModelsStore(options.modelsStorePath
      ?? join(dirname(modelsPath), "models-store.json"))
      : new InMemoryCodingAgentModelsStore());
  const providers = builtinProviderCatalog.builtinProviders().map((provider) => {
    // OpenRouter 按用户 key 个性化目录；其余走远程目录缓存
    return withRemoteCatalog(provider, options.catalogBaseUrl, builtinModelDataGeneratedAt);
  });
  const runtime = new ModelRuntime(credentials, config, modelsPath, modelsStore, providers, process.env.PI_OFFLINE === undefined);
  runtime.rebuildProviders();
  if (options.refreshOnCreate !== false) {
    await runtime.refresh({ allowNetwork: refreshFromNetwork, signal });
  }
  return runtime;
}

/** The provider without virtual models, or undefined when nothing defines it. */
private composeProvider(providerId: string): Provider | undefined {
  const base = this.nativeExtensionProviders.get(providerId) ?? this.builtins.get(providerId);
  const extension = this.extensionProviders.get(providerId);
  if (!this.config.getProvider(providerId) && !extension) {
    // No overlays: use the builtin untouched so its auth/login/stream behavior is exact.
    return base;
  }
  try {
    const provider = composeModelProvider(providerId, base, this.config, extension);
    return provider;
  } catch (error) {
    this.compositionErrors.set(providerId, /* 错误信息 */ String(error));
    return base;   // 合成失败退回内置，不炸启动
  }
}

export function composeModelProvider(providerId, base, config, extension) {
  // 1. 校验扩展自定义 streamSimple 必须带 api 适配
  // 2. applyModelsJson：把 models.json 里的 baseUrl/headers/models 覆盖到 base
  // 3. applyExtension：oauth / headers / fetchDeferred / classify 等能力叠加
  // 4. 校验：no authentication method configured 时抛错
}

/** Default chat model IDs for providers with built-in chat models. */
export const defaultModelPerProvider: Partial<Record<KnownProvider, string>> = {
  anthropic: "claude-opus-5-5",
  openai: "gpt-6.1-sol",
  google: "gemini-4-argon",
  openrouter: "moonshotai/kimi-k2.7-code",
  zai: "glm-5.3",
  // ……
};

const agent = new Agent({
  initialState: {
    systemPrompt: "", model, thinkingLevel, tools: [],
    messages: existingSession.messages,
  },
  convertToLlm: convertToLlmWithBlockImages,
  streamFn: async (model, context, options) => {
    const requestOptions = buildRequestOptions(options);
    if (options?.sessionId === sessionManager.getSessionId()) {
      cacheWarmer.start({ model, context, options: requestOptions }, cacheContextIsCurrent(model));
    }
    return modelRuntime.streamSimple(model, context, requestOptions);
  },
  onPayload: transformProviderPayload,       // before_provider_request
  onResponse: handleProviderResponse,        // after_provider_response
  onProviderStreamEvent: handleProviderStreamEvent,
  sessionId: sessionManager.getSessionId(),
  transformContext: async (messages) => {  // 扩展的上下文变换钩子
    const runner = extensionRunnerRef.current;
    if (!runner) return messages;
    return runner.emitContext(messages);
  },
  // steeringMode / followUpMode / transport / thinkingBudgets 等均来自 SettingsManager
});
// This stream resolves auth through modelRuntime, so summaries must not pre-resolve it (#11520).
markRuntimeStreamFunction(agent.streamFunction, modelRuntime);

/**
 * Merge `defaultTools` of two settings layers. A list with plain tool names replaces
 * the inherited one; a list of only `+name`/`-name` entries is appended, so it
 * modifies the inherited selection.
 */
function mergeDefaultTools(base, overrides) {
  if (overrides === undefined) return base;
  if (!Array.isArray(base) || !Array.isArray(overrides) || !overrides.every(isToolModifier)) return overrides;
  return [...base, ...overrides];
}

/** Resolve a merged `defaultTools` list: plain names replace `DEFAULT_TOOL_NAMES`,
 * then `+name` adds and `-name` removes a tool, in list order. */
function resolveDefaultTools(entries: string[]): string[] { /* …… */ }

/** Deep merge settings: project/overrides take precedence, nested objects merge recursively */
function deepMergeSettings(base: Settings, overrides: Settings): Settings {
  const merged = deepMergeObjects(base, overrides);
  const defaultTools = mergeDefaultTools(base.defaultTools, overrides.defaultTools);
  return defaultTools === undefined ? merged : { ...merged, defaultTools };
}

private static function loadFromStorage(storage, scope: SettingsScope, projectTrusted = true): Settings {
  if (scope === "project" && !projectTrusted) {
    return {};   // 未信任：项目设置整体不读
  }
  // 读文件 → stripBom → JSON.parse → migrateSettings 迁移旧字段
}

const TRUST_REQUIRING_PROJECT_CONFIG_RESOURCES = [
  "settings.json",
  "mcp.json",
  "extensions",
  "skills",
  "prompts",
  "themes",
  "SYSTEM.md",
  "APPEND_SYSTEM.md",
] as const;

/**
 * Returns true when cwd has project-local resources that must be gated by
 * project trust: trust-requiring entries under cwd/.pi, or .agents/skills in
 * cwd or one of its ancestors.
 */
export function hasTrustRequiringProjectResources(cwd: string): boolean {
  const configDir = join(canonicalizePath(resolvePath(cwd)), ".pi");
  if (TRUST_REQUIRING_PROJECT_CONFIG_RESOURCES.some((entry) => existsSync(join(configDir, entry)))) {
    return true;
  }
  // 向上遍历祖先目录检查 .agents/skills（排除用户主目录下的 ~/.agents/skills）
}

export async function resolveProjectTrusted(options: ResolveProjectTrustedOptions): Promise<boolean> {
  // 1. 命令行 --approve / --no-approve 直接裁决
  if (options.trustOverride !== undefined) return options.trustOverride;

  // 没有需要门控的资源 → 读已存决策，缺省 true
  if (!hasTrustRequiringProjectResources(options.cwd)) {
    return options.trustStore.get(options.cwd) ?? true;
  }

  // 2. 用户级扩展可接管 project_trust 事件，第一个返回者拥有决策
  if (options.extensionsResult) {
    const { result, errors } = await emitProjectTrustEvent(options.extensionsResult,
      { type: "project_trust", cwd: options.cwd }, options.projectTrustContext);
    if (result) {
      const trusted = result.trusted === "yes";
      if (result.remember === true) options.trustStore.set(options.cwd, trusted);
      return trusted;
    }
  }

  // 3. 已保存决策（最近祖先）
  const decision = options.trustStore.get(options.cwd);
  if (decision !== null) return decision;

  // 4. defaultProjectTrust 设置："always" | "never" | "ask"
  switch (options.defaultProjectTrust ?? SETTINGS_DEFAULTS.defaultProjectTrust) {
    case "always": return true;
    case "never": return false;
  }

  // 5. 无 UI（print/json/rpc）直接拒绝；有 UI 弹选择框
  if (!options.projectTrustContext.hasUI) return false;
  const selected = await selectProjectTrustOption(options.cwd, options.projectTrustContext);
  if (selected !== undefined) {
    saveProjectTrustPromptResult(options.trustStore, selected);
    return selected.trusted;
  }
  return false;
}

// If streaming, use prompt() with steer behavior
// This handles extension commands (execute immediately), prompt template expansion, and queueing
if (this.session.isStreaming) {
  this.editor.addToHistory(text);
  this.editor.setText("");
  await this.session.prompt(text, { streamingBehavior: "steer" });
  this.updatePendingMessagesDisplay();
  this.ui.requestRender();
  return;
}
// Normal message submission ……

/**
 * Send a prompt to the agent.
 * - Handles extension commands (registered via pi.registerCommand) immediately, even during streaming
 * - Expands file-based prompt templates by default
 * - During streaming, queues via steer() or followUp() based on streamingBehavior option
 * - Validates model and API key before sending (when not streaming)
 */
async prompt(text: string, options?: PromptOptions): Promise<void> {
  if (this._isEmittingAgentSettled) {
    this._deferredSettledActions.push(async () => await this.prompt(text, options));
    return;
  }
  // Handle extension commands first (execute immediately, even during streaming)
  if (expandPromptTemplates && text.startsWith("/")) {
    const handled = await this._tryExecuteExtensionCommand(text);
    if (handled) { preflightResult?.("handled"); return; }
  }
  if (this._compactionAbortController !== undefined /* …… */) {
    throw new Error("Cannot submit a prompt while compaction is in progress. ……");
  }
  // Reserve the context through async input hooks and auth checks so tree navigation cannot
  // start meanwhile (#9154). _startRun marks the run active synchronously, so there is no gap.
  this._promptPreflightCount++;
  let trigger: BeforeAgentStartTrigger | undefined;
  try {
    trigger = await this._preparePrompt(text, options);
  } finally {
    this._promptPreflightCount--;
  }
  if (trigger) await this._startRun(trigger, () => preflightResult?.("started"));
}

private async function _startRun(trigger, onStarted?): Promise<void> {
  this._isAgentRunActive = true;
  this._agentRunAbortRequested = false;
  let messages: AgentMessage[];
  try {
    messages = await this._prepareRun(trigger);
  } catch (error) {
    this._isAgentRunActive = false;
    this._flushPendingBashMessages(); this._flushPendingCustomMessages();
    this._resolveIdleWaitIfIdle();
    throw error;
  }
  this._flushPendingBashMessages(); this._flushPendingCustomMessages();
  onStarted?.();
  this._failedResponse = undefined;
  this._recordSelection();
  this._pendingToolNames.clear();
  try {
    await this.agent.prompt(messages);
    while (!this._agentRunAbortRequested) {
      if (await this._handlePostAgentRun()) {          // 重试 / 压缩 / 队列续跑
        await this.agent.continue(); continue;
      }
      if (!(await this._runBeforeSettleBoundary())) break;
      await this.agent.continue();
    }
  } finally {
    await this._emitAgentSettled();               // agent_settled 事件
  }
}