puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-03 23:23:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): show hook rejections without failing accepted work

Browse Source 
A blocked queued submission can share a native turn with accepted input.
Show its feedback through the existing error notification instead of
replacing Codex's successful turn outcome. Keep protocol framing and
completion waits inside the existing test fixture.

Co-authored-by: Scott Zhang <scottzh@fireworks.ai>
This commit is contained in:
paseo-bot[bot]
2026-11-03 06:17:49 +00:00
co-authored by Scott Zhang
parent 470f4b3e8b
commit c679822164
3 changed files with 150 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+115 -69
View File
@@ -33,6 +33,55 @@ import {
} from "./codex-app-server-agent.js";
describe("Codex prompt hook rejection", () => {
test("shows a rejected queued follow-up without failing accepted work in the same turn", async () => {
const appServer = createFakeCodexAppServer({
"turn/steer": () => ({ turn: { id: "native-A" } }),
});
const { session, paseoTurnId } = await startPublicSteeringSession(appServer);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await session.steerActiveTurn!("Allowed follow-up", { expectedTurnId: paseoTurnId });
await session.steerActiveTurn!("Rejected follow-up", { expectedTurnId: paseoTurnId });
appServer.completesHook({
threadId: "thread-1",
turnId: "native-A",
eventName: "userPromptSubmit",
status: "completed",
entries: [],
});
appServer.completesHook({
threadId: "thread-1",
turnId: "native-A",
eventName: "userPromptSubmit",
status: "blocked",
entries: [{ kind: "feedback", text: "Follow-up rejected" }],
});
appServer.says({ threadId: "thread-1", text: "Accepted work completed" });
const completion = waitForNextTurnCompletion(session);
appServer.completeTurn();
await expect(completion).resolves.toMatchObject({ type: "turn_completed" });
expect(events).toContainEqual(
expect.objectContaining({
type: "timeline",
item: expect.objectContaining({
type: "assistant_message",
text: "Accepted work completed",
}),
}),
);
expect(events).toContainEqual({
type: "timeline",
provider: "codex",
item: { type: "notification", level: "error", message: "Follow-up rejected" },
turnId: paseoTurnId,
});
appServer.assertNoErrors();
} finally {
await session.close();
}
});
test("keeps the rejection visible in the manager timeline and permits a corrected submission", async () => {
const appServer = createFakeCodexAppServer();
const provider = createProviderWithFakeAppServer(appServer);
@@ -53,25 +102,17 @@ describe("Codex prompt hook rejection", () => {
const run = manager.runAgent(agent.id, "Disallowed prompt", {
clientMessageId: "blocked-prompt",
});
const failure = expect(run).rejects.toThrow("Remove the disallowed text and retry.");
await appServer.waitForTurnStart();
appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" });
appServer.child.stdout.write(
JSON.stringify({
method: "hook/completed",
params: {
threadId: "thread-1",
turnId: "native-A",
run: {
eventName: "userPromptSubmit",
status: "blocked",
entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }],
},
},
}) + "\n",
);
appServer.completesHook({
threadId: "thread-1",
turnId: "native-A",
eventName: "userPromptSubmit",
status: "blocked",
entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }],
});
appServer.completeTurn();
await failure;
await expect(run).resolves.toMatchObject({ finalText: "", canceled: false });
expect(manager.getTimeline(agent.id)).toEqual([
{
@@ -80,18 +121,15 @@ describe("Codex prompt hook rejection", () => {
messageId: "blocked-prompt",
clientMessageId: "blocked-prompt",
},
{ type: "assistant_message", text: "[System Error] Remove the disallowed text and retry." },
{ type: "notification", level: "error", message: "Remove the disallowed text and retry." },
]);
expect(manager.hasInFlightRun(agent.id)).toBe(false);
expect(manager.getAgent(agent.id)?.lastError).toBe("Remove the disallowed text and retry.");
expect(manager.getAgent(agent.id)?.lastError).toBeUndefined();
const retry = manager.runAgent(agent.id, "Corrected prompt", {
clientMessageId: "corrected-prompt",
});
function isTurnStart(request: Record<string, unknown>): boolean {
return request.method === "turn/start";
}
await vi.waitFor(() => expect(appServer.requests().filter(isTurnStart)).toHaveLength(2));
await appServer.waitForTurnStart(2);
appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" });
appServer.says({ threadId: "thread-1", text: "Accepted" });
appServer.completeTurn();
@@ -103,35 +141,39 @@ describe("Codex prompt hook rejection", () => {
}
});
test("reports the blocking reason instead of successful completion and accepts the next turn", async () => {
test("shows blocking feedback and accepts the next turn", async () => {
const appServer = createFakeCodexAppServer();
const { session, paseoTurnId } = await startPublicSteeringSession(appServer);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
appServer.child.stdout.write(
JSON.stringify({
method: "hook/completed",
params: {
threadId: "thread-1",
turnId: "native-A",
run: {
eventName: "userPromptSubmit",
status: "blocked",
entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }],
},
},
}) + "\n",
);
appServer.completesHook({
threadId: "thread-1",
turnId: "native-A",
eventName: "userPromptSubmit",
status: "blocked",
entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }],
});
const completion = waitForNextTurnCompletion(session);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
await completion;
expect(events).toEqual([
{
type: "turn_failed",
type: "timeline",
provider: "codex",
error: "Remove the disallowed text and retry.",
item: {
type: "notification",
level: "error",
message: "Remove the disallowed text and retry.",
},
turnId: paseoTurnId,
},
{
type: "turn_completed",
provider: "codex",
usage: undefined,
turnId: paseoTurnId,
},
]);
@@ -140,8 +182,9 @@ describe("Codex prompt hook rejection", () => {
const replacement = await session.startTurn("Allowed replacement prompt");
appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" });
appServer.says({ threadId: "thread-1", text: "Accepted" });
const replacementCompletion = waitForNextTurnCompletion(session);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
await replacementCompletion;
expect(events.filter((event) => event.type.startsWith("turn_"))).toEqual([
{ type: "turn_started", provider: "codex", turnId: replacement.turnId },
@@ -183,22 +226,16 @@ describe("Codex prompt hook rejection", () => {
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
appServer.child.stdout.write(
JSON.stringify({
method: "hook/completed",
params: {
threadId,
turnId,
run: {
eventName,
status,
entries: [{ kind: "feedback", text: "Unrelated feedback" }],
},
},
}) + "\n",
);
appServer.completesHook({
threadId,
turnId,
eventName,
status,
entries: [{ kind: "feedback", text: "Unrelated feedback" }],
});
const completion = waitForNextTurnCompletion(session);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
await completion;
expect(events).toEqual([
{ type: "turn_completed", provider: "codex", turnId: paseoTurnId, usage: undefined },
]);
@@ -215,24 +252,32 @@ describe("Codex prompt hook rejection", () => {
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
appServer.child.stdout.write(
JSON.stringify({
method: "hook/completed",
params: {
threadId: "thread-1",
turnId: "native-A",
run: { eventName: "userPromptSubmit", status: "blocked", entries: [] },
},
}) + "\n",
);
appServer.completesHook({
threadId: "thread-1",
turnId: "native-A",
eventName: "userPromptSubmit",
status: "blocked",
entries: [],
});
const completion = waitForNextTurnCompletion(session);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
await completion;
expect(events).toEqual([
{
type: "turn_failed",
type: "timeline",
provider: "codex",
item: {
type: "notification",
level: "error",
message: "Prompt rejected by a UserPromptSubmit hook.",
},
turnId: paseoTurnId,
},
{
type: "turn_completed",
provider: "codex",
usage: undefined,
turnId: paseoTurnId,
error: "Prompt rejected by a UserPromptSubmit hook.",
},
]);
appServer.assertNoErrors();
@@ -301,6 +346,7 @@ import {
fakeCodexThreadHistory,
type FakeCodexAppServer,
waitForNextEvent,
waitForNextTurnCompletion,
waitForNextPermission,
waitForNextTimelineItem,
waitForProviderSubagent,
packages/server/src/server/agent/providers/codex-app-server-agent.ts
+11 -6
View File
@@ -3521,7 +3521,6 @@ export class CodexAppServerAgentSession implements AgentSession {
// in this app-server, so no other app-server can resume it and it is not reported.
private savedThreadId: string | null = null;
private currentTurnId: string | null = null;
private promptRejection: string | null = null;
private pendingForegroundTurnIdentification: {
foregroundTurnId: string;
promise: Promise<string | null>;
@@ -6452,7 +6451,15 @@ export class CodexAppServerAgentSession implements AgentSession {
.map((entry) => entry.text.trim())
.filter((text) => text.length > 0)
.join("\n\n");
this.promptRejection = feedback || "Prompt rejected by a UserPromptSubmit hook.";
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: {
type: "notification",
level: "error",
message: feedback || "Prompt rejected by a UserPromptSubmit hook.",
},
});
}
private handleTurnCompletedNotification(
@@ -6476,12 +6483,11 @@ export class CodexAppServerAgentSession implements AgentSession {
if (parsed.status !== "interrupted" && this.currentTurnId) {
this.terminalProcessIdsByTurnId.delete(this.currentTurnId);
}
// Codex completes rejected submissions successfully without accepting a user message.
if (parsed.status === "failed" || this.promptRejection !== null) {
if (parsed.status === "failed") {
this.emitEvent({
type: "turn_failed",
provider: CODEX_PROVIDER,
error: this.promptRejection ?? parsed.errorMessage ?? "Codex turn failed",
error: parsed.errorMessage ?? "Codex turn failed",
});
} else if (parsed.status === "interrupted") {
this.dismissInterruptedAsyncQuestions();
@@ -6506,7 +6512,6 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private resetTurnTrackingState(): void {
this.promptRejection = null;
this.latestPlanResult = null;
this.emittedItemStartedIds.clear();
this.emittedItemCompletedIds.clear();
packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts
+24 -6
View File
@@ -56,6 +56,13 @@ interface FakeLegacyPatch {
path: string;
diff: string;
}
interface FakeHookCompletion {
threadId: string;
turnId: string | null;
eventName: string;
status: string;
entries: Array<{ kind: string; text: string }>;
}
type CodexAppServerChildProcess = ChildProcessWithoutNullStreams & {
stdin: PassThrough;
stdout: PassThrough;
@@ -67,13 +74,14 @@ export interface FakeCodexAppServer {
readonly recordedRollbacks: JsonObject[];
requests(): readonly JsonObject[];
assertNoErrors(): void;
waitForTurnStart(): Promise<JsonObject>;
waitForTurnStart(occurrence?: number): Promise<JsonObject>;
waitForRequest(method: string): Promise<JsonObject>;
disconnect(): void;
nextResponse(): Promise<string>;
startsTurn(params: { threadId: string; turnId?: string }): void;
startsCompaction(params: { threadId: string; itemId: string }): void;
updatesPlan(params: { threadId: string; steps: string[] }): void;
completesHook(params: FakeHookCompletion): void;
completeTurn(params?: {
threadId?: string;
status?: "completed" | "failed" | "interrupted";
@@ -368,9 +376,11 @@ export function createFakeCodexAppServer(
throw errors[0];
}
},
async waitForTurnStart() {
async waitForTurnStart(occurrence = 1) {
const message = await waitForMessage(
(candidate) => candidate.method === "turn/start",
(candidate) =>
messages.filter((request) => request.method === "turn/start")[occurrence - 1] ===
candidate,
"turn start request",
);
return toJsonObject(message.params);
@@ -418,6 +428,9 @@ export function createFakeCodexAppServer(
})}\n`,
);
},
completesHook({ threadId, turnId, ...run }) {
writeNotification("hook/completed", { threadId, turnId, run });
},
completeTurn(params = {}) {
child.stdout.write(
`${JSON.stringify({
@@ -689,16 +702,17 @@ type StreamEventOfType<TType extends StreamEventType> = Extract<AgentStreamEvent
export function waitForNextEvent<TType extends StreamEventType>(
session: AgentSession,
type: TType,
type: TType | readonly TType[],
accepts?: (event: StreamEventOfType<TType>) => boolean,
): Promise<StreamEventOfType<TType>> {
const types: readonly StreamEventType[] = typeof type === "string" ? [type] : type;
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error(`Timed out waiting for ${type}`));
reject(new Error(`Timed out waiting for ${types.join(" or ")}`));
}, 1000);
const unsubscribe = session.subscribe((event) => {
if (event.type !== type) {
if (!types.includes(event.type)) {
return;
}
const typedEvent = event as StreamEventOfType<TType>;
@@ -712,6 +726,10 @@ export function waitForNextEvent<TType extends StreamEventType>(
});
}
export function waitForNextTurnCompletion(session: AgentSession) {
return waitForNextEvent(session, ["turn_completed", "turn_failed", "turn_canceled"]);
}
type TimelineEvent = StreamEventOfType<"timeline">;
type ProviderSubagentEvent = StreamEventOfType<"provider_subagent">;