admin/infinite-canvas
1
0
Fork 0
mirror of https://github.com/basketikun/infinite-canvas.git synced 2026-08-05 08:54:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(codex): add structured task progress support in conversation flow, displaying status updates for pending, in-progress, and completed tasks

Browse Source
This commit is contained in:
HouYunFei
2026-07-29 12:57:21 +08:00
parent ea67a5abf5
commit 20844d5e99
9 changed files with 195 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
canvas-agent/src/agent/codex-client.ts
+30 -3
View File
@@ -6,7 +6,7 @@ import { fileURLToPath } from "node:url";
import { VERSION } from "../config.js";
import { logger } from "../utils/logger.js";
import { field, type JsonRecord } from "../utils/value.js";
import type { CodexNotificationParams, CodexRequestMethod, CodexRequestParams, CodexRequestResult, CodexTurnInput } from "./codex-protocol.js";
import type { CodexNotificationParams, CodexPlanUpdate, CodexRequestMethod, CodexRequestParams, CodexRequestResult, CodexTurnInput } from "./codex-protocol.js";
import type { AgentEmit } from "./types.js";
type AgentEvent = JsonRecord & { type: string; usage?: unknown };
@@ -30,6 +30,7 @@ export class CodexAppClient {
private activeTurns = new Map<string, PendingRequest>();
private completedTurns = new Map<string, Error | null>();
private pendingDeltas = new Map<string, PendingDelta>();
private plansByTurn = new Map<string, CodexPlanUpdate>();
/** 保存 app-server 子进程和事件出口。 */
private constructor(private child: ChildProcess, private emit: AgentEmit) {}
@@ -89,12 +90,24 @@ export class CodexAppClient {
return this.request("thread/archive", { threadId });
}
/** 返回指定线程在当前进程中收到的最新任务计划。 */
planUpdates(threadId: string) {
return [...this.plansByTurn.values()].filter((item) => item.threadId === threadId);
}
/** 清理已归档线程的任务计划缓存。 */
clearPlanUpdates(threadId: string) {
this.plansByTurn.forEach((item, turnId) => {
if (item.threadId === threadId) this.plansByTurn.delete(turnId);
});
}
/** 启动一个 Codex turn 并等待完成通知。 */
async startTurn(threadId: string, prompt: string, images: string[], onTurn?: (turnId: string) => void) {
this.currentThreadId = threadId;
const { turn } = await this.request("turn/start", { threadId, input: codexInput(prompt, images), approvalPolicy: "never" });
const turnId = turn.id;
if (!turnId) throw new Error("Codex app-server 没有返回 turn id");
this.currentThreadId = threadId;
this.currentTurnId = turnId;
onTurn?.(turnId);
const completed = this.completedTurns.get(turnId);
@@ -174,6 +187,7 @@ export class CodexAppClient {
/** 转换并广播 app-server 通知。 */
private handleNotification(method: string, params: JsonRecord) {
if (!field(params, "threadId") && this.currentThreadId && (method === "turn/started" || method === "turn/completed" || method === "turn/plan/updated")) params = { ...params, threadId: this.currentThreadId };
if (method === "item/agentMessage/delta") {
const value = params as unknown as CodexNotificationParams<"item/agentMessage/delta">;
this.textByItem.set(value.itemId, `${this.textByItem.get(value.itemId) || ""}${value.delta}`);
@@ -182,6 +196,12 @@ export class CodexAppClient {
if (method === "item/plan/delta") return this.emitDelta("plan", params as unknown as CodexNotificationParams<"item/plan/delta">);
if (method === "item/reasoning/summaryTextDelta") return this.emitDelta("reasoning", params as unknown as CodexNotificationParams<"item/reasoning/summaryTextDelta">);
if (method === "item/commandExecution/outputDelta") return this.emitDelta("command_execution", params as unknown as CodexNotificationParams<"item/commandExecution/outputDelta">);
if (method === "turn/plan/updated") {
const value = params as unknown as CodexNotificationParams<"turn/plan/updated">;
const update: CodexPlanUpdate = { ...value, threadId: value.threadId || "" };
if (update.threadId && update.turnId) this.plansByTurn.set(update.turnId, update);
params = update as unknown as JsonRecord;
}
if (method === "thread/tokenUsage/updated") {
this.lastUsage = normalizeUsage(params as unknown as CodexNotificationParams<"thread/tokenUsage/updated">);
this.emit("agent_event", { agent: "codex", type: "usage.updated", usage: this.lastUsage, ...codexEventScope(params) });
@@ -197,6 +217,12 @@ export class CodexAppClient {
if (item?.type === "agent_message" && streamedText && !item.text) item.text = streamedText;
if (id) this.textByItem.delete(id);
}
if (event.type === "turn.completed") {
const turn = field(params, "turn");
const turnId = String(field(turn, "id") || field(params, "turnId") || "");
const plan = this.plansByTurn.get(turnId);
if (plan) this.plansByTurn.set(turnId, { ...plan, turnStatus: String(field(turn, "status") || "completed") });
}
if (event.type === "turn.completed") event.usage = this.lastUsage;
this.emit("agent_event", { agent: "codex", ...event });
if (event.type === "turn.completed") {
@@ -301,7 +327,8 @@ function normalizeCodexNotification(method: string, params: JsonRecord): AgentEv
const scope = codexEventScope(params);
if (method === "thread/started") return { type: "thread.started", ...scope };
if (method === "turn/started") return { type: "turn.started", ...scope };
if (method === "turn/completed") return { type: "turn.completed", usage: null, duration_ms: field(field(params, "turn"), "durationMs"), ...scope };
if (method === "turn/completed") return { type: "turn.completed", status: field(field(params, "turn"), "status"), usage: null, duration_ms: field(field(params, "turn"), "durationMs"), ...scope };
if (method === "turn/plan/updated") return { type: "plan.updated", explanation: field(params, "explanation"), plan: field(params, "plan"), ...scope };
if (method === "item/started") return { type: "item.started", item: normalizeItem(field(params, "item")), ...scope };
if (method === "item/completed") return { type: "item.completed", item: normalizeItem(field(params, "item")), ...scope };
if (method === "error") return { type: "error", message: field(field(params, "error"), "message"), ...scope };
canvas-agent/src/agent/codex-history.ts
+36 -1
View File
@@ -1,4 +1,5 @@
import { field } from "../utils/value.js";
import type { CodexPlanUpdate } from "./codex-protocol.js";
type AgentHistoryMessage = { id: string; role: "user" | "assistant" | "tool" | "error"; title?: string; text: string; detail?: unknown; streamId?: string };
@@ -19,16 +20,24 @@ export function summarizeCodexThread(thread: unknown) {
}
/** 将 Codex turn items 转换为网页聊天历史。 */
export function threadMessages(thread: unknown): AgentHistoryMessage[] {
export function threadMessages(thread: unknown, planUpdates: CodexPlanUpdate[] = []): AgentHistoryMessage[] {
const turns = arrayValue(field(thread, "turns"));
const plansByTurn = new Map(planUpdates.map((item) => [item.turnId, item]));
const messages: AgentHistoryMessage[] = [];
turns.forEach((turn, turnIndex) => {
const turnId = String(field(turn, "id") || turnIndex);
const planMessage = structuredPlanMessage(plansByTurn.get(turnId) || { threadId: "", turnId, explanation: stringOrNull(field(turn, "explanation")), plan: arrayValue(field(turn, "plan")) as CodexPlanUpdate["plan"], turnStatus: String(field(turn, "status") || "") });
let planAdded = false;
arrayValue(field(turn, "items")).forEach((item, itemIndex) => {
const type = String(field(item, "type") || "");
const id = String(field(item, "id") || `${turnIndex}-${itemIndex}`);
if (type === "userMessage") {
const text = displayUserText(userInputText(field(item, "content")));
if (text) messages.push({ id, role: "user", text });
if (planMessage && !planAdded) {
messages.push(planMessage);
planAdded = true;
}
}
if (type === "agentMessage") {
const text = String(field(item, "text") || "").trim();
@@ -62,10 +71,36 @@ export function threadMessages(thread: unknown): AgentHistoryMessage[] {
if (type === "dynamicToolCall") messages.push({ id, role: "tool", title: "使用工具", text: "已完成工具操作", detail: { kind: "tool", status: field(item, "status") } });
if (type === "collabToolCall") messages.push({ id, role: "tool", title: "协作处理", text: "已完成协作任务", detail: { kind: "tool", status: field(item, "status") } });
});
if (planMessage && !planAdded) messages.push(planMessage);
});
return messages.filter((item) => item.text).slice(-120);
}
/** 将结构化任务计划转换为聊天进度卡片。 */
function structuredPlanMessage(update: CodexPlanUpdate): AgentHistoryMessage | null {
const tasks = arrayValue(update.plan).flatMap((item) => {
const step = String(field(item, "step") || "").trim();
return step ? [{ step, status: String(field(item, "status") || "pending") }] : [];
});
if (!tasks.length) return null;
const completed = tasks.filter((item) => item.status === "completed").length;
return {
id: `plan-${update.turnId}`,
role: "tool",
title: "任务进度",
text: `已完成 ${completed}/${tasks.length} 项`,
detail: { kind: "todo", status: planStatus(tasks, update.turnStatus), tasks, explanation: update.explanation || "" },
};
}
/** 根据步骤和 turn 状态生成任务卡片状态。 */
function planStatus(tasks: Array<{ status: string }>, turnStatus?: string) {
if (turnStatus === "failed") return "failed";
if (turnStatus === "interrupted") return "interrupted";
if (tasks.every((item) => item.status === "completed")) return "completed";
return turnStatus === "completed" ? "finished" : "inProgress";
}
/** 提取用户输入条目中的文本与附件占位信息。 */
function userInputText(content: unknown) {
return arrayValue(content)
canvas-agent/src/agent/codex-protocol.ts
+5 -2
View File
@@ -4,6 +4,8 @@ export type CodexThread = JsonRecord & { id: string; cwd: string; turns?: CodexT
export type CodexTurn = JsonRecord & { id: string; error?: CodexTurnError | null; durationMs?: number | null };
export type CodexTurnError = JsonRecord & { message: string };
export type CodexItem = JsonRecord & { id: string; type: string; text?: string };
export type CodexPlanStep = { step: string; status: "pending" | "inProgress" | "completed" };
export type CodexPlanUpdate = { threadId: string; turnId: string; explanation?: string | null; plan: CodexPlanStep[]; turnStatus?: string };
export type CodexTurnInput =
| { type: "text"; text: string; text_elements: [] }
@@ -74,8 +76,9 @@ type TokenUsageBreakdown = {
type CodexNotificationSpec = {
"thread/started": { thread: CodexThread };
"turn/started": { threadId: string; turn: CodexTurn };
"turn/completed": { threadId: string; turn: CodexTurn };
"turn/started": { threadId?: string; turn: CodexTurn };
"turn/completed": { threadId?: string; turn: CodexTurn };
"turn/plan/updated": { threadId?: string; turnId: string; explanation?: string | null; plan: CodexPlanStep[] };
"item/started": { threadId: string; turnId: string; item: CodexItem };
"item/completed": { threadId: string; turnId: string; item: CodexItem };
"item/agentMessage/delta": { threadId: string; turnId: string; itemId: string; delta: string };
canvas-agent/src/agent/codex.ts
+4 -2
View File
@@ -47,7 +47,7 @@ export async function resumeCodexThread(emit: AgentEmit, threadId: string, cwd?:
const thread = await app.resumeThread(threadId, cwd);
assertThreadWorkspace(thread, cwd);
codexThreadId = String(field(thread, "id") || threadId);
return { thread, messages: threadMessages(thread) };
return { thread, messages: threadMessages(thread, app.planUpdates(threadId)) };
}
/** 查询当前工作空间中的 Codex 线程。 */
@@ -67,6 +67,7 @@ export async function listCodexThreads(emit: AgentEmit, options: { cwd: string;
/** 读取指定 Codex 线程及其聊天历史。 */
export async function readCodexThread(emit: AgentEmit, threadId: string, cwd?: string) {
const app = await getCodexApp(emit);
let thread: unknown;
try {
thread = await loadCodexThread(emit, threadId, cwd, !unmaterializedThreadIds.has(threadId));
@@ -75,7 +76,7 @@ export async function readCodexThread(emit: AgentEmit, threadId: string, cwd?: s
unmaterializedThreadIds.add(threadId);
thread = await loadCodexThread(emit, threadId, cwd, false);
}
return { thread: summarizeCodexThread(thread), messages: threadMessages(thread) };
return { thread: summarizeCodexThread(thread), messages: threadMessages(thread, app.planUpdates(threadId)) };
}
/** 确认指定 Codex 线程属于当前工作空间。 */
@@ -88,6 +89,7 @@ export async function archiveCodexThread(emit: AgentEmit, threadId: string, cwd?
const app = await getCodexApp(emit);
await loadCodexThread(emit, threadId, cwd, false);
await app.archiveThread(threadId);
app.clearPlanUpdates(threadId);
unmaterializedThreadIds.delete(threadId);
}