Skip to content
Merged
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
Original file line number Diff line number Diff line change
Expand Up @@ -1007,6 +1007,42 @@ describe("AcpRegistryAdapterV2", () => {
}).pipe(Effect.provide(layerTest), Effect.scoped),
);

// regression: #26695 cancelling a Copilot prompt kills its background agents
it.effect("never steers Copilot by cancelling its prompt", () =>
Effect.gen(function* () {
const steering = Effect.fn("steering")(function* (config: Record<string, unknown>) {
const instanceId = ProviderInstanceId.make("acp-registry-steering");
const adapter = yield* makeAcpRegistryAdapterV2({
selfInvocation: yield* resolveSelfInvocation(),
instanceId,
settings: yield* decodeAcpRegistryAdapterSettings(config),
environment: {},
}).pipe(Effect.provide(layerInjectedRuntimeCatalog));
const { turns } = yield* adapter.getCapabilities();
return {
supportsInterrupt: turns.supportsInterrupt,
supportsActiveSteering: turns.supportsActiveSteering,
supportsSteeringByInterruptRestart: turns.supportsSteeringByInterruptRestart,
};
});
const copilot = {
supportsInterrupt: true,
supportsActiveSteering: false,
supportsSteeringByInterruptRestart: false,
};

assert.deepEqual(yield* steering({ agentId: "github-copilot-cli" }), copilot);
assert.deepEqual(
yield* steering({ source: "local", commandPath: "C:\\tools\\copilot.cmd" }),
copilot,
);
assert.deepEqual(yield* steering({ agentId: "gemini" }), {
...copilot,
supportsSteeringByInterruptRestart: true,
});
}).pipe(Effect.provide(layerTest), Effect.scoped),
);

it.effect("settles Claude's accepted foreground reply while retaining child output", () =>
Effect.gen(function* () {
const gate = makeProviderReplayGate(["accepted", "child-output", "wake"]);
Expand Down
47 changes: 47 additions & 0 deletions apps/server/src/orchestration-v2/CommandPolicy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,53 @@ it.each([
);
});

// regression: #26695 an interrupt-restart steer cancels Copilot's background agents
it("queues a steer for a provider that cannot steer either way", () => {
const projection = dispatchProjection(
capabilities((current) => ({
...current,
turns: {
...current.turns,
supportsActiveSteering: false,
supportsSteeringByInterruptRestart: false,
supportsQueuedMessages: true,
},
})),
);
for (const deliveryIntent of ["auto", "steer"] as const) {
assert.deepEqual(
CommandPolicy.resolveMessageDispatchIntent(
projection,
{ type: "start_immediately" },
deliveryIntent,
),
{ type: "queue_after_active" },
);
}
assert.deepEqual(
CommandPolicy.resolveMessageDispatchIntent(projection, {
type: "steer_active",
targetRunId: activeRunId,
}),
{ type: "queue_after_active" },
);
assert.deepEqual(
CommandPolicy.resolveMessageDispatchIntent(
projection,
{ type: "start_immediately" },
"restart",
),
{ type: "restart_active", targetRunId: activeRunId },
);
assert.deepEqual(
CommandPolicy.resolveMessageDispatchIntent(projection, {
type: "steer_active",
targetRunId: RunId.make("command-policy-missing-run"),
}),
{ type: "steer_active", targetRunId: RunId.make("command-policy-missing-run") },
);
});

it.each(["/compact", " /LOGOUT "])("queues a steer sent while %s runs", (text) => {
const projection = dispatchProjection(baseCapabilities);
const maintenance = {
Expand Down
57 changes: 39 additions & 18 deletions apps/server/src/orchestration-v2/CommandPolicy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -99,25 +99,55 @@ export function resolveMessageDispatchIntent(
// Compaction and sign-out cannot accept steers; queue follow-ups.
const target = projection.runs.find((run) => run.id === mode.targetRunId);
const targetMessage = projection.messages.find((message) => message.id === target?.userMessageId);
return targetMessage !== undefined && isNativeMaintenanceCommand(targetMessage)
if (targetMessage !== undefined && isNativeMaintenanceCommand(targetMessage)) {
return { type: "queue_after_active" };
}
// A provider that can neither steer nor restart its turn takes the message after the turn.
const turns =
target === undefined || !isActiveRun(target)
? undefined
: providerTurnCapabilities(projection, target);
return turns !== undefined &&
!turns.supportsActiveSteering &&
!(turns.supportsInterrupt && turns.supportsSteeringByInterruptRestart) &&
turns.supportsQueuedMessages
? { type: "queue_after_active" }
: mode;
}

function isActiveRun(run: OrchestrationV2Run): boolean {
return (
run.status === "preparing" ||
run.status === "starting" ||
run.status === "running" ||
run.status === "waiting"
);
}

function providerTurnCapabilities(
projection: OrchestrationV2ThreadProjection,
run: OrchestrationV2Run,
) {
const providerThread = projection.providerThreads.find(
(candidate) => candidate.id === run.providerThreadId,
);
const providerSession =
providerThread?.providerSessionId == null
? undefined
: projection.providerSessions.find(
(candidate) => candidate.id === providerThread.providerSessionId,
);
return providerSession?.capabilities.turns;
}

function resolveDeliveryIntent(
projection: OrchestrationV2ThreadProjection,
requestedMode: MessageDispatchMode,
deliveryIntent?: "auto" | "steer" | "restart",
): MessageDispatchMode {
if (deliveryIntent === undefined) return requestedMode;

const activeRun = projection.runs.findLast(
(run) =>
run.status === "preparing" ||
run.status === "starting" ||
run.status === "running" ||
run.status === "waiting",
);
const activeRun = projection.runs.findLast(isActiveRun);
if (activeRun === undefined) return { type: "start_immediately" };
if (deliveryIntent === "steer") {
return { type: "steer_active", targetRunId: activeRun.id };
Expand All @@ -137,16 +167,7 @@ function resolveDeliveryIntent(
return { type: "queue_after_active" };
}

const providerThread = projection.providerThreads.find(
(candidate) => candidate.id === activeRun.providerThreadId,
);
const providerSession =
providerThread?.providerSessionId == null
? undefined
: projection.providerSessions.find(
(candidate) => candidate.id === providerThread.providerSessionId,
);
const capabilities = providerSession?.capabilities.turns;
const capabilities = providerTurnCapabilities(projection, activeRun);
if (capabilities?.supportsActiveSteering === true) {
return { type: "steer_active", targetRunId: activeRun.id };
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -447,16 +447,18 @@ const composerSelection = {
const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* (
name: string,
canRestart = false,
canSteer = true,
) {
const cwd = yield* checkpointWorkspace(name);
const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = [];
const steered: string[] = [];
const interrupted: ProviderTurnId[] = [];
const capabilities = {
...CodexProviderCapabilitiesV2,
turns: {
...CodexProviderCapabilitiesV2.turns,
supportsActiveSteering: true,
supportsActiveSteering: canSteer,
supportsSteeringByInterruptRestart: canRestart,
},
};
Expand Down Expand Up @@ -533,7 +535,10 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function*
Effect.sync(() => {
steered.push(turn.message.text);
}),
interruptTurn: () => Effect.void,
interruptTurn: ({ providerTurnId }) =>
Effect.sync(() => {
interrupted.push(providerTurnId);
}),
respondToRuntimeRequest: () => Effect.void,
readThreadSnapshot: () => Effect.die("unused"),
rollbackThread: () => Effect.die("unused"),
Expand Down Expand Up @@ -588,7 +593,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function*
yield* Fiber.join(running);
return threadId;
});
return { cwd, events, started, steered, layer, startFirstTurn };
return { cwd, events, started, steered, interrupted, layer, startFirstTurn };
});

it.effect("sends a selected held message once and preserves the remaining queue order", () =>
Expand Down Expand Up @@ -938,6 +943,82 @@ it.effect("starts a steer that missed the turn on the saved next-turn selection"
),
);

// regression: #26695 cancelling a Copilot prompt to steer it kills its background agents
it.effect("queues a steer until the turn ends when the provider cannot steer it", () =>
Effect.scoped(
Effect.gen(function* () {
const { events, started, steered, interrupted, layer, startFirstTurn } =
yield* nextTurnSelectionHarness("steering-unsupported", false, false);
yield* Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
const threadId = yield* startFirstTurn;
const first = started[0]!;
const messageId = MessageId.make("message:steer");
yield* orchestrator.dispatch({
type: "message.dispatch",
commandId: CommandId.make("steer"),
threadId,
messageId,
text: "follow up",
attachments: [],
modelSelection: runSelection,
dispatchMode: { type: "start_immediately" },
deliveryIntent: "steer",
createdBy: "user",
creationSource: "web",
});
yield* worker.drain();

const held = yield* orchestrator.getThreadProjection(threadId);
assert.deepEqual(steered, []);
assert.deepEqual(interrupted, []);
assert.equal(started.length, 1);
assert.equal(held.runs.find((run) => run.id === first.runId)?.status, "running");
const queued = held.runs.find((run) => run.userMessageId === messageId);
assert.equal(queued?.status, "queued");

const settled = yield* orchestrator.streamDomainEvents.pipe(
Stream.filter(
(event) =>
event.type === "run.updated" &&
event.payload.id === first.runId &&
event.payload.status === "waiting",
),
Stream.take(1),
Stream.runDrain,
Effect.forkScoped,
);
const turn = held.providerTurns[0]!;
yield* Queue.offer(events, {
type: "provider_turn.updated",
driver,
providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now },
});
yield* Queue.offer(events, {
type: "turn.terminal",
driver,
providerThreadId: turn.providerThreadId,
providerTurnId: turn.id,
runOrdinal: first.runOrdinal,
status: "completed",
failure: null,
threadDisposition: "reusable",
});
yield* Fiber.join(settled);
yield* worker.drain();
yield* orchestrator.resumeQueuedRuns;
yield* worker.drain();

assert.deepEqual(interrupted, []);
assert.equal(started.length, 2);
assert.equal(started[1]?.runId, queued?.id);
assert.equal(started[1]?.message.text, "follow up");
}).pipe(Effect.provide(layer));
}),
),
);

// Codex flushes held drafts after the final answer; reserve their pre-steer position before turn end.
it.effect("keeps a held commentary draft before the steer when turn end flushes it", () =>
Effect.scoped(
Expand Down
8 changes: 7 additions & 1 deletion packages/provider-acp-registry/src/server/adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -210,7 +210,13 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu
const copilotRouting = isCopilot ? makeCopilotSubagentRouting() : undefined;
const flavor: AcpAdapterV2Flavor = {
driver: ACP_REGISTRY_PROVIDER,
capabilities: AcpProviderCapabilitiesV2,
// Cancelling a Copilot prompt also kills the background agents it started.
capabilities: isCopilot
? {
...AcpProviderCapabilitiesV2,
turns: { ...AcpProviderCapabilitiesV2.turns, supportsSteeringByInterruptRestart: false },
}
: AcpProviderCapabilitiesV2,
customModels: readCustomModelEntries(options.settings.customModels).map((entry) => entry.slug),
promptFailure: (cause) => acpRegistryPromptFailure(registryAgentId, cause),
...(registryAgentId === "factory-droid"
Expand Down
Loading