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(server): surface Codex prompt hook rejections

Browse Source 
Codex can emit a blocked UserPromptSubmit hook followed by a completed
turn without accepting a user message. Capture the blocking feedback for
the current turn and report it through the existing failure timeline.

Ignore unrelated hook results and clear rejection state between turns so
corrected submissions can proceed without reloading the session.

Validated with 195 Codex adapter tests, server and CLI builds, server
typechecking, lint, and a real Codex blocked-prompt/retry check.

Refs: #6503
Co-authored-by: Scott Zhang <scottzh@fireworks.ai>
This commit is contained in:
Scott Zhang
2026-11-03 05:59:44 +00:00
committed by paseo-bot[bot]
parent f4eeb8a538
commit 470f4b3e8b
2 changed files with 276 additions and 2 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
+210
View File
@@ -32,6 +32,216 @@ import {
toAgentUsage,
} from "./codex-app-server-agent.js";
describe("Codex prompt hook rejection", () => {
test("keeps the rejection visible in the manager timeline and permits a corrected submission", async () => {
const appServer = createFakeCodexAppServer();
const provider = createProviderWithFakeAppServer(appServer);
const client: AgentClient = {
provider: "codex",
capabilities: provider.capabilities,
isAvailable: async () => true,
fetchCatalog: async () => ({ models: [], modes: [] }),
createSession: (...args) => provider.createSession(...args),
resumeSession: (...args) => provider.resumeSession(...args),
};
const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() });
const agent = await manager.resumeAgentFromPersistence(
{ provider: "codex", sessionId: "thread-1" },
createConfig({ cwd: process.cwd() }),
);
try {
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.completeTurn();
await failure;
expect(manager.getTimeline(agent.id)).toEqual([
{
type: "user_message",
text: "Disallowed prompt",
messageId: "blocked-prompt",
clientMessageId: "blocked-prompt",
},
{ type: "assistant_message", text: "[System Error] 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.");
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));
appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" });
appServer.says({ threadId: "thread-1", text: "Accepted" });
appServer.completeTurn();
await expect(retry).resolves.toMatchObject({ finalText: "Accepted", canceled: false });
expect(manager.hasInFlightRun(agent.id)).toBe(false);
appServer.assertNoErrors();
} finally {
await manager.closeAgent(agent.id);
}
});
test("reports the blocking reason instead of successful completion 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.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
expect(events).toEqual([
{
type: "turn_failed",
provider: "codex",
error: "Remove the disallowed text and retry.",
turnId: paseoTurnId,
},
]);
events.length = 0;
const replacement = await session.startTurn("Allowed replacement prompt");
appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" });
appServer.says({ threadId: "thread-1", text: "Accepted" });
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
expect(events.filter((event) => event.type.startsWith("turn_"))).toEqual([
{ type: "turn_started", provider: "codex", turnId: replacement.turnId },
{ type: "turn_completed", provider: "codex", usage: undefined, turnId: replacement.turnId },
]);
appServer.assertNoErrors();
} finally {
await session.close();
}
});
test.each([
{ eventName: "preToolUse", status: "blocked", turnId: "native-A", threadId: "thread-1" },
{
eventName: "userPromptSubmit",
status: "completed",
turnId: "native-A",
threadId: "thread-1",
},
{ eventName: "userPromptSubmit", status: "failed", turnId: "native-A", threadId: "thread-1" },
{
eventName: "userPromptSubmit",
status: "blocked",
turnId: "older-turn",
threadId: "thread-1",
},
{ eventName: "userPromptSubmit", status: "blocked", turnId: null, threadId: "thread-1" },
{
eventName: "userPromptSubmit",
status: "blocked",
turnId: "native-A",
threadId: "child-thread",
},
])(
"does not fail the foreground turn for an unrelated hook: $eventName $status $turnId $threadId",
async ({ eventName, status, turnId, threadId }) => {
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,
turnId,
run: {
eventName,
status,
entries: [{ kind: "feedback", text: "Unrelated feedback" }],
},
},
}) + "\n",
);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
expect(events).toEqual([
{ type: "turn_completed", provider: "codex", turnId: paseoTurnId, usage: undefined },
]);
appServer.assertNoErrors();
} finally {
await session.close();
}
},
);
test("provides an actionable failure when the blocking hook has no feedback", 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: [] },
},
}) + "\n",
);
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
expect(events).toEqual([
{
type: "turn_failed",
provider: "codex",
turnId: paseoTurnId,
error: "Prompt rejected by a UserPromptSubmit hook.",
},
]);
appServer.assertNoErrors();
} finally {
await session.close();
}
});
});
describe("mapCodexPlanUpdateToTodo", () => {
test("preserves checklist progress without creating a plan card", () => {
expect(
packages/server/src/server/agent/providers/codex-app-server-agent.ts
+66 -2
View File
@@ -2574,6 +2574,14 @@ const CodexEventThreadRolledBackNotificationSchema = z
type ParsedCodexNotification =
| { kind: "thread_started"; threadId: string }
| { kind: "turn_started"; turnId: string; threadId: string | null }
| {
kind: "hook_completed";
threadId: string;
turnId: string | null;
eventName: string;
status: string;
entries: Array<{ kind: string; text: string }>;
}
| {
kind: "turn_completed";
status: string;
@@ -2694,6 +2702,36 @@ function isCodexDeltaNotification(
}
const CodexNotificationSchema = z.union([
z
.object({
method: z.literal("hook/completed"),
params: z.object({
threadId: z.string(),
turnId: z.string().nullish(),
run: z.object({
eventName: z.string(),
status: z.string(),
entries: z.array(z.object({ kind: z.string(), text: z.string() })),
}),
}),
})
.transform(
({ params }): ParsedCodexNotification => ({
kind: "hook_completed",
threadId: params.threadId,
turnId: params.turnId ?? null,
eventName: params.run.eventName,
status: params.run.status,
entries: params.run.entries,
}),
),
z.object({ method: z.literal("hook/completed"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({
kind: "invalid_payload",
method,
params,
}),
),
z
.object({ method: z.literal("thread/started"), params: ThreadStartedNotificationSchema })
.transform(
@@ -3483,6 +3521,7 @@ 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>;
@@ -5778,6 +5817,9 @@ export class CodexAppServerAgentSession implements AgentSession {
case "turn_completed":
this.handleTurnCompletedNotification(parsed);
return;
case "hook_completed":
this.handleHookCompletedNotification(parsed);
return;
case "plan_updated":
this.handlePlanUpdatedNotification(parsed);
return;
@@ -6393,6 +6435,26 @@ export class CodexAppServerAgentSession implements AgentSession {
this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: current });
}
private handleHookCompletedNotification(
parsed: Extract<ParsedCodexNotification, { kind: "hook_completed" }>,
): void {
const rejectsCurrentPrompt =
parsed.eventName === "userPromptSubmit" &&
parsed.status === "blocked" &&
this.currentTurnId !== null &&
parsed.turnId === this.currentTurnId;
if (!rejectsCurrentPrompt) return;
const feedback = parsed.entries
.filter(
(entry) => entry.kind === "feedback" || entry.kind === "stop" || entry.kind === "error",
)
.map((entry) => entry.text.trim())
.filter((text) => text.length > 0)
.join("\n\n");
this.promptRejection = feedback || "Prompt rejected by a UserPromptSubmit hook.";
}
private handleTurnCompletedNotification(
parsed: Extract<ParsedCodexNotification, { kind: "turn_completed" }>,
): void {
@@ -6414,11 +6476,12 @@ export class CodexAppServerAgentSession implements AgentSession {
if (parsed.status !== "interrupted" && this.currentTurnId) {
this.terminalProcessIdsByTurnId.delete(this.currentTurnId);
}
if (parsed.status === "failed") {
// Codex completes rejected submissions successfully without accepting a user message.
if (parsed.status === "failed" || this.promptRejection !== null) {
this.emitEvent({
type: "turn_failed",
provider: CODEX_PROVIDER,
error: parsed.errorMessage ?? "Codex turn failed",
error: this.promptRejection ?? parsed.errorMessage ?? "Codex turn failed",
});
} else if (parsed.status === "interrupted") {
this.dismissInterruptedAsyncQuestions();
@@ -6443,6 +6506,7 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private resetTurnTrackingState(): void {
this.promptRejection = null;
this.latestPlanResult = null;
this.emittedItemStartedIds.clear();
this.emittedItemCompletedIds.clear();