Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
41 changes: 41 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2721,6 +2721,47 @@ describe("CodexAdapterV2 post-settle continuation", () => {
],
});

it.effect("marks a root command it wakes the agent for once the command finishes", () =>
Effect.gen(function* () {
const harness = yield* makeCodexReplayHarness({
...backgroundExecTranscript,
entries: backgroundExecTranscript.entries.map((entry) =>
entry.type === "emit_inbound" && entry.label === "item/completed/command-late"
? { ...entry, afterMs: 0 }
: entry,
),
});
yield* harness.runtime.startTurn(
makeCodexTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make("attempt-command-completion-wake"),
text: BG_PROMPT,
}),
);
const commands = () =>
harness.events.flatMap((event) =>
event.type === "turn_item.updated" && event.turnItem.type === "command_execution"
? [[event.turnItem.status, event.turnItem.completionWake]]
: [],
);
yield* harness.until(
() => harness.continuationRequests.length === 1 && commands().length === 2,
);
// Restart recovery reads this to keep the wake the restart cut.
assert.deepEqual(commands(), [
["running", "always"],
["completed", "always"],
]);
}).pipe(
Effect.scoped,
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
);

it.effect.each([false, true])(
"keeps commands interrupted after native thread closure (persistent tool: %s)",
(retainContext) =>
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3600,6 +3600,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
...(item.exitCode === null || item.exitCode === undefined
? {}
: { exitCode: item.exitCode }),
// Matches the continuation offered when it completes after the turn.
...(context.subagent === null && continuationRequests !== undefined
? { completionWake: "always" as const }
: {}),
};
return { node, turnItem };
});
Expand Down
9 changes: 5 additions & 4 deletions apps/server/src/orchestration-v2/RestartBackgroundNote.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ export function cancelledTurnItemWork(
kind: "shell",
label: compactLabel(item.input) ?? compactLabel(item.title) ?? "background command",
id: item.id,
...(item.completionWake === undefined ? {} : { completionWake: item.completionWake }),
};
case "dynamic_tool": {
const monitor =
Expand Down Expand Up @@ -158,11 +159,11 @@ export function isRestartNoteSource(
}

/**
* A shell does not wake it: most are dev servers that nobody waits on, so their
* note waits for the next user message.
* A shell does not wake it unless its provider would have: most are dev
* servers that nobody waits on, so their note waits for the next user message.
*/
export function isAwaitedWork(work: Pick<Work, "kind">): boolean {
return work.kind !== "shell";
export function isAwaitedWork(work: Pick<Work, "kind" | "completionWake">): boolean {
return work.kind !== "shell" || work.completionWake === "always";
}

export function runOutcome(
Expand Down
66 changes: 66 additions & 0 deletions apps/server/src/orchestration-v2/RestartContinuation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1145,3 +1145,69 @@ it.effect.each(["completed", "failed", "interrupted"] as const)(
);
}),
);

it.effect("wakes a settled Codex run whose own yielded command the restart cut", () =>
Effect.gen(function* () {
// Codex promises the root agent a continuation when a command it yielded
// finishes after the turn; a restart that cuts the command must keep that.
const base = makeProjection();
const projection = {
...base,
runs: [{ ...base.runs[0]!, status: "completed", completedAt: now }],
providerThreads: [{ ...base.providerThreads[0]!, status: "idle" }],
providerSessions: [{ ...base.providerSessions[0]!, status: "ready" }],
providerTurns: [{ ...base.providerTurns[0]!, status: "completed" }],
turnItems: [
{
id: "turn-item:sleep",
runId,
nodeId: null,
providerThreadId,
providerInstanceId: instanceId,
type: "command_execution",
input: "sleep 180",
status: "running",
completionWake: "always",
},
],
} as unknown as OrchestrationV2ThreadProjection;
const commits: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0][] = [];
const recovery = yield* ProviderRuntimeRecovery.make.pipe(
Effect.provide(
Layer.mergeAll(
ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
Layer.mock(ProjectionStore.ProjectionStoreV2)({
getRecoveryThreadIds: () => Effect.succeed([threadId]),
getRuntimeRecoveryProjection: () => Effect.succeed(projection),
}),
Layer.mock(EventSink.EventSinkV2)({
commitCommand: (input) =>
Effect.sync(() => {
commits.push(input);
return { committed: true, cancelledEffectCount: 0 } as never;
}),
}),
IdAllocator.layer,
Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
runRecoveryOnce: Effect.succeed(false),
}),
Layer.mock(EffectOutbox.EffectOutboxV2)({
listByCommandId: () => Effect.succeed([]),
reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
}),
),
),
);
yield* recovery.reconcile("startup");
assert.lengthOf(commits, 1);
assert.isTrue(
commits[0]!.events.some(
(event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
),
);
assert.deepEqual(
commits[0]!.effects.map((effect) => effect.request),
[{ type: "provider-runtime.continue", sourceRunId: runId }],
);
}),
);
6 changes: 6 additions & 0 deletions packages/contracts/src/orchestrationV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -556,6 +556,8 @@ export const OrchestrationV2RestartCancelledBackgroundWork = Schema.Struct({
kind: Schema.Literals(["subagent", "shell", "monitor", "task"]),
label: TrimmedNonEmptyString,
id: Schema.optional(TrimmedNonEmptyString),
/** The provider wakes the agent when this work finishes, so it is awaited. */
completionWake: Schema.optional(Schema.Literal("always")),
});
export type OrchestrationV2RestartCancelledBackgroundWork =
typeof OrchestrationV2RestartCancelledBackgroundWork.Type;
Expand Down Expand Up @@ -1477,6 +1479,8 @@ export const OrchestrationV2TurnItem = Schema.Union([
outputOmitted: Schema.optional(Schema.Boolean),
outputIndicatesFailure: Schema.optional(Schema.Boolean),
exitCode: Schema.optional(Schema.Int),
/** The provider wakes the agent when this command finishes after its turn. */
completionWake: Schema.optional(Schema.Literal("always")),
}),
Schema.Struct({
...OrchestrationV2TurnItemBaseFields,
Expand Down Expand Up @@ -2244,6 +2248,8 @@ export const OrchestrationV2TurnItemJson = Schema.Union([
outputOmitted: Schema.optional(Schema.Boolean),
outputIndicatesFailure: Schema.optional(Schema.Boolean),
exitCode: Schema.optional(Schema.Int),
/** The provider wakes the agent when this command finishes after its turn. */
completionWake: Schema.optional(Schema.Literal("always")),
}),
Schema.Struct({
...OrchestrationV2TurnItemJsonBaseFields,
Expand Down
Loading