From ac59ad5c4dc75ecb5862460f1788269fc8e7cc2f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:08:36 -0700 Subject: [PATCH 1/2] fix(server): exclude exhausted providers from delegation --- .../src/mcp/OrchestratorMcpService.test.ts | 239 +++++++++++++++--- apps/server/src/mcp/OrchestratorMcpService.ts | 45 +++- .../orchestration-v2/ProviderPoolRouting.ts | 2 +- .../ThreadLaunchService.test.ts | 41 +++ .../orchestration-v2/ThreadLaunchService.ts | 25 ++ .../src/provider/orchestrationTargets.ts | 34 +++ 6 files changed, 333 insertions(+), 53 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index b240b308a262..0198867aa97c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -28,6 +28,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; @@ -1220,7 +1221,7 @@ describe("OrchestratorMcpService provider resolution", () => { const byId = new Map( capabilities.providers.map((provider) => [provider.providerInstanceId, provider]), ); - for (const provider of providers) { + for (const provider of providers.filter((provider) => provider.enabled)) { assert.deepEqual( byId.get(provider.instanceId)?.models.map((model) => model.id), provider.models.map((model) => model.slug), @@ -1240,7 +1241,7 @@ describe("OrchestratorMcpService provider resolution", () => { ], })); const refreshed = yield* service.capabilities(scope); - for (const provider of providers) { + for (const provider of providers.filter((provider) => provider.enabled)) { assert.deepEqual( refreshed.providers .find((entry) => entry.providerInstanceId === provider.instanceId) @@ -1266,30 +1267,9 @@ describe("OrchestratorMcpService provider resolution", () => { assert.deepEqual(entry!.constraints, []); } - const disabled = byId.get(disabledAntigravityInstanceId); - assert.isDefined(disabled); - assert.isFalse(disabled!.canRunChildTask); - assert.isFalse(disabled!.canRunCrossProviderChildTask); - assert.deepEqual(disabled!.constraints, ["Provider instance is disabled."]); - - const disabledWithoutAdapter = byId.get(disabledWithoutAdapterInstanceId); - assert.isDefined(disabledWithoutAdapter); - assert.isFalse(disabledWithoutAdapter!.canRunChildTask); - assert.isFalse(disabledWithoutAdapter!.canRunCrossProviderChildTask); - assert.deepEqual(disabledWithoutAdapter!.constraints, [ - "No V2 provider adapter is registered.", - "Provider instance is disabled.", - ]); - - const fork = byId.get(forkOnlyInstanceId); - assert.isDefined(fork); - assert.isFalse(fork!.canRunChildTask); - assert.isFalse(fork!.canRunCrossProviderChildTask); - assert.deepEqual(fork!.constraints, [ - "No V2 provider adapter is registered.", - "Driver 'forkOnly' is not registered in this build.", - "Provider instance is disabled.", - ]); + assert.isUndefined(byId.get(disabledAntigravityInstanceId)); + assert.isUndefined(byId.get(disabledWithoutAdapterInstanceId)); + assert.isUndefined(byId.get(forkOnlyInstanceId)); }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1465,6 +1445,172 @@ describe("OrchestratorMcpService provider resolution", () => { ); }); + it.effect( + "hides cooling providers and rejects new delegations until their cooldown expires", + () => + Effect.gen(function* () { + const providers: ReadonlyArray = [ + { + ...providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + limitedUntil: "1970-01-01T00:01:00.000Z", + }, + ]; + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + getThreadRecords: () => Effect.succeed(parentProjection([])), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getPublishedProviders: Effect.succeed(providers), + refreshInstance: () => Effect.succeed(providers), + }), + adapterRegistryLayer([codexInstanceId]), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + assert.deepEqual((yield* service.capabilities(scope)).providers, []); + for (const target of [ + { providerInstanceId: codexInstanceId }, + { driverKind: ProviderDriverKind.make("codex") }, + ]) { + const error = yield* service + .delegateTask(scope, { + task: "Summarize", + target, + mode: "async", + clientRequestId: "cooling", + }) + .pipe(Effect.flip); + assert.equal(error.code, "provider_unavailable"); + } + yield* TestClock.adjust("1 minute"); + assert.deepEqual( + (yield* service.capabilities(scope)).providers.map( + (provider) => provider.providerInstanceId, + ), + [codexInstanceId], + ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(dependencies)))); + }), + ); + + it.effect.each([ + "all limited", + "healthy capacity", + "shared login", + "shared credential", + "project disabled", + "unavailable", + "expired", + "unknown quota", + ] as const)("pool delegation availability: %s", (scenario) => + Effect.gen(function* () { + const poolInstanceId = ProviderInstanceId.make("pool_codex"); + const workInstanceId = ProviderInstanceId.make("codex_work"); + const codex = ProviderDriverKind.make("codex"); + const limitedUntil = "2099-01-01T00:00:00.000Z"; + const members: ServerProvider[] = [codexInstanceId, workInstanceId].map( + (instanceId, index) => ({ + ...providerSnapshot({ instanceId, driver: codex, model: "gpt-5.4" }), + pool: { role: "member", poolInstanceId }, + ...(index === 0 || scenario === "all limited" ? { limitedUntil } : {}), + ...(scenario === "expired" ? { limitedUntil: "1969-01-01T00:00:00.000Z" } : {}), + ...(scenario === "shared login" + ? { auth: { status: "authenticated", email: "same@example.com" } } + : {}), + ...(scenario === "shared credential" + ? { + usageLimits: { + checkedAt: "2026-01-01T00:00:00.000Z", + credentialFingerprint: "same", + windows: [], + }, + } + : {}), + ...(index === 1 && scenario === "unavailable" ? { installed: false } : {}), + }), + ); + const providers: ServerProvider[] = [ + ...members, + { + ...providerSnapshot({ instanceId: poolInstanceId, driver: codex, model: "gpt-5.4" }), + pool: { + role: "pool", + memberInstanceIds: [codexInstanceId, workInstanceId], + droppedModels: [], + }, + }, + ]; + const available = ["healthy capacity", "expired", "unknown quota"].includes(scenario); + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + getThreadRecords: () => Effect.succeed(parentProjection([])), + isProviderInstanceEnabledForProject: () => Effect.succeed(true), + }), + ServerSettings.layerTest({ + providerInstances: { + [workInstanceId]: { driver: codex, displayName: "Work", config: {}, enabled: true }, + }, + projectSettingsOverrides: { + [projectId]: { + providerInstances: { + [workInstanceId]: { enabled: scenario !== "project disabled" }, + }, + }, + }, + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getPublishedProviders: Effect.succeed(providers), + refreshInstance: () => + Effect.succeed(members.map(({ pool: _pool, ...member }) => member)), + }), + adapterRegistryLayer([codexInstanceId, workInstanceId]), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + assert.deepEqual( + (yield* service.capabilities(scope)).providers.map( + (provider) => provider.providerInstanceId, + ), + available ? [poolInstanceId] : [], + ); + if (!available) { + for (const target of [ + { providerInstanceId: poolInstanceId }, + { providerInstanceId: workInstanceId }, + { driverKind: codex }, + ]) { + const error = yield* service + .delegateTask(scope, { + task: "Summarize", + target, + mode: "async", + clientRequestId: "limited-pool", + }) + .pipe(Effect.flip); + assert.equal(error.code, "provider_unavailable"); + } + } + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(dependencies)))); + }), + ); + it.effect( "delegates a driverKind target to the driver's pool and lists the pool over its members", () => @@ -1529,9 +1675,11 @@ describe("OrchestratorMcpService provider resolution", () => { } as never), ), }), - ServerSettings.layerTest(), + ServerSettings.layerTest({ + providerInstances: { [workInstanceId]: { driver: codex, config: {}, enabled: true } }, + }), providerRegistryLayer([ - member(codexInstanceId), + { ...member(codexInstanceId), limitedUntil: "2099-01-01T00:00:00.000Z" }, member(workInstanceId), { ...providerSnapshot({ instanceId: poolInstanceId, driver: codex, model: "gpt-5.4" }), @@ -1737,7 +1885,7 @@ describe("OrchestratorMcpService provider resolution", () => { ); it.effect( - "retrying an accepted driver-only delegation replays even after the project disables every candidate", + "retrying an accepted driver-only delegation replays after every candidate cools and the project disables them", () => Effect.gen(function* () { const altInstanceId = ProviderInstanceId.make("codex-alt"); @@ -1806,18 +1954,27 @@ describe("OrchestratorMcpService provider resolution", () => { } 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", - }), - ]), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getPublishedProviders: Ref.get(accepted).pipe( + Effect.map((isAccepted) => + [ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + providerSnapshot({ + instanceId: altInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ].map((provider) => ({ + ...provider, + ...(isAccepted ? { limitedUntil: "2099-01-01T00:00:00.000Z" } : {}), + })), + ), + ), + }), adapterRegistryLayer([codexInstanceId, altInstanceId]), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 510889294b54..231aa92fe94b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1,5 +1,6 @@ import { providerConstraints, + delegationProviderConstraints, invalidOptionSelections, readOrchestrationProviders, } from "../provider/orchestrationTargets.ts"; @@ -1180,9 +1181,10 @@ const make = Effect.gen(function* () { Effect.catchIf( (error) => error.code === "provider_unavailable", () => - providerRegistry - .refreshInstance(instanceId) - .pipe(Effect.flatMap((providers) => resolveTarget({ ...input, providers }))), + providerRegistry.refreshInstance(instanceId).pipe( + Effect.andThen(loadProviders), + Effect.flatMap((providers) => resolveTarget({ ...input, providers })), + ), ), ); }; @@ -1191,12 +1193,28 @@ const make = Effect.gen(function* () { readonly parent: Pick; readonly target: OrchestratorMcpTarget | undefined; readonly providers: ReadonlyArray; - /** Receipt replays bypass current settings because they start no work. Dispatch must reuse the checked command ID. */ + /** Receipt replays bypass current cooldowns and project settings because they start no work. Dispatch must reuse the checked command ID. */ readonly skipProjectEnablementFilter?: boolean; }): Effect.fn.Return { const requestedInstanceId = input.target?.providerInstanceId; const requestedDriver = input.target?.driverKind; const orchestrationCapableInstanceIds = yield* loadOrchestrationCapableInstanceIds(); + const availability = { + providers: input.providers, + settings: yield* serverSettings.getSettings.pipe( + Effect.mapError(() => failure("orchestration_error", "Unable to read server settings.")), + ), + projectId: input.parent.thread.projectId, + nowMs: yield* Clock.currentTimeMillis, + }; + const constraintsFor = (provider: ServerProvider) => + input.skipProjectEnablementFilter === true + ? providerConstraints(provider, orchestrationCapableInstanceIds.has(provider.instanceId)) + : delegationProviderConstraints( + provider, + orchestrationCapableInstanceIds.has(provider.instanceId), + availability, + ); let instanceId = requestedInstanceId; if (instanceId === undefined && requestedDriver !== undefined) { @@ -1212,7 +1230,7 @@ const make = Effect.gen(function* () { ); } const healthyCandidates = candidates.filter( - (candidate) => providerConstraints(candidate, true).length === 0, + (candidate) => constraintsFor(candidate).length === 0, ); const poolCandidate = healthyCandidates.find((candidate) => candidate.pool?.role === "pool"); const inheritedCandidate = healthyCandidates.find( @@ -1261,10 +1279,7 @@ const make = Effect.gen(function* () { `Provider instance ${instanceId} uses driver ${provider.driver}, not ${requestedDriver}.`, ); } - const constraints = providerConstraints( - provider, - orchestrationCapableInstanceIds.has(provider.instanceId), - ); + const constraints = constraintsFor(provider); if (constraints.length > 0) { return yield* failure( "provider_unavailable", @@ -2052,6 +2067,12 @@ const make = Effect.gen(function* () { const settings = yield* serverSettings.getSettings.pipe( Effect.mapError(() => failure("orchestration_error", "Unable to read server settings.")), ); + const availability = { + providers, + settings, + projectId: parent?.thread.projectId ?? null, + nowMs: yield* Clock.currentTimeMillis, + }; const inherited = parent?.thread.modelSelection; const searchTerms = input.search?.toLowerCase().split(/\s+/) ?? []; return { @@ -2063,11 +2084,13 @@ const make = Effect.gen(function* () { interactionMode: limits.interactionMode, providers: providers .filter((provider) => provider.pool?.role !== "member") - .map((provider) => { - const constraints = providerConstraints( + .flatMap((provider) => { + const constraints = delegationProviderConstraints( provider, orchestrationCapableInstanceIds.has(provider.instanceId), + availability, ); + if (constraints.length > 0) return []; const { listed, omitted } = listCapabilityModels({ provider, preferences: settings.providerModelPreferences?.[provider.instanceId], diff --git a/apps/server/src/orchestration-v2/ProviderPoolRouting.ts b/apps/server/src/orchestration-v2/ProviderPoolRouting.ts index e5f2c4666692..d7a5135852a7 100644 --- a/apps/server/src/orchestration-v2/ProviderPoolRouting.ts +++ b/apps/server/src/orchestration-v2/ProviderPoolRouting.ts @@ -149,7 +149,7 @@ const sharesLogin = (a: ServerProvider, b: ServerProvider): boolean => { export function poolMemberHealth(input: { readonly members: ReadonlyArray; readonly settings: Parameters[0]; - readonly projectId: ProjectId; + readonly projectId: ProjectId | null; readonly nowMs: number; }): ReadonlyArray { return input.members.map((member) => { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 1e3fd1a8b368..e41dfdd59911 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -15,6 +15,7 @@ import { ComposerContextId, type ChatAttachment, CommandId, + EnvironmentId, DEFAULT_SERVER_SETTINGS, GitCommandError, MessageId, @@ -308,6 +309,46 @@ function launchInput(input: { }; } +it.effect( + "rejects a remote delegation to a cooling provider but preserves ordinary launches and receipt replay", + () => { + const providers: ServerProvider[] = [ + { + instanceId: modelSelection.instanceId, + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: "test", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-01-01T00:00:00.000Z", + models: [], + slashCommands: [], + skills: [], + }, + ]; + const harness = makeHarness({ providers }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const input = launchInput({ command: "command:cooling", thread: "thread:cooling" }); + const delegatedFrom = { + environmentId: EnvironmentId.make("environment:remote"), + threadId: ThreadId.make("thread:remote"), + title: "Parent", + }; + const launched = yield* launches.launch({ ...input, delegatedFrom }); + providers[0] = { ...providers[0]!, limitedUntil: "2099-01-01T00:00:00.000Z" }; + const next = launchInput({ command: "command:next", thread: "thread:next" }); + const refused = yield* launches.launch({ ...next, delegatedFrom }).pipe(Effect.result); + assert.isTrue(Result.isFailure(refused)); + const ordinary = yield* launches.launch(next); + assert.equal(ordinary.threadId, next.threadId); + const replay = yield* launches.launch({ ...input, delegatedFrom }); + assert.equal(replay.threadId, launched.threadId); + }).pipe(Effect.provide(harness.layer)); + }, +); + it.effect.each( (["new", "existing"] as const).flatMap((target) => (["user", "agent"] as const).map((createdBy) => ({ target, createdBy })), diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index e1d7f5b6733a..a2f64b5b69ab 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -21,6 +21,8 @@ import { type ScheduledTaskId, ThreadId, } from "@t3tools/contracts"; +import { delegationProviderConstraints } from "../provider/orchestrationTargets.ts"; +import * as Clock from "effect/Clock"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; @@ -836,6 +838,29 @@ const make = Effect.gen(function* () { } const launchReceipt = yield* readReceipt(input, input.commandId); + if ( + input.delegatedFrom !== undefined && + !(Option.isSome(launchReceipt) && launchReceipt.value.status === "accepted") + ) { + const providers = yield* providerRegistry.getPublishedProviders; + const provider = providers.find( + (candidate) => candidate.instanceId === input.modelSelection.instanceId, + ); + const settings = yield* serverSettings.getSettings.pipe( + Effect.mapError(mapError(input, "create-thread")), + ); + const constraints = + provider === undefined + ? ["Provider instance is not registered."] + : delegationProviderConstraints(provider, true, { + providers, + settings, + projectId: input.projectId, + nowMs: yield* Clock.currentTimeMillis, + }); + if (constraints.length > 0) + return yield* mapError(input, "create-thread")(constraints.join(" ")); + } yield* threads.assertProviderInstanceEnabled({ projectId: input.projectId, commandId: input.commandId, diff --git a/apps/server/src/provider/orchestrationTargets.ts b/apps/server/src/provider/orchestrationTargets.ts index f8b96e64cfd2..8ae91490deac 100644 --- a/apps/server/src/provider/orchestrationTargets.ts +++ b/apps/server/src/provider/orchestrationTargets.ts @@ -1,9 +1,13 @@ import { isProviderAvailable, + findProviderPool, + providerLimitedUntilMs, + type ProjectId, type ServerProvider, type ProviderOptionDescriptor, type ProviderOptionSelection, } from "@t3tools/contracts"; +import { poolMemberHealth } from "../orchestration-v2/ProviderPoolRouting.ts"; import * as Effect from "effect/Effect"; import * as ProviderRegistry from "./ProviderRegistry.ts"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; @@ -35,6 +39,36 @@ export function providerConstraints( return constraints; } +export function delegationProviderConstraints( + provider: ServerProvider, + supportsOrchestrationV2: boolean, + input: { + readonly providers: ReadonlyArray; + readonly settings: Parameters[0]["settings"]; + readonly projectId: ProjectId | null; + readonly nowMs: number; + }, +): ReadonlyArray { + const constraints = [...providerConstraints(provider, supportsOrchestrationV2)]; + const owner = findProviderPool(input.providers, provider.instanceId); + if (owner === null) { + if (providerLimitedUntilMs(provider, input.nowMs) !== null) { + constraints.push("Provider is rate limited. Retry after its cooldown expires."); + } + return constraints; + } + const members = poolMemberHealth({ ...input, members: owner.members }); + const eligible = members.filter((member) => member.eligible); + if (eligible.length === 0) { + constraints.push("No pool member is available for this project."); + } else if (eligible.every((member) => member.limitedUntilMs !== null)) { + constraints.push( + "Every available pool member is rate limited. Retry after a cooldown expires.", + ); + } + return constraints; +} + /** Skip checks without descriptors; reject duplicate IDs because consumers disagree on precedence. */ export function invalidOptionSelections( selections: ReadonlyArray, From 272f56efdc510bb8ec02c9e8f08ba3c53135ee6b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:27:32 -0700 Subject: [PATCH 2/2] fix(server): recheck availability before starting delegated work --- .../src/mcp/OrchestratorMcpService.test.ts | 124 +++++++++++++++--- apps/server/src/mcp/OrchestratorMcpService.ts | 29 +++- .../ThreadLaunchService.test.ts | 19 +++ .../orchestration-v2/ThreadLaunchService.ts | 6 +- 4 files changed, 149 insertions(+), 29 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 0198867aa97c..6d815f60d2c3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -1197,7 +1197,14 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadRecords: () => Effect.succeed(parentProjection([])), }), - ServerSettings.layerTest(), + ServerSettings.layerTest({ + providerInstances: Object.fromEntries( + providers.map((provider) => [ + provider.instanceId, + { driver: provider.driver, enabled: provider.enabled, config: {} }, + ]), + ), + }), Layer.mock(ProviderRegistry.ProviderRegistry)({ getPublishedProviders: Effect.sync(() => providers), }), @@ -1355,6 +1362,13 @@ describe("OrchestratorMcpService provider resolution", () => { ), }), ServerSettings.layerTest({ + providerInstances: { + [opencodeInstanceId]: { + driver: "opencode", + config: {}, + enabled: true, + }, + }, favorites: [ { provider: codexInstanceId, model: "gpt-4" }, { provider: opencodeInstanceId, model: "oc/m10" }, @@ -1504,6 +1518,57 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); + it.effect("hides standalone providers disabled for the caller's project until reenabled", () => + Effect.gen(function* () { + const workInstanceId = ProviderInstanceId.make("codex_work"); + const driver = ProviderDriverKind.make("codex"); + const providers = [codexInstanceId, workInstanceId].map((instanceId) => + providerSnapshot({ instanceId, driver, model: "gpt-5.4" }), + ); + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: () => Effect.succeed(parentProjection([])), + }), + ServerSettings.layerTest({ + providerInstances: { + [workInstanceId]: { driver, config: {}, enabled: true }, + }, + projectSettingsOverrides: { + [projectId]: { providerInstances: { [codexInstanceId]: { enabled: false } } }, + }, + }), + providerRegistryLayer(providers), + adapterRegistryLayer([codexInstanceId, workInstanceId]), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const settings = yield* ServerSettings.ServerSettingsService; + assert.deepEqual( + (yield* service.capabilities(scope)).providers.map( + (provider) => provider.providerInstanceId, + ), + [workInstanceId], + ); + yield* settings.updateSettings({ + projectSettingsOverrides: { + [projectId]: { providerInstances: { [codexInstanceId]: { enabled: true } } }, + }, + }); + assert.deepEqual( + (yield* service.capabilities(scope)).providers.map( + (provider) => provider.providerInstanceId, + ), + [codexInstanceId, workInstanceId], + ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provideMerge(dependencies)))); + }), + ); + it.effect.each([ "all limited", "healthy capacity", @@ -2013,12 +2078,19 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); - it.live.each([ - { name: "parent instance", parentEnabled: true, model: "gpt-5.4" }, - { name: "alternate instance", parentEnabled: false, model: "gpt-5.5" }, - ])( - "replays accepted creation on the $name without authorizing new disabled work", - ({ parentEnabled, model }) => + it.live.each( + [ + { name: "parent instance", parentEnabled: true, model: "gpt-5.4" }, + { name: "alternate instance", parentEnabled: false, model: "gpt-5.5" }, + ].flatMap((scenario) => + (["disabled", "rate limited"] as const).map((availability) => ({ + ...scenario, + availability, + })), + ), + )( + "replays accepted creation on the $name without authorizing new $availability work", + ({ parentEnabled, model, availability }) => Effect.scoped( Effect.gen(function* () { const altInstanceId = ProviderInstanceId.make("codex-alt"); @@ -2061,6 +2133,10 @@ describe("OrchestratorMcpService provider resolution", () => { Layer.provide(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))), Layer.provideMerge(layerSettings), ); + const providers = [ + providerSnapshot({ instanceId: codexInstanceId, driver, model: "gpt-5.4" }), + providerSnapshot({ instanceId: altInstanceId, driver, model: "gpt-5.5" }), + ]; const layerTest = layerOrchestratorService.pipe( Layer.provideMerge(layerThreads), Layer.provide( @@ -2068,10 +2144,7 @@ describe("OrchestratorMcpService provider resolution", () => { NodeServices.layer, layerRegistry, layerSettings, - providerRegistryLayer([ - providerSnapshot({ instanceId: codexInstanceId, driver, model: "gpt-5.4" }), - providerSnapshot({ instanceId: altInstanceId, driver, model: "gpt-5.5" }), - ]), + providerRegistryLayer(providers), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), @@ -2137,16 +2210,22 @@ describe("OrchestratorMcpService provider resolution", () => { clientRequestId: "empty", }); - yield* settings.updateSettings({ - projectSettingsOverrides: { - [projectId]: { - providerInstances: { - [codexInstanceId]: { enabled: false }, - [altInstanceId]: { enabled: false }, + if (availability === "disabled") { + yield* settings.updateSettings({ + projectSettingsOverrides: { + [projectId]: { + providerInstances: { + [codexInstanceId]: { enabled: false }, + [altInstanceId]: { enabled: false }, + }, }, }, - }, - }); + }); + } else { + providers.forEach((provider, index) => { + providers[index] = { ...provider, limitedUntil: "2099-01-01T00:00:00.000Z" }; + }); + } assert.deepEqual(yield* service.createThreads(scope, withPrompt), started); const after = yield* orchestrator.getThreadProjection(child.threadId); assert.deepEqual(after.thread.modelSelection, selection); @@ -2166,8 +2245,11 @@ describe("OrchestratorMcpService provider resolution", () => { clientRequestId: "empty", }) .pipe(Effect.flip); - assert.equal(promptError.code, "invalid_request"); - assert.include(promptError.message, "disabled"); + assert.equal( + promptError.code, + availability === "disabled" ? "invalid_request" : "provider_unavailable", + ); + assert.include(promptError.message, availability); const untouched = yield* orchestrator.getThreadProjection(empty.threads[0]!.threadId); assert.equal(untouched.messages.length, 0); assert.equal(untouched.runs.length, 0); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 231aa92fe94b..d3aebcf6cbf1 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -66,6 +66,7 @@ import { type ServerProviderModel, ThreadId, isOrchestrationV2WorkActive, + resolveProjectProviderInstanceEnabled, } from "@t3tools/contracts"; import { getProviderOptionCurrentValue } from "@t3tools/shared/model"; import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -2085,6 +2086,15 @@ const make = Effect.gen(function* () { providers: providers .filter((provider) => provider.pool?.role !== "member") .flatMap((provider) => { + if ( + provider.pool?.role !== "pool" && + !resolveProjectProviderInstanceEnabled( + settings, + availability.projectId, + provider.instanceId, + ) + ) + return []; const constraints = delegationProviderConstraints( provider, orchestrationCapableInstanceIds.has(provider.instanceId), @@ -2588,7 +2598,17 @@ const make = Effect.gen(function* () { requestKey: key, index, }); + const messageCommandId = stableCommandId({ + scope, + requestKey: key, + operation: "dispatch-thread", + index, + }); const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId); + const isRetryOfAcceptedWork = + isRetryOfAcceptedCreation && + (request.prompt === undefined || + (yield* threadManagement.hasReceipt(messageCommandId))); const committedThread = isRetryOfAcceptedCreation ? yield* threadManagement .getThreadShell(threadId) @@ -2604,7 +2624,7 @@ const make = Effect.gen(function* () { providerInstanceId: committedThread.modelSelection.instanceId, }, providers, - skipProjectEnablementFilter: isRetryOfAcceptedCreation, + skipProjectEnablementFilter: isRetryOfAcceptedWork, }); const runtimeMode = yield* resolveRuntimeMode( parent.thread.runtimeMode, @@ -2658,12 +2678,7 @@ const make = Effect.gen(function* () { type: "message.dispatch", createdBy: "agent", creationSource: "mcp", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "dispatch-thread", - index, - }), + commandId: messageCommandId, threadId, senderThreadId: scope.thread.threadId, messageId: stableMessageId({ diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index e41dfdd59911..e1f411b8b0ea 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -337,6 +337,12 @@ it.effect( title: "Parent", }; const launched = yield* launches.launch({ ...input, delegatedFrom }); + const startedInput = launchInput({ + command: "command:started", + thread: "thread:started", + message: "Accepted work", + }); + const started = yield* launches.launch({ ...startedInput, delegatedFrom }); providers[0] = { ...providers[0]!, limitedUntil: "2099-01-01T00:00:00.000Z" }; const next = launchInput({ command: "command:next", thread: "thread:next" }); const refused = yield* launches.launch({ ...next, delegatedFrom }).pipe(Effect.result); @@ -345,6 +351,19 @@ it.effect( assert.equal(ordinary.threadId, next.threadId); const replay = yield* launches.launch({ ...input, delegatedFrom }); assert.equal(replay.threadId, launched.threadId); + const startedReplay = yield* launches.launch({ ...startedInput, delegatedFrom }); + assert.equal(startedReplay.threadId, started.threadId); + assert.equal(startedReplay.projection.messages.length, 1); + const freshWork = yield* launches + .launch({ + ...input, + delegatedFrom, + initialMessage: { text: "New work", attachments: [] }, + }) + .pipe(Effect.result); + assert.isTrue(Result.isFailure(freshWork)); + const untouched = yield* launches.launch({ ...input, delegatedFrom }); + assert.equal(untouched.projection.messages.length, 0); }).pipe(Effect.provide(harness.layer)); }, ); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index a2f64b5b69ab..840c08153997 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -838,9 +838,13 @@ const make = Effect.gen(function* () { } const launchReceipt = yield* readReceipt(input, input.commandId); + const delegationReceipt = + input.delegatedFrom !== undefined && input.initialMessage !== undefined + ? yield* readReceipt(input, CommandId.make(`${input.commandId}:initial-message`)) + : launchReceipt; if ( input.delegatedFrom !== undefined && - !(Option.isSome(launchReceipt) && launchReceipt.value.status === "accepted") + !(Option.isSome(delegationReceipt) && delegationReceipt.value.status === "accepted") ) { const providers = yield* providerRegistry.getPublishedProviders; const provider = providers.find(