mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 保留 completed agentMessage 事件 (#56)

Browse Source 
Co-authored-by: Codex <codex@pikas.tech>
This commit is contained in:
Lyon
2026-06-02 07:09:12 +08:00
committed by GitHub
parent 5544db96fb
commit 719584e2ce
5 changed files with 52 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/codex-stdio.ts
+35 -7
View File
@@ -57,6 +57,11 @@ interface PendingRequest {
reject: (error: Error) => void;
}
interface CompletedAssistantMessage {
itemId: string | null;
text: string;
}
interface CodexStdioCloseInfo extends JsonRecord {
code: number | null;
signal: string | null;
@@ -376,7 +381,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
return { terminalStatus: cancelled.status, failureKind: cancelled.failureKind, failureMessage: cancelled.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })) };
}
let assistantText = "";
let finalAssistantText = "";
const completedAssistantMessages: CompletedAssistantMessage[] = [];
let threadId: string | undefined = options.threadId;
let turnId: string | undefined;
let terminal: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } | null = null;
@@ -404,7 +409,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
if (normalized.threadId) threadId = normalized.threadId;
if (normalized.turnId) turnId = normalized.turnId;
if (normalized.assistantDelta) assistantText += normalized.assistantDelta;
if (typeof normalized.assistantFinal === "string" && normalized.assistantFinal.trim().length > 0) finalAssistantText = normalized.assistantFinal;
if (normalized.completedAssistantMessage) completedAssistantMessages.push(normalized.completedAssistantMessage);
emitEvents(normalized.events);
if (normalized.terminal && !terminal) {
terminal = normalized.terminal;
@@ -471,9 +476,8 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
}
if (!terminal) terminal = { status: "failed", failureKind: "backend-response-invalid", message: "codex app-server finished without terminal status" };
if (terminal.status !== "completed") emitEvents(await session.close());
const reply = finalAssistantText.trim().length > 0 ? finalAssistantText : assistantText;
if (reply.trim().length > 0) events.push({ type: "assistant_message", payload: { text: reply } });
events.push({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
emitEvents(assistantMessageEventsForTurn(completedAssistantMessages, assistantText, terminal.status === "completed"));
emitEvent({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
await liveEventWrite;
return { terminalStatus: terminal.status, failureKind: terminal.failureKind, failureMessage: terminal.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })), ...(threadId ? { threadId } : {}), ...(turnId ? { turnId } : {}) };
}
@@ -534,7 +538,7 @@ function codexHomeReadiness(codexHome: string): BackendTurnResult | null {
};
}
function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent[]; assistantDelta?: string; assistantFinal?: string; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } {
function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent[]; assistantDelta?: string; completedAssistantMessage?: CompletedAssistantMessage; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } {
const method = typeof message.method === "string" ? message.method : "unknown";
const params = asRecordAt(message, "params");
if (method === "thread/started") {
@@ -551,9 +555,10 @@ function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent
const item = asRecordAt(params, "item");
const itemId = stringAt(item, "id") ?? stringAt(params, "itemId");
const text = method === "item/completed" ? agentMessageText(item) : "";
const completedAssistantMessage = text.trim().length > 0 ? { itemId: itemId ?? null, text } : undefined;
return {
events: [{ type: "backend_status", payload: { phase: method === "item/completed" ? "item/agentMessage:completed" : "item/agentMessage:started", itemId, textBytes: Buffer.byteLength(text, "utf8") } }],
...(text.trim().length > 0 ? { assistantFinal: text } : {}),
...(completedAssistantMessage ? { completedAssistantMessage } : {}),
};
}
if (method === "item/started" || method === "item/completed") return { events: [{ type: "tool_call", payload: toolCallPayload(method, asRecordAt(params, "item")) }] };
@@ -580,6 +585,29 @@ function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent
return { events: [{ type: "backend_status", payload: { phase: method } }] };
}
function assistantMessageEventsForTurn(completedMessages: CompletedAssistantMessage[], assistantDeltaText: string, completed: boolean): BackendEvent[] {
const messages = completedMessages.length > 0
? completedMessages.map((message) => ({ ...message, source: "completed-agent-message" }))
: assistantDeltaText.trim().length > 0
? [{ itemId: null, text: assistantDeltaText, source: "agent-message-delta-fallback" }]
: [];
return messages.map((message, index) => {
const replyAuthority = completed && index === messages.length - 1;
return {
type: "assistant_message",
payload: {
text: message.text,
itemId: message.itemId,
source: message.source,
messageIndex: index + 1,
messageCount: messages.length,
replyAuthority,
final: replyAuthority,
},
};
});
}
function terminalStatusFromValue(value: unknown): TerminalStatus {
if (value === "completed") return "completed";
if (value === "cancelled" || value === "canceled" || value === "interrupted") return "cancelled";
src/mgr/result.ts
+7 -4
View File
@@ -90,11 +90,14 @@ function messageFromEvents(events: RunEvent[]): string | null {
}
function assistantReply(events: RunEvent[]): string {
return events
.filter((event) => event.type === "assistant_message")
const assistantEvents = events.filter((event) => event.type === "assistant_message");
const authoritative = assistantEvents
.filter((event) => event.payload.replyAuthority === true || event.payload.final === true)
.map((event) => textPayload(event.payload))
.filter((text) => text.length > 0)
.join("");
.filter((text) => text.length > 0);
const latestAuthoritative = authoritative.at(-1);
if (latestAuthoritative) return latestAuthoritative;
return assistantEvents.map((event) => textPayload(event.payload)).filter((text) => text.length > 0).join("");
}
function textPayload(payload: JsonRecord): string {
src/selftest/cases/30-codex-stdio.ts
+7 -2
View File
@@ -65,8 +65,13 @@ const selfTest: SelfTestCase = async (context) => {
assert.equal(finalMessageEnvelope.reply, "Final answer only.", "result reply should use the final completed agentMessage instead of concatenating progress deltas");
const finalMessageEvents = await client.get(`/api/v1/runs/${finalMessage.runId}/events?afterSeq=0&limit=100`) as { items?: Array<{ type: string; payload: unknown }> };
const assistantEvents = finalMessageEvents.items?.filter((event) => event.type === "assistant_message") ?? [];
assert.equal(assistantEvents.length, 1, "backend should emit one final assistant_message event");
assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).text, "Final answer only.");
assert.equal(assistantEvents.length, 2, "backend should preserve each completed agentMessage as assistant_message event");
assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).text, "I am checking the workspace.");
assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).itemId, "msg_progress");
assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).replyAuthority, false);
assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).text, "Final answer only.");
assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).itemId, "msg_final");
assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).replyAuthority, true);
const staleResume = await createStaleThreadRun(client, context);
const staleResumeResult = await runOnce({