Skip to content
Merged
91 changes: 91 additions & 0 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1605,6 +1605,97 @@ describe("OrchestratorMcpService provider resolution", () => {
}),
);

it.effect(
"retrying an accepted thread creation replays even after the project disables every candidate",
() =>
Effect.gen(function* () {
const altInstanceId = ProviderInstanceId.make("codex-alt");
const dispatched = yield* Ref.make<ReadonlyArray<unknown>>([]);
const accepted = yield* Ref.make(false);
const everyCandidateDisabled = yield* Ref.make(false);
const layerDependencies = Layer.mergeAll(
NodeServices.layer,
Layer.mock(ThreadManagementService.ThreadManagementService)({
getThreadRecords: (threadId) =>
Effect.succeed(threadId === parentThreadId ? parentProjection([]) : childProjection),
// A command id already accepted once, even if the project later
// disables every candidate: its retry must replay, not re-resolve.
hasReceipt: () => Ref.get(accepted),
// Effective settings once the project turns every instance off,
// ignoring the already-accepted command's replay exemption.
isProviderInstanceEnabledForProject: () =>
Ref.get(everyCandidateDisabled).pipe(Effect.map((disabled) => !disabled)),
assertProviderInstanceEnabled: ({ projectId, instanceId }) =>
Effect.all([Ref.get(accepted), Ref.get(everyCandidateDisabled)]).pipe(
Effect.flatMap(([isAccepted, disabled]) =>
isAccepted || !disabled
? Effect.void
: Effect.fail(
new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
projectId,
instanceId,
}),
),
),
),
dispatch: (command) =>
Ref.update(dispatched, (commands) => [...commands, command]).pipe(
Effect.andThen(
command.type === "thread.create" ? Ref.set(accepted, true) : Effect.void,
),
Effect.as({ sequence: 1, storedEvents: [] } as never),
),
}),
providerRegistryLayer([
providerSnapshot({
instanceId: codexInstanceId,
driver: ProviderDriverKind.make("codex"),
model: "gpt-5.4",
}),
providerSnapshot({
instanceId: altInstanceId,
driver: ProviderDriverKind.make("codex"),
model: "gpt-5.4",
}),
]),
adapterRegistryLayer([codexInstanceId, altInstanceId]),
Layer.mock(ProjectService.ProjectService)({}),
Layer.mock(SecretRequests.SecretRequests)({}),
Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
ServerSettings.layerTest(),
);

yield* Effect.gen(function* () {
const service = yield* OrchestratorMcpService.OrchestratorMcpService;
const first = yield* service.createThreads(scope, {
threads: [{ target: { driverKind: ProviderDriverKind.make("codex") } }],
clientRequestId: "create-thread-retry-after-disable-1",
});
assert.equal(first.threads.length, 1);

// The project turns off every instance of this driver only after
// the thread creation above was already accepted.
yield* Ref.set(everyCandidateDisabled, true);

const retry = yield* service.createThreads(scope, {
threads: [{ target: { driverKind: ProviderDriverKind.make("codex") } }],
clientRequestId: "create-thread-retry-after-disable-1",
});
assert.equal(retry.threads.length, 1);

const commands = yield* Ref.get(dispatched);
const createCommands = commands.filter(
(command): command is { type: string; modelSelection: { instanceId: string } } =>
(command as { type: string }).type === "thread.create",
);
assert.equal(createCommands.length, 2);
assert.equal(createCommands[1]?.modelSelection.instanceId, codexInstanceId);
}).pipe(
Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))),
);
}),
);

it.effect("re-probes an unavailable target once before refusing it", () =>
Effect.gen(function* () {
const claudeInstanceId = ProviderInstanceId.make("claudeAgent");
Expand Down
26 changes: 15 additions & 11 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2264,10 +2264,23 @@ const make = Effect.gen(function* () {
input.threads,
(request, index) =>
Effect.gen(function* () {
// Computed before resolving the target so a retry of an
// already accepted thread creation can be recognized:
// dispatching under the same command id only replays the
// recorded result, so candidate selection must not refuse it
// over the project's current settings.
const commandId = stableCommandId({
scope,
requestKey: key,
operation: "create-thread",
index,
});
const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId);
Comment thread
yordis marked this conversation as resolved.
Resolved
const target = yield* resolveTargetRechecking({
parent,
target: request.target,
providers,
skipProjectEnablementFilter: isRetryOfAcceptedCreation,
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
});
const runtimeMode = yield* resolveRuntimeMode(
parent.thread.runtimeMode,
Expand All @@ -2283,11 +2296,7 @@ const make = Effect.gen(function* () {
requestKey: key,
index,
});
yield* assertTargetEnabledForProject(
parent,
stableCommandId({ scope, requestKey: key, operation: "create-thread", index }),
target,
);
yield* assertTargetEnabledForProject(parent, commandId, target);
const title = threadTitle({
parentTitle: parent.thread.title,
prompt: request.prompt,
Expand All @@ -2299,12 +2308,7 @@ const make = Effect.gen(function* () {
type: "thread.create",
createdBy: "agent",
creationSource: "mcp",
commandId: stableCommandId({
scope,
requestKey: key,
operation: "create-thread",
index,
}),
commandId,
threadId,
projectId: parent.thread.projectId,
title,
Expand Down
Loading