Skip to content

return promptRequired when no turn is running #363

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
lifeizhou-ap wants to merge 2 commits into
base: main
Choose a base branch
from
Open

return promptRequired when no turn is running #363

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fbb7f6a
return promptRequired when no turn is running
lifeizhou-ap Aug 4, 2026
bfd5e7d
renamed function
lifeizhou-ap Aug 4, 2026
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
72 changes: 55 additions & 17 deletions examples/steering.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,23 @@ const steeringPrompt = [
type SteeringRequest = {
sessionId: acp.SessionId;
prompt: acp.ContentBlock[];
_meta?: {
steering?: {
idleBehavior?: "promptRequired";
};
};
};

type SteeringResponse = {
outcome: "injected" | "startedNewTurn";
};
/**
* `startedNewTurn` is the default idle behavior when `promptRequired` is not
* requested. This example requests `promptRequired`, so adapters that honor
* the option return `promptRequired` instead. The `startedNewTurn` branch
* remains for compatibility with adapters that do not honor the opt-in.
*/
type SteeringResponse =
| {outcome: "injected"}
| {outcome: "promptRequired"; reason: "noRunningTurn"}
| {outcome: "startedNewTurn"};

type ThreadStatusType = "active" | "idle" | "systemError";
type StateListener = () => void;
Expand Down Expand Up @@ -273,19 +285,31 @@ function printBanner(text: string): void {
lastChannel = null;
}

function printSummary(clueCount: number, cluesAtSteer: number, stopReason: string, steered: boolean): void {
function printSummary(
clueCount: number,
cluesAtSteer: number,
stopReason: string,
steeringOutcome: SteeringResponse["outcome"] | null,
): void {
const line = "─".repeat(66);
const stoppedEarly = toolCallsSeen.size < clueCount;
console.log(`\n\n${c.bold(line)}`);
console.log(c.bold(" Summary"));
console.log(line);
console.log(` tool calls total : ${toolCallsSeen.size} of up to ${clueCount} clues`);
console.log(` steered after : ${steered ? `${cluesAtSteer} clue(s)` : "not steered"}`);
console.log(` attempted after : ${steeringOutcome ? `${cluesAtSteer} clue(s)` : "not attempted"}`);
console.log(` steering outcome : ${steeringOutcome ?? "not sent"}`);
console.log(` stop reason : ${stopReason}`);
console.log(line);
if (!steered) {
if (steeringOutcome === null) {
console.log(c.yellow(" • The turn finished before we could steer. Lower STEER_AFTER_TOOL_CALLS"));
console.log(c.yellow(" or use a slower model to catch the turn while it is still running."));
} else if (steeringOutcome === "promptRequired") {
console.log(c.green(" ✔ The turn ended before the adapter could apply the message, so the"));
console.log(c.green(" client submitted the same message through session/prompt."));
} else if (steeringOutcome === "startedNewTurn") {
console.log(c.yellow(" • The adapter started a new turn with the message, so the client did"));
console.log(c.yellow(" not submit the message again."));
} else if (stoppedEarly) {
console.log(c.green(" ✔ The agent stopped BEFORE reading every clue — the steering message"));
console.log(c.green(" was picked up mid-turn and changed its course."));
Expand Down Expand Up @@ -387,6 +411,7 @@ async function main(): Promise<void> {
promptDone = true;
notifyStateListeners();
});
let promptResponsePromise = promptPromise;
promptPromise.catch(() => {});

// Let the agent work through a couple of clues, then steer mid-turn.
Expand All @@ -400,33 +425,46 @@ async function main(): Promise<void> {

const cluesAtSteer = toolCallsSeen.size;
const turnAlreadyFinished = promptDone || finishedTransitions > 0;
let steered = false;
let steeringOutcome: SteeringResponse["outcome"] | null = null;

if (turnAlreadyFinished) {
writeEvent(c.red("⚠ The turn finished before we could steer — skipping the steering step."));
} else {
steered = true;
printBanner(`Injecting steering message after ${cluesAtSteer} clue(s)`);
process.stdout.write(`${c.magenta(`✋ steer → ${steeringPrompt}`)}\n`);
lastChannel = null;

const steeringResponse = await agent.request<SteeringResponse, SteeringRequest>(STEERING_METHOD, {
sessionId: trackedSessionId,
prompt: [{type: "text", text: steeringPrompt}],
_meta: {steering: {idleBehavior: "promptRequired"}},
});
if (steeringResponse.outcome !== "injected" && steeringResponse.outcome !== "startedNewTurn") {
throw new Error(`Unexpected steering response: ${JSON.stringify(steeringResponse)}`);
}
steeringOutcome = steeringResponse.outcome;
writeEvent(c.magenta(c.bold(` outcome: ${steeringResponse.outcome}`)));
if (steeringResponse.outcome === "injected") {
writeEvent(c.dim(" → injected into the running turn; the agent picks it up at its next step."));
} else {
writeEvent(c.dim(" → the turn had already ended, so this started a fresh turn."));
switch (steeringResponse.outcome) {
case "injected":
writeEvent(c.dim(" → the adapter added the message to the running turn."));
break;
case "promptRequired":
if (steeringResponse.reason !== "noRunningTurn") {
throw new Error(`Unexpected steering response: ${JSON.stringify(steeringResponse)}`);
}
writeEvent(c.dim(" → the turn ended before the adapter could apply the message; the client is now submitting it through session/prompt."));
promptResponsePromise = agent.request(acp.methods.agent.session.prompt, {
sessionId: trackedSessionId,
prompt: [{type: "text", text: steeringPrompt}],
});
break;
case "startedNewTurn":
writeEvent(c.dim(" → the adapter started a new turn with the message; the client will not submit it again."));
break;
default:
throw new Error(`Unexpected steering response: ${JSON.stringify(steeringResponse)}`);
}
}

const promptResponse = await promptPromise;
printSummary(clueCount, cluesAtSteer, promptResponse.stopReason, steered);
const promptResponse = await promptResponsePromise;
printSummary(clueCount, cluesAtSteer, promptResponse.stopReason, steeringOutcome);

await agent.request(acp.methods.agent.session.close, {
sessionId: trackedSessionId,
Expand Down
17 changes: 14 additions & 3 deletions src/AcpExtensions.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,14 +84,25 @@ export async function legacySetSessionModel(
return await connection.request<LegacySetSessionModelResponse, LegacySetSessionModelRequest>(LEGACY_SET_SESSION_MODEL_METHOD, params);
}

export type SessionSteerMeta = {
[key: string]: unknown;
steering?: {
[key: string]: unknown;
idleBehavior?: "promptRequired";
};
}

export type SessionSteerRequest = {
sessionId: SessionId;
prompt: ContentBlock[];
_meta?: SessionSteerMeta | null;
}

export type SessionSteeringResponse = {
outcome: "injected" | "startedNewTurn" | "failed";
}
export type SessionSteeringResponse =
| {outcome: "injected"}
| {outcome: "startedNewTurn"}
| {outcome: "failed"}
| {outcome: "promptRequired"; reason: "noRunningTurn"};

export type SessionSteeringExtRequest = {
method: typeof SESSION_STEERING_METHOD;
Expand Down
68 changes: 54 additions & 14 deletions src/CodexAcpServer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,32 @@ import {
const IMPLEMENT_PLAN_OPTION_ID = "implement_plan";
const REVISE_PLAN_OPTION_ID = "revise_plan";

function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}

function parseSessionSteerMeta(value: unknown): SessionSteerRequest["_meta"] {
if (value === undefined || value === null) {
return value;
}
if (!isRecord(value)) {
throw RequestError.invalidParams();
}

const steering = value["steering"];
if (steering === undefined) {
return value as SessionSteerRequest["_meta"];
}
if (
!isRecord(steering) ||
(steering["idleBehavior"] !== undefined && steering["idleBehavior"] !== "promptRequired")
) {
throw RequestError.invalidParams();
}

return value as SessionSteerRequest["_meta"];
}

export interface SessionState {
sessionId: string,
currentModelId: string,
Expand Down Expand Up @@ -898,8 +924,9 @@ export class CodexAcpServer {
* check guards against deleting a queue a later request has since reused).
*
* @param params The target session id and the prompt to steer with.
* @returns Whether the prompt joined the active turn ("injected"), started a
* new one ("startedNewTurn"), or could not be applied ("failed"); see
* @returns Whether the prompt joined the active turn ("injected"), requires
* a normal prompt ("promptRequired"), started a new one
* ("startedNewTurn"), or could not be applied ("failed"); see
* {@link performSteeringRequest}.
*/
async executeOrQueueSteeringRequest(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
Expand Down Expand Up @@ -937,11 +964,13 @@ export class CodexAcpServer {

/**
* Delivers a steering prompt to the session: injects it into the live turn
* when there is one, otherwise starts a new turn.
* when there is one, otherwise either asks the client to send a normal
* prompt or starts a new turn.
*
* @param params The target session id and the prompt to steer with.
* @returns "injected" when the prompt joined an existing turn, otherwise the
* outcome of starting a new turn.
* @returns "injected" when the prompt joined an existing turn,
* "promptRequired" when the opted-in client must send a normal prompt,
* otherwise the outcome of starting a new turn.
*/
private async performSteeringRequest(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
logger.log("Steering session requested", {
Expand All @@ -959,6 +988,10 @@ export class CodexAcpServer {
return {outcome: "injected"};
}
}
if (params._meta?.steering?.idleBehavior === "promptRequired") {
await this.waitForSessionToBeReadyForPrompt(params.sessionId);
return {outcome: "promptRequired", reason: "noRunningTurn"};
}
return await this.startNewTurnFromSteering(params);
}

Expand Down Expand Up @@ -1019,15 +1052,7 @@ export class CodexAcpServer {
* fails or is cancelled before the turn starts.
*/
private async startNewTurnFromSteering(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
// A prompt can outlive its turn (post-turn cleanup runs before it leaves
// activePrompts), so a steer can miss the turn while the prompt is still
// winding down. Starting a new turn now would run a second prompt on the
// same session, so wait for the current one to drain first (a no-op when idle).
const previousPrompt = this.activePrompts.get(params.sessionId);
await previousPrompt?.completion;
if (this.sessionIsClosing(params.sessionId)) {
throw RequestError.invalidRequest(`Session ${params.sessionId} is closing`);
}
await this.waitForSessionToBeReadyForPrompt(params.sessionId);

return await new Promise<SessionSteeringResponse>((resolve, reject) => {
let turnStarted = false;
Expand Down Expand Up @@ -1069,6 +1094,18 @@ export class CodexAcpServer {
});
}

private async waitForSessionToBeReadyForPrompt(sessionId: SessionId): Promise<void> {
// A prompt can outlive its turn (post-turn cleanup runs before it leaves
// activePrompts), so a steer can miss the turn while the prompt is still
// winding down. Starting a new turn now would run a second prompt on the
// same session, so wait for the current one to drain first (a no-op when idle).
const previousPrompt = this.activePrompts.get(sessionId);
await previousPrompt?.completion;
if (this.sessionIsClosing(sessionId)) {
throw RequestError.invalidRequest(`Session ${sessionId} is closing`);
}
}

private isNoActiveTurnToSteerError(error: unknown): boolean {
const messages = error instanceof Error ? [error.message] : [];
if (typeof error === "object" && error !== null && "data" in error) {
Expand Down Expand Up @@ -1103,12 +1140,15 @@ export class CodexAcpServer {
private parseSessionSteerParams(params: Record<string, unknown>): SessionSteerRequest {
const sessionId = params["sessionId"];
const prompt = params["prompt"];
const meta = parseSessionSteerMeta(params["_meta"]);
if (typeof sessionId !== "string" || !Array.isArray(prompt)) {
throw RequestError.invalidParams();
}

return {
sessionId: sessionId,
prompt: prompt as acp.ContentBlock[],
...(meta === undefined ? {} : {_meta: meta}),
};
}

Expand Down
61 changes: 59 additions & 2 deletions src/__tests__/CodexACPAgent/steer-events.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,14 @@ function deferred<T>(): {promise: Promise<T>, resolve: (value: T) => void} {
function startActiveTurn(sessionOverrides?: Partial<SessionState>) {
const mockFixture = createCodexMockTestFixture();
const sessionState = createTestSessionState(sessionOverrides);
vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart").mockResolvedValue({
const turnStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart").mockResolvedValue({
turn: createTurn("turn-id", "inProgress"),
});
const turnCompleted = deferred<TurnCompletedNotification>();
vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted")
.mockReturnValue(turnCompleted.promise);
vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState);
return {mockFixture, sessionState, turnCompleted};
return {mockFixture, sessionState, turnCompleted, turnStartSpy};
}

describe('_session/steering', () => {
Expand All @@ -65,6 +65,7 @@ describe('_session/steering', () => {
await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, {
sessionId: "session-id",
prompt: [{type: "text", text: "also keep backward compatibility"}],
_meta: {steering: {idleBehavior: "promptRequired"}},
})).resolves.toEqual({outcome: "injected"});

expect(turnSteerSpy).toHaveBeenCalledWith({
Expand Down Expand Up @@ -109,6 +110,21 @@ describe('_session/steering', () => {
});
});

it('reports promptRequired without starting a turn when opted in and idle', async () => {
const mockFixture = createCodexMockTestFixture();
const sessionState = createTestSessionState();
vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState);
const turnStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart");

await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, {
sessionId: "session-id",
prompt: [{type: "text", text: "send this as a normal prompt"}],
_meta: {steering: {idleBehavior: "promptRequired"}},
})).resolves.toEqual({outcome: "promptRequired", reason: "noRunningTurn"});

expect(turnStartSpy).not.toHaveBeenCalled();
});

it('starts a new turn when Codex reports that the tracked turn is no longer active', async () => {
const {mockFixture, sessionState, turnCompleted} = startActiveTurn();
const nextTurnCompleted = deferred<TurnCompletedNotification>();
Expand Down Expand Up @@ -152,6 +168,37 @@ describe('_session/steering', () => {
});
});

it('reports promptRequired when the tracked turn ends during injection', async () => {
const {mockFixture, sessionState, turnCompleted, turnStartSpy} = startActiveTurn();
vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer").mockImplementation(async () => {
turnCompleted.resolve({
threadId: "session-id",
turn: createTurn("turn-id", "completed"),
});
throw Object.assign(new Error("Internal error"), {
data: {details: "no active turn to steer"},
});
});

const promptPromise = mockFixture.getCodexAcpAgent().prompt({
sessionId: "session-id",
prompt: [{type: "text", text: "long running prompt"}],
});
await vi.waitFor(() => {
expect(sessionState.currentTurnId).toBe("turn-id");
});

await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, {
sessionId: "session-id",
prompt: [{type: "text", text: "racing follow-up"}],
_meta: {steering: {idleBehavior: "promptRequired"}},
})).resolves.toEqual({outcome: "promptRequired", reason: "noRunningTurn"});

await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"});
expect(sessionState.currentTurnId).toBeNull();
expect(turnStartSpy).toHaveBeenCalledTimes(1);
});

it('serializes concurrent late steering requests without dropping either prompt', async () => {
const mockFixture = createCodexMockTestFixture();
const sessionState = createTestSessionState();
Expand Down Expand Up @@ -212,6 +259,16 @@ describe('_session/steering', () => {
})).rejects.toThrow(RequestError);
});

it('rejects an unsupported idle steering behavior', async () => {
const mockFixture = createCodexMockTestFixture();

await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, {
sessionId: "session-id",
prompt: [{type: "text", text: "do not start a fallback turn"}],
_meta: {steering: {idleBehavior: "startedNewTurn"}},
})).rejects.toThrow(RequestError);
});

it('rejects image input when the model does not support it', async () => {
const {mockFixture} = startActiveTurn({supportedInputModalities: ["text"]});
const turnSteerSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer");
Expand Down
Loading