Skip to content

fix: surface async user input events #496

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jbulpitt wants to merge 1 commit into
base: main
Choose a base branch
from
+147 −0
Open

fix: surface async user input events #496

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions src/CodexEventHandler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,7 @@ export class CodexEventHandler {
private planUpdateChain: Promise<void> = Promise.resolve();
private disposed = false;
private readonly seenReasoningDeltaItemIds = new Set<string>();
private readonly seenAgentMessageDeltaItemIds = new Set<string>();
private readonly terminalCommandIds = new Set<string>();
private readonly terminalCommandOutputIds = new Set<string>();
private readonly agentMessagePhases = new Map<string, string | null>();
Expand Down Expand Up @@ -448,6 +449,7 @@ export class CodexEventHandler {
this.pendingPlanItemIds.clear();
this.planDeltaTextByItemId.clear();
this.lastEmittedPlanTextByItemId.clear();
this.seenAgentMessageDeltaItemIds.clear();
}

private async createUpdateEvent(notification: ServerNotification): Promise<UpdateSessionEvent | null> {
Expand Down Expand Up @@ -623,6 +625,7 @@ export class CodexEventHandler {
}

private async createTextEvent(event: AgentMessageDeltaNotification): Promise<UpdateSessionEvent> {
this.seenAgentMessageDeltaItemIds.add(event.itemId);
const phase = this.agentMessagePhases.get(event.itemId) ?? null;
return createAgentTextMessageChunk(event.delta, event.itemId, createCodexMessagePhaseMeta(phase));
}
Expand Down Expand Up @@ -802,6 +805,26 @@ export class CodexEventHandler {
return this.subagents.legacyCollaborationCompleted(event.item);
case "agentMessage":
this.rememberAgentMessagePhase(event.item);
if (event.item.delivery === "async" && event.item.questions !== null) {
const streamed = this.seenAgentMessageDeltaItemIds.delete(event.item.id);
return createAgentTextMessageChunk(
streamed ? "" : event.item.text,
event.item.id,
{
codex: {
...(event.item.phase ? {phase: event.item.phase} : {}),
asyncUserInput: {
delivery: event.item.delivery,
threadId: event.threadId,
turnId: event.turnId,
itemId: event.item.id,
questions: event.item.questions,
},
},
},
);
}
this.seenAgentMessageDeltaItemIds.delete(event.item.id);
return null;
case "plan": {
const deltaText = this.planDeltaTextByItemId.get(event.item.id) ?? "";
Expand Down
124 changes: 124 additions & 0 deletions src/__tests__/CodexACPAgent/agent-message-events.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,4 +86,128 @@ describe("CodexEventHandler - agent message events", () => {
"data/agent-message-phases.json"
);
});

it("publishes async user input metadata once without repeating streamed question text", async () => {
const question = {
type: "agentMessage" as const,
id: "async-question-streamed",
text: "Choose a lane\n- Stable\n- Preview",
phase: "final_answer" as const,
memoryCitation: null,
delivery: "async" as const,
questions: [{title: "Choose a lane", options: ["Stable", "Preview"]}],
};
const notifications: ServerNotification[] = [
{
method: "item/started",
params: {threadId: sessionId, turnId: "turn-async", startedAtMs: 1, item: question},
},
{
method: "item/agentMessage/delta",
params: {
threadId: sessionId,
turnId: "turn-async",
itemId: question.id,
delta: question.text,
},
},
{
method: "item/completed",
params: {threadId: sessionId, turnId: "turn-async", completedAtMs: 2, item: question},
},
];

await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, notifications);

expect(mockFixture.getAcpConnectionEvents([]).filter(event => event.method === "sessionUpdate"))
.toEqual([
{
method: "sessionUpdate",
args: [{
sessionId,
update: {
sessionUpdate: "agent_message_chunk",
messageId: question.id,
content: {type: "text", text: question.text},
_meta: {codex: {phase: "final_answer"}},
},
}],
},
{
method: "sessionUpdate",
args: [{
sessionId,
update: {
sessionUpdate: "agent_message_chunk",
messageId: question.id,
content: {type: "text", text: ""},
_meta: {
codex: {
phase: "final_answer",
asyncUserInput: {
delivery: "async",
threadId: sessionId,
turnId: "turn-async",
itemId: question.id,
questions: [{title: "Choose a lane", options: ["Stable", "Preview"]}],
},
},
},
},
}],
},
]);
});

it("publishes a question-only completed async item when no text delta arrived", async () => {
const question = {
type: "agentMessage" as const,
id: "async-question-completed-only",
text: "What should change?",
phase: "final_answer" as const,
memoryCitation: null,
delivery: "async" as const,
questions: [{title: "What should change?", options: null}],
};
await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, [
{
method: "item/started",
params: {threadId: sessionId, turnId: "turn-completed-only", startedAtMs: 2, item: question},
},
{
method: "item/completed",
params: {
threadId: sessionId,
turnId: "turn-completed-only",
completedAtMs: 3,
item: question,
},
},
]);

expect(mockFixture.getAcpConnectionEvents([]).filter(event => event.method === "sessionUpdate"))
.toEqual([{
method: "sessionUpdate",
args: [{
sessionId,
update: {
sessionUpdate: "agent_message_chunk",
messageId: question.id,
content: {type: "text", text: "What should change?"},
_meta: {
codex: {
phase: "final_answer",
asyncUserInput: {
delivery: "async",
threadId: sessionId,
turnId: "turn-completed-only",
itemId: question.id,
questions: [{title: "What should change?", options: null}],
},
},
},
},
}],
}]);
});
});