实现最小 Coding Agent

目标不是复刻 Codex

我们只实现能验证架构的最小切片：

• 接收一个仓库任务；
• 把历史和工具 schema 送给支持 function calling 的模型；
• 只提供 read_file、search、apply_patch、run_test；
• 写操作前经过策略；
• 工具结果回到历史；
• 发出事件；
• 达到停止条件后报告。

不会实现 TUI、MCP、恢复、跨平台沙箱和完整 compaction。那些是从原型走向产品的主要工作量。

flowchart LR
G["Goal"] --> C["ContextStore"]
C --> M["ModelClient"]
M --> R["ToolRegistry"]
R --> P["Policy"]
P --> X["Executor"]
X --> C
C --> E["EventSink"]

第一步：先定义协议

下面是教学实现，不是 Codex 官方源码，也没有使用 Codex 私有 API。ModelClient 可以由任意支持结构化工具调用的模型适配。

type Role = "user" | "assistant" | "tool";

type Message =
  | { role: "user" | "assistant"; content: string }
  | { role: "assistant"; toolCalls: ToolCall[] }
  | { role: "tool"; callId: string; content: string; isError?: boolean };

type ToolCall = {
  id: string;
  name: "read_file" | "search" | "apply_patch" | "run_test";
  arguments: unknown;
};

type ModelOutput =
  | { type: "message"; text: string }
  | { type: "tool_calls"; calls: ToolCall[] };

interface ModelClient {
  sample(input: {
    instructions: string;
    messages: Message[];
    tools: ToolSpec[];
  }): Promise<ModelOutput>;
}

interface ToolSpec {
  name: ToolCall["name"];
  description: string;
  inputSchema: Record<string, unknown>;
}

interface AgentEvent {
  type: "turn.start" | "tool.start" | "tool.end" | "message" | "turn.end";
  data: Record<string, unknown>;
}

协议先行带来一个好处：测试 Agent Loop 时，可以用假模型和假工具，不需要真的写磁盘或调用网络。

第二步：工具必须窄

interface ToolContext {
  root: string;
  signal: AbortSignal;
}

interface ToolHandler {
  spec: ToolSpec;
  mutates: boolean;
  execute(args: unknown, context: ToolContext): Promise<string>;
}

class ToolRegistry {
  #handlers = new Map<string, ToolHandler>();

  register(handler: ToolHandler) {
    if (this.#handlers.has(handler.spec.name)) {
      throw new Error(`duplicate tool: ${handler.spec.name}`);
    }
    this.#handlers.set(handler.spec.name, handler);
  }

  specs() {
    return [...this.#handlers.values()].map((handler) => handler.spec);
  }

  get(name: string) {
    const handler = this.#handlers.get(name);
    if (!handler) throw new Error(`unknown tool: ${name}`);
    return handler;
  }
}

实现 handler 时至少做三件事：

1. 用 schema 校验 args，不要相信模型输出；
2. 将路径解析到仓库根后确认没有逃逸；
3. 限制输出长度，同时保留退出码和错误摘要。

第三步：策略独立于工具

interface Policy {
  authorize(input: {
    call: ToolCall;
    handler: ToolHandler;
  }): Promise<"allow" | "deny">;
}

class ReadOnlyPolicy implements Policy {
  async authorize({ handler }: { handler: ToolHandler }) {
    return handler.mutates ? "deny" : "allow";
  }
}

原型先用只读策略。确认事件、错误和停止条件正确后，再增加一次性批准；不要先给子进程宿主的全部权限。

第四步：闭合 Agent Loop

class CodingAgent {
  constructor(
    private model: ModelClient,
    private tools: ToolRegistry,
    private policy: Policy,
    private emit: (event: AgentEvent) => void,
    private maxSteps = 24,
  ) {}

  async run(goal: string, context: ToolContext): Promise<string> {
    const messages: Message[] = [{ role: "user", content: goal }];
    this.emit({ type: "turn.start", data: { goal } });

    for (let step = 0; step < this.maxSteps; step += 1) {
      if (context.signal.aborted) throw new Error("cancelled");

      const output = await this.model.sample({
        instructions: [
          "Work only inside the repository root.",
          "Inspect before editing.",
          "Use run_test to validate changes.",
          "State unverified assumptions in the final message.",
        ].join("\n"),
        messages,
        tools: this.tools.specs(),
      });

      if (output.type === "message") {
        messages.push({ role: "assistant", content: output.text });
        this.emit({ type: "message", data: { text: output.text } });
        this.emit({ type: "turn.end", data: { step } });
        return output.text;
      }

      messages.push({ role: "assistant", toolCalls: output.calls });

      for (const call of output.calls) {
        const handler = this.tools.get(call.name);
        const decision = await this.policy.authorize({ call, handler });

        this.emit({
          type: "tool.start",
          data: { callId: call.id, name: call.name, decision },
        });

        if (decision === "deny") {
          messages.push({
            role: "tool",
            callId: call.id,
            content: "Denied by policy. Choose a read-only alternative or ask the user.",
            isError: true,
          });
          continue;
        }

        try {
          const content = await handler.execute(call.arguments, context);
          messages.push({ role: "tool", callId: call.id, content });
          this.emit({ type: "tool.end", data: { callId: call.id, ok: true } });
        } catch (error) {
          const content = error instanceof Error ? error.message : String(error);
          messages.push({ role: "tool", callId: call.id, content, isError: true });
          this.emit({
            type: "tool.end",
            data: { callId: call.id, ok: false, error: content },
          });
        }
      }
    }

    throw new Error(`step limit exceeded: ${this.maxSteps}`);
  }
}

这段循环复现了核心机制：普通消息终止，工具调用执行后继续，错误也回到 Context，外部可以消费事件。

第五步：用假模型测试控制流

const scriptedModel: ModelClient = {
  #step: 0,
  async sample() {
    this.#step += 1;
    if (this.#step === 1) {
      return {
        type: "tool_calls",
        calls: [{ id: "1", name: "search", arguments: { query: "retry" } }],
      };
    }
    return { type: "message", text: "找到重试入口；当前为只读模式，未修改文件。" };
  },
};

更好的可运行版本：

let sampleCount = 0;
const scriptedModel: ModelClient = {
  async sample() {
    sampleCount += 1;
    return sampleCount === 1
      ? {
          type: "tool_calls" as const,
          calls: [{ id: "1", name: "search" as const, arguments: { query: "retry" } }],
        }
      : { type: "message" as const, text: "只读探索完成。" };
  },
};

测试应断言事件顺序：

turn.start
tool.start(search)
tool.end(search)
message
turn.end

从原型到可用，还缺什么

最小产品的正确验收

不要用“模型成功改了一次 Todo App”验收。至少准备：

1. 只读问题定位；
2. 单文件可测试修改；
3. Patch 上下文冲突；
4. 测试失败后二次修复；
5. 被拒绝的写操作；
6. 超长命令输出；
7. 用户取消；
8. 恶意仓库文本诱导越权。

最后一章记录这套教材的 研究基线与证据边界。