Skip to content
Open
27 changes: 23 additions & 4 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1535,7 +1535,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {
event.type === "message.updated" && event.message.role === "assistant",
);

it.effect("keeps an asynchronous Codex question actionable after the turn completes", () =>
it.effect("marks an asynchronous Codex question awaited while the agent sleeps on it", () =>
Effect.scoped(
Effect.gen(function* () {
const nativeThreadId = "async-question-thread";
Expand Down Expand Up @@ -1573,6 +1573,20 @@ describe("CodexAdapterV2 post-settle continuation", () => {
},
},
},
...(["item/started", "item/completed"] as const).map((method) => ({
type: "emit_inbound" as const,
label: `sleep ${method}`,
frame: {
method,
params: {
threadId: nativeThreadId,
turnId: nativeTurnId,
startedAtMs: 1782622441000,
completedAtMs: 1782622501000,
item: { type: "sleep", id: "sleep-item", durationMs: 60_000 },
},
},
})),
{
type: "emit_inbound",
label: "usage",
Expand Down Expand Up @@ -1612,9 +1626,14 @@ describe("CodexAdapterV2 post-settle continuation", () => {
const requests = harness.events.flatMap((event) =>
event.type === "runtime_request.updated" ? [event.runtimeRequest] : [],
);
assert.lengthOf(requests, 1);
assert.equal(requests[0]?.status, "pending");
assert.deepEqual(requests[0]?.responseCapability, { type: "message" });
assert.deepEqual(
requests.map((request) => [request.status, request.responseCapability]),
[
["pending", { type: "message" }],
["pending", { type: "message", agentWaiting: true }],
["pending", { type: "message" }],
],
);
const questionItem = harness.events.find(
(event) =>
event.type === "turn_item.updated" && event.turnItem.type === "user_input_request",
Expand Down
34 changes: 34 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1131,6 +1131,7 @@ interface ActiveCodexTurnContext {
readonly runningCompactionItemIds: Set<string>;
// Modern approval requests omit paths; keep the kinds without retaining patch bodies.
readonly fileChangeKindsByItem: Map<string, CodexFileChangeKinds>;
readonly asyncQuestions: Map<string, OrchestrationV2RuntimeRequest>;
readonly itemPositions: Map<
string,
{ readonly ordinal: number; readonly startedAt: DateTime.Utc }
Expand Down Expand Up @@ -1961,6 +1962,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
startedAt: input.startedAt,
itemPositions: new Map(),
fileChangeKindsByItem: new Map(),
asyncQuestions: new Map(),
runningCompactionItemIds: new Set(),
};
const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId;
Expand Down Expand Up @@ -2560,6 +2562,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
startedAt: turn.startedAt,
itemPositions: new Map(),
fileChangeKindsByItem: new Map(),
asyncQuestions: new Map(),
runningCompactionItemIds: new Set(),
};
beginTurnTokenUsage(activeContext);
Expand Down Expand Up @@ -4639,6 +4642,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
}).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)),
);

// Codex agents wait on an async question's answer with the interruptible clock.sleep tool.
const markAsyncQuestionsAwaited = Effect.fn("CodexAdapterV2.markAsyncQuestionsAwaited")(
function* (context: ActiveCodexTurnContext, agentWaiting: boolean) {
for (const request of context.asyncQuestions.values()) {
yield* emitProviderEvent({
type: "runtime_request.updated",
driver: CODEX_PROVIDER,
threadId: context.projectionThreadId,
runtimeRequest: {
...request,
responseCapability: agentWaiting
? { type: "message", agentWaiting }
: { type: "message" },
},
});
}
},
);

yield* client.handleServerNotification("item/started", (payload) =>
Effect.gen(function* () {
const context = yield* awaitActiveTurn(payload.turnId);
Expand Down Expand Up @@ -4675,6 +4697,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
}

yield* completeProviderRetry(context, yield* DateTime.now);
if (payload.item.type === "sleep") {
yield* markAsyncQuestionsAwaited(context, true);
return;
}
if (payload.item.type !== "reasoning") {
yield* resolveItemPosition(
context,
Expand Down Expand Up @@ -4800,6 +4826,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
}
const { context, settled } = resolved;

if (payload.item.type === "sleep") {
yield* markAsyncQuestionsAwaited(context, false);
return;
}

if (payload.item.type === "reasoning") {
yield* completeReasoning(payload.turnId, payload.item);
return;
Expand Down Expand Up @@ -5015,6 +5046,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
threadId: artifacts.node.threadId,
runtimeRequest: artifacts.request,
});
context.asyncQuestions.set(String(artifacts.request.id), artifacts.request);
yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
const ids = current.get(payload.turnId);
if (!ids?.has(payload.item.id)) return current;
Expand Down Expand Up @@ -5853,6 +5885,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
return false;
}
input.context.fileChangeKindsByItem.clear();
input.context.asyncQuestions.clear();
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
const providerRetry = yield* Ref.modify(providerRetries, (current) => {
const retry = current.get(input.context.providerTurnId);
if (retry === undefined) {
Expand Down Expand Up @@ -6334,6 +6367,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
startedAt: now,
itemPositions: new Map(),
fileChangeKindsByItem: new Map(),
asyncQuestions: new Map(),
runningCompactionItemIds: new Set(),
};
const providerTurn = {
Expand Down
16 changes: 13 additions & 3 deletions apps/server/src/orchestration-v2/EventSink.ts
Original file line number Diff line number Diff line change
Expand Up @@ -284,23 +284,31 @@ const layerBase: Layer.Layer<
});

// A user can answer after terminal normalization reads the pending request.
// Recheck inside the write transaction so stale cleanup cannot erase answers.
// Recheck inside the write transaction so stale cleanup cannot erase answers,
// and so a provider repeating a pending question cannot reopen it.
const guardUserInputCancellations =
/* This helper runs for individual orchestration events; the enclosing operation owns the span. */
Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
const staleRequests = new Set<RuntimeRequestId>();
const staleNodes = new Set<NodeId>();
const settledRequests = new Set<RuntimeRequestId>();
for (const event of events) {
if (
event.type !== "runtime-request.updated" ||
event.payload.kind !== "user_input" ||
event.payload.status !== "cancelled"
(event.payload.status !== "cancelled" && event.payload.status !== "pending")
)
continue;
const current = yield* projectionStore.getRuntimeRequest(
event.threadId,
event.payload.id,
);
if (event.payload.status === "pending") {
if (current !== undefined && current.status !== "pending") {
settledRequests.add(event.payload.id);
}
continue;
}
if (
current?.status !== "pending" ||
current.kind !== "user_input" ||
Expand All @@ -314,7 +322,9 @@ const layerBase: Layer.Layer<
return events.filter((event) => {
switch (event.type) {
case "runtime-request.updated":
return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
return event.payload.status === "pending"
? !settledRequests.has(event.payload.id)
: event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
case "node.updated":
return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
case "turn-item.updated":
Expand Down
5 changes: 4 additions & 1 deletion apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3709,10 +3709,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
const requestId = RuntimeRequestId.make("request:projection-shell-question");
for (const [responseCapability, blocking] of [
[{ type: "message" }, false],
[{ type: "message", agentWaiting: true }, true],
[{ type: "live", providerSessionId: ProviderSessionId.make("session:question") }, true],
] as const) {
yield* projectionStore.apply({
id: EventId.make(`event:projection-shell-question:${responseCapability.type}`),
id: EventId.make(
`event:projection-shell-question:${responseCapability.type}:${blocking}`,
),
type: "runtime-request.updated",
threadId,
nodeId: rootNodeId,
Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1358,7 +1358,9 @@ function pendingRuntimeRequestSummary(
id: request.id,
kind: request.kind,
createdAt: request.createdAt,
blocking: request.responseCapability.type !== "message",
blocking:
request.responseCapability.type !== "message" ||
request.responseCapability.agentWaiting === true,
};
}

Expand Down
26 changes: 26 additions & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1450,6 +1450,32 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => {
.pipe(Effect.result);
assert.equal(duplicate._tag, "Failure");
assert.equal((yield* orchestrator.getThreadProjection(threadId)).messages.length, 1);

// Codex repeats the question as pending when the sleep the answer interrupted ends.
const answeredRequest = answered.runtimeRequests[0]!;
yield* eventSink.write({
guardPendingUserInputCancellations: true,
commandId: CommandId.make("runtime-async-question-repeat"),
events: [
{
id: EventId.make("runtime-async-question-repeat"),
type: "runtime-request.updated",
threadId,
nodeId,
occurredAt: yield* DateTime.now,
payload: {
...answeredRequest,
status: "pending",
resolvedAt: null,
responseCapability: { type: "message" },
},
},
],
});
assert.deepEqual(
(yield* orchestrator.getThreadProjection(threadId)).runtimeRequests,
answered.runtimeRequests,
);
}),
);

Expand Down
4 changes: 4 additions & 0 deletions docs/internals/providers.md
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,10 @@ no pending RPC response to send. The
`user_input_request` turn items and runtime requests with `responseCapability: { type: "message" }`.
Their execution nodes do not block the run. Web, desktop, and mobile use their normal question
panels, and requests remain pending after a turn finishes, a provider exits, or the server restarts.
The agent waits for an answer by calling the interruptible `clock.sleep` tool, so the adapter sets
`agentWaiting` on the request while a sleep item runs; status surfaces then show the thread waiting
for input instead of Working. The event sink drops the request's later `pending` repeats once it
is answered.

`runtime-request.respond` reads the persisted request and question item, validates required
answers, and commits the resolution and a user message in one transaction. Repeating the same
Expand Down
3 changes: 2 additions & 1 deletion docs/user/providers-codex.md
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,8 @@ contains a copied Codex setup. Use a fresh shadow directory and sign in again.
## Answer questions while Codex works

Codex can ask a question and keep working. Answer it in the thread's question
panel. The answer becomes a new message: it reaches the active turn, or starts
panel. While Codex stops to wait for your answer, the thread shows that it needs
input and you get the usual notification. The answer becomes a new message: it reaches the active turn, or starts
another turn if Codex has finished. Unanswered questions survive reconnects.
If you do not want to answer, dismiss the question from its panel. Dismissing
closes it without sending anything to Codex. This requires a Codex version that
Expand Down
6 changes: 5 additions & 1 deletion packages/contracts/src/orchestrationV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1190,7 +1190,11 @@ export const OrchestrationV2RuntimeRequest = Schema.Struct({
providerSessionId: ProviderSessionId,
supportsFeedback: Schema.optional(Schema.Boolean),
}),
Schema.Struct({ type: Schema.Literal("message") }),
Schema.Struct({
type: Schema.Literal("message"),
// The agent has stopped working to wait for this answer.
agentWaiting: Schema.optional(Schema.Boolean),
}),
Schema.Struct({ type: Schema.Literal("not_resumable"), reason: Schema.String }),
]),
createdAt: Schema.DateTimeUtc,
Expand Down
Loading