diff --git a/apps/server/scripts/record-grok-acp-replay-fixture.ts b/apps/server/scripts/record-grok-acp-replay-fixture.ts index 49b667fecec12f794ce61cefd9d944bf7c780e8f..f04b04cfe348a7dd06015f6c55edc684fd77cf8f 100644 --- a/apps/server/scripts/record-grok-acp-replay-fixture.ts +++ b/apps/server/scripts/record-grok-acp-replay-fixture.ts @@ -21,7 +21,6 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { ChildProcessSpawner } from "effect/process"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; @@ -443,7 +442,6 @@ const recordScenario = Effect.fn("recordGrokScenario")(function* (fixtureName: s const settings = { ...DEFAULT_GROK_SETTINGS, binaryPath: process.env.T3_GROK_BIN ?? "grok" }; const layerRegistry = ProviderAdapterRegistry.layerFromAdaptersEffect( Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const environment = yield* HostProcessEnvironment; const adapter = yield* makeGrokAdapterV2({ instanceId: GROK_DEFAULT_INSTANCE_ID, @@ -460,7 +458,6 @@ const recordScenario = Effect.fn("recordGrokScenario")(function* (fixtureName: s interruptPromptOnCancel: input.interruptPromptOnCancel ?? false, grokSettings: settings, environment, - childProcessSpawner, runtimeMode: grokLaunchRuntimeMode(runtimePolicy), }), }); diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index a8c9677eeb0e36660effe879bc89f7b124472892..0802fef78f9efc46096f713287b25cfc4ee96be5 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -48,7 +48,7 @@ import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as Scope from "effect/Scope"; import * as TestClock from "effect/testing/TestClock"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import type * as EffectAcpSchema from "effect-acp/compat"; @@ -1150,7 +1150,7 @@ describe("AcpAdapterV2", () => { const adapter = yield* makeAcpAdapterV2({ instanceId, // Production Devin runs commands through client terminals. - clientTerminals: { childProcessSpawner, shellCommands: true }, + clientTerminals: { shellCommands: true }, selfInvocation: yield* resolveSelfInvocation(), flavor: { driver: ACP_TEST_DRIVER, @@ -2190,7 +2190,7 @@ describe("AcpAdapterV2", () => { makeRuntime, }, selfInvocation, - clientTerminals: { childProcessSpawner }, + clientTerminals: {}, }); const sourceThreadId = ThreadId.make("thread-acp-native-fork-source"); const targetThreadId = ThreadId.make("thread-acp-native-fork-target"); @@ -2476,7 +2476,7 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.andThen(makeRuntime(input))), }, selfInvocation, - clientTerminals: { childProcessSpawner }, + clientTerminals: {}, }); const threadId = ThreadId.make("thread-acp-unknown-permission-grant"); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ @@ -12045,9 +12045,11 @@ describe("AcpAdapterV2", () => { Effect.gen(function* () { const commands: Array<{ readonly command: string; readonly args: ReadonlyArray }> = []; - yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill( - makeTaskkillSpawner({ exitCode: 0, commands }), - 4321, + yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill(4321).pipe( + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + makeTaskkillSpawner({ exitCode: 0, commands }), + ), ); assert.deepEqual(commands, [{ command: "taskkill", args: ["/PID", "4321", "/T", "/F"] }]); @@ -12058,10 +12060,13 @@ describe("AcpAdapterV2", () => { { exitCode: 128, output: "ERROR: Access is denied." }, { exitCode: 1, output: "generic failure" }, ]) { - const failed = yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill( - makeTaskkillSpawner(fixture), - 4321, - ).pipe(Effect.exit); + const failed = yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill(4321).pipe( + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + makeTaskkillSpawner(fixture), + ), + Effect.exit, + ); if (Exit.isSuccess(failed)) assert.fail(`taskkill ${fixture.exitCode} must fail`); const error = Cause.squash(failed.cause); assert.instanceOf(error, AcpSessionRuntime.AcpProcessGroupTerminationError); @@ -12082,10 +12087,13 @@ describe("AcpAdapterV2", () => { { outputFailure: true }, { exitFailure: true }, ]) { - const failed = yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill( - makeTaskkillSpawner(fixture), - 4321, - ).pipe(Effect.exit); + const failed = yield* AcpSessionRuntime.terminateWindowsProcessTreeWithTaskkill(4321).pipe( + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + makeTaskkillSpawner(fixture), + ), + Effect.exit, + ); if (Exit.isSuccess(failed)) assert.fail("taskkill infrastructure failure must fail"); const error = Cause.squash(failed.cause); assert.instanceOf(error, AcpSessionRuntime.AcpProcessGroupTerminationError); diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts index 8a4310c3d542c478131397950be2df899b4444e8..d95506e2a005c5eba4feee592c276a5f15f7fb8f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.testkit.ts @@ -10,7 +10,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index e7327f034db7904546ce751a1611541f52f519f3..171f7ae7eedc72d0eb9458716c4883008cca5763 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -9,8 +9,9 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { HttpClient, HttpClientResponse } from "effect/http"; -import { ChildProcessSpawner } from "effect/process"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; @@ -22,6 +23,7 @@ import type { AcpRegistryLiveConfiguration, } from "@t3tools/provider-acp-registry/testing"; import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; +import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; import { ACP_SESSION_MODE_OPTION_ID } from "@t3tools/provider-acp/server/sessionConfig"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -82,6 +84,11 @@ const layerRegistry = Layer.succeed( ), ); +// Adapters given an explicit runtime never resolve through the catalog. +const layerInjectedRuntimeCatalog = Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({ + resolve: () => Effect.die("the runtime is injected"), +}); + const layerTest = Layer.mergeAll( NodeServices.layer, IdAllocator.layer, @@ -197,8 +204,6 @@ describe("AcpRegistryAdapterV2", () => { authMethodId: "test", }), environment: {}, - childProcessSpawner, - resolver: { resolve: () => Effect.die("the runtime is injected") }, makeRuntime: makeAcpReplayRuntime({ transcript, statusPath, @@ -208,7 +213,7 @@ describe("AcpRegistryAdapterV2", () => { childProcessSpawner, fileSystem, }), - }); + }).pipe(Effect.provide(layerInjectedRuntimeCatalog)); yield* adapter .openSession({ threadId: ThreadId.make("thread-acp-registry-mode-pick"), @@ -279,8 +284,6 @@ describe("AcpRegistryAdapterV2", () => { instanceId, settings: yield* decodeAcpRegistryAdapterSettings({ agentId, authMethodId: "test" }), environment: {}, - childProcessSpawner, - resolver: { resolve: () => Effect.die("the runtime is injected") }, makeRuntime: (input) => Effect.gen(function* () { clientCapabilities = input.clientCapabilities; @@ -305,7 +308,7 @@ describe("AcpRegistryAdapterV2", () => { Effect.provide(context), ); }), - }); + }).pipe(Effect.provide(layerInjectedRuntimeCatalog)); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", interactionMode: "default", @@ -333,7 +336,6 @@ describe("AcpRegistryAdapterV2", () => { it.effect("opens a real ACP child process resolved from registry configuration", () => Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; const host = yield* ProviderHost.ProviderHost; const mockAgentPath = yield* path.fromFileUrl( @@ -365,55 +367,59 @@ describe("AcpRegistryAdapterV2", () => { T3_ACP_SESSION_LIFECYCLE: "1", T3_ACP_COMMAND_ADVERTISEMENT_DELAY_MS: "750", }, - childProcessSpawner, - runtimeCoordinator: { - withForegroundStartup: (agentId, effect) => - Effect.acquireUseRelease( - Effect.sync(() => { - assert.equal(agentId, "fixture-agent"); - startupActive = true; - startupCount += 1; - }), - () => effect, - () => - Effect.sync(() => { - startupActive = false; - }), - ), - runBackgroundProbe: (_agentId, effect) => effect.pipe(Effect.map(Option.some)), - withSessionMutation: (effect) => effect, - clearAvailableCommands: () => Effect.void, - publishAvailableCommands: (publishedInstanceId, commands) => - Deferred.succeed(commandsPublished, { - instanceId: publishedInstanceId, - commands, - }).pipe(Effect.asVoid), - getAvailableCommands: () => Effect.succeed(Option.none()), - watchAvailableCommands: () => Effect.never, - clearLiveConfiguration: () => Effect.void, - publishLiveConfiguration: (_publishedInstanceId, configuration) => - Deferred.succeed(configurationPublished, configuration).pipe(Effect.asVoid), - getLiveConfiguration: () => Effect.succeed(Option.none()), - watchLiveConfiguration: () => Effect.never, - requestUrlAuthentication: () => Effect.succeed(false), - acceptUrlAuthentication: () => Effect.succeed(false), - getUrlAuthAction: () => Effect.succeed(Option.none()), - watchUrlAuthAction: () => Effect.never, - }, - resolver: { - resolve: (configuredSettings, cwd, environment) => - Effect.sync(() => assert.isTrue(startupActive)).pipe( - Effect.andThen(resolver.resolve(configuredSettings, cwd, environment)), - Effect.map((resolved) => ({ - ...resolved, - spawn: { - ...resolved.spawn, - args: [mockAgentPath], - }, - })), - ), - }, - }); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.mock(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator)({ + withForegroundStartup: (agentId, effect) => + Effect.acquireUseRelease( + Effect.sync(() => { + assert.equal(agentId, "fixture-agent"); + startupActive = true; + startupCount += 1; + }), + () => effect, + () => + Effect.sync(() => { + startupActive = false; + }), + ), + runBackgroundProbe: (_agentId, effect) => effect.pipe(Effect.map(Option.some)), + withSessionMutation: (effect) => effect, + clearAvailableCommands: () => Effect.void, + publishAvailableCommands: (publishedInstanceId, commands) => + Deferred.succeed(commandsPublished, { + instanceId: publishedInstanceId, + commands, + }).pipe(Effect.asVoid), + getAvailableCommands: () => Effect.succeed(Option.none()), + watchAvailableCommands: () => Effect.never, + clearLiveConfiguration: () => Effect.void, + publishLiveConfiguration: (_publishedInstanceId, configuration) => + Deferred.succeed(configurationPublished, configuration).pipe(Effect.asVoid), + getLiveConfiguration: () => Effect.succeed(Option.none()), + watchLiveConfiguration: () => Effect.never, + requestUrlAuthentication: () => Effect.succeed(false), + acceptUrlAuthentication: () => Effect.succeed(false), + getUrlAuthAction: () => Effect.succeed(Option.none()), + watchUrlAuthAction: () => Effect.never, + }), + Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({ + resolve: (configuredSettings, cwd, environment) => + Effect.sync(() => assert.isTrue(startupActive)).pipe( + Effect.andThen(resolver.resolve(configuredSettings, cwd, environment)), + Effect.map((resolved) => ({ + ...resolved, + spawn: { + ...resolved.spawn, + args: [mockAgentPath], + }, + })), + ), + }), + ), + ), + ); const threadId = ThreadId.make("thread-acp-registry-fixture"); const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.testkit.ts index 93254166f183ab5ab15df16a5c3f48e09bd5782d..fb4c93aba7ddc403d2e830a6e2f8c4d53bfb60a0 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.testkit.ts @@ -6,7 +6,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -25,6 +25,7 @@ import { ACP_REGISTRY_PROVIDER, makeAcpRegistryAdapterV2, } from "@t3tools/provider-acp-registry/testing"; +import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; const REPLAY_SETTINGS = Schema.decodeUnknownSync(AcpRegistrySettings)({ agentId: "replay-agent", @@ -56,10 +57,6 @@ function layerAcpRegistryProviderAdapterRegistryReplay( instanceId: ACP_REGISTRY_DEFAULT_INSTANCE_ID, settings: REPLAY_SETTINGS, environment: {}, - childProcessSpawner, - resolver: { - resolve: () => Effect.die("ACP registry resolver must not run during replay"), - }, selfInvocation: yield* resolveSelfInvocation(), makeRuntime: makeAcpReplayRuntime({ transcript, @@ -74,7 +71,16 @@ function layerAcpRegistryProviderAdapterRegistryReplay( return [adapter]; }), ).pipe( - Layer.provide(Layer.mergeAll(layerHost, NodeServices.layer, IdAllocator.layer)), + Layer.provide( + Layer.mergeAll( + layerHost, + NodeServices.layer, + IdAllocator.layer, + Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({ + resolve: () => Effect.die("ACP registry resolver must not run during replay"), + }), + ), + ), // Held inbound lines must not outlive the scenario and wedge teardown. Layer.merge( Layer.effectDiscard( diff --git a/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.test.ts index 86a6df752f6933a5f4afa9a8c6db761447e11464..732bc6904f40dbb136adcfa46828c6169bde9f0c 100644 --- a/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.test.ts @@ -17,7 +17,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import type * as EffectAcpSchema from "effect-acp/compat"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; diff --git a/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.testkit.ts index 443019cea47283128d0d0e56eb1349a9b5aac2dc..b6b67c7a0d56fb42d87558c2c81783aea7b5b6c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/GrokAdapterV2.testkit.ts @@ -5,7 +5,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { resolveSelfInvocation } from "@t3tools/shared/nodeRuntime"; diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index 562307477dcf7a045bd073748e907939a8c4612b..8aa64408993e36d884e37549105dc8160b290189 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -2253,13 +2253,14 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func options.managedBinaryToolsDir !== undefined ) { const managedDirectories = yield* acpRegistryManagedBinaryDirectories({ - fileSystem, - path, cacheDir: options.managedBinaryCacheDir, toolsDir: options.managedBinaryToolsDir, platform, architecture, - }); + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ); if (managedDirectories.length > 0) { const delimiter = platform === "win32" ? ";" : ":"; const pathKey = diff --git a/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts b/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts index 960602c8cc947f7a05bb1715c67fd062b71e5bb6..62fd6274e64d8a87c6fc846a75dbaf7d36ce606a 100644 --- a/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts +++ b/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts @@ -19,9 +19,11 @@ import * as Path from "effect/Path"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; -import { HttpClient, HttpClientResponse } from "effect/http"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; import * as TestClock from "effect/testing/TestClock"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as AcpRegistrySupport from "./AcpRegistrySupport.ts"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; @@ -1713,8 +1715,6 @@ describe("acpRegistryManagedBinaryDirectories", () => { ); const directories = yield* AcpRegistrySupport.acpRegistryManagedBinaryDirectories({ - fileSystem, - path, cacheDir, toolsDir: path.join(cacheDir, "tools"), platform: "linux", @@ -1727,8 +1727,6 @@ describe("acpRegistryManagedBinaryDirectories", () => { ]); const missing = yield* AcpRegistrySupport.acpRegistryManagedBinaryDirectories({ - fileSystem, - path, cacheDir: path.join(cacheDir, "does-not-exist"), toolsDir: path.join(cacheDir, "does-not-exist", "tools"), platform: "linux", diff --git a/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts b/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts index 9055f9e070ebbba03e3f861edaf5b0f5dfd2430a..fba8e656e615a5728385b236986f45796632a32a 100644 --- a/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts +++ b/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts @@ -39,8 +39,11 @@ import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { sha256 } from "@noble/hashes/sha2"; -import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { collectUint8StreamText } from "@t3tools/provider-core/server/collectStreamText"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; @@ -308,26 +311,26 @@ function withPreferredPath( * managed binary agents are both available by name. Best effort: unreadable * directories resolve to no entries. */ -export const acpRegistryManagedBinaryDirectories = (input: { - readonly fileSystem: FileSystem.FileSystem; - readonly path: Path.Path; - readonly cacheDir: string; - readonly toolsDir: string; - readonly platform: NodeJS.Platform; - readonly architecture: NodeJS.Architecture; -}): Effect.Effect> => - Effect.gen(function* () { +export const acpRegistryManagedBinaryDirectories = Effect.fn("acpRegistryManagedBinaryDirectories")( + function* (input: { + readonly cacheDir: string; + readonly toolsDir: string; + readonly platform: NodeJS.Platform; + readonly architecture: NodeJS.Architecture; + }): Effect.fn.Return, never, FileSystem.FileSystem | Path.Path> { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const target = resolveAcpRegistryPlatformTarget(input.platform, input.architecture); - const registryDirectory = input.path.join(input.cacheDir, "acp-registry"); + const registryDirectory = path.join(input.cacheDir, "acp-registry"); const installsDirectory = input.toolsDir; - const packageReceiptsDirectory = input.path.join(registryDirectory, "package-installs"); + const packageReceiptsDirectory = path.join(registryDirectory, "package-installs"); const listDirectories = (directory: string) => - input.fileSystem.readDirectory(directory).pipe(Effect.orElseSucceed((): Array => [])); + fileSystem.readDirectory(directory).pipe(Effect.orElseSucceed((): Array => [])); const directories: Array = []; for (const receiptFile of (yield* listDirectories(packageReceiptsDirectory)).toSorted()) { - const receipt = yield* input.fileSystem - .readFileString(input.path.join(packageReceiptsDirectory, receiptFile)) + const receipt = yield* fileSystem + .readFileString(path.join(packageReceiptsDirectory, receiptFile)) .pipe( Effect.map(decodePackageInstallReceipt), Effect.orElseSucceed(() => Option.none()), @@ -335,15 +338,15 @@ export const acpRegistryManagedBinaryDirectories = (input: { if ( Option.isSome(receipt) && receipt.value.binDirectory === - input.path.join( + path.join( installsDirectory, receipt.value.agentId, encodeURIComponent(receipt.value.agentVersion), receipt.value.distribution === "npx" ? "npm" : "python", ...(receipt.value.distribution === "npx" && input.platform === "win32" ? [] : ["bin"]), ) && - input.path.dirname(receipt.value.executablePath) === receipt.value.binDirectory && - (yield* input.fileSystem + path.dirname(receipt.value.executablePath) === receipt.value.binDirectory && + (yield* fileSystem .exists(receipt.value.executablePath) .pipe(Effect.orElseSucceed(() => false))) ) { @@ -352,8 +355,8 @@ export const acpRegistryManagedBinaryDirectories = (input: { } if (target === undefined) return Array.from(new Set(directories)); - const cachedAgents = yield* input.fileSystem - .readFileString(input.path.join(registryDirectory, "registry.json")) + const cachedAgents = yield* fileSystem + .readFileString(path.join(registryDirectory, "registry.json")) .pipe( Effect.flatMap(decodeJson), Effect.flatMap(decodeRegistryIndexEnvelope), @@ -372,12 +375,12 @@ export const acpRegistryManagedBinaryDirectories = (input: { ), ); for (const agent of (yield* listDirectories(installsDirectory)).toSorted()) { - const versions = (yield* listDirectories(input.path.join(installsDirectory, agent))).toSorted( + const versions = (yield* listDirectories(path.join(installsDirectory, agent))).toSorted( (left, right) => right.localeCompare(left, undefined, { numeric: true }), ); for (const version of versions) { - const installRoot = input.path.join(installsDirectory, agent, version, target); - if (yield* input.fileSystem.exists(installRoot).pipe(Effect.orElseSucceed(() => false))) { + const installRoot = path.join(installsDirectory, agent, version, target); + if (yield* fileSystem.exists(installRoot).pipe(Effect.orElseSucceed(() => false))) { const binaryTarget = cachedAgentByInstall.get(`${agent}\0${version}`)?.distribution .binary?.[target]; const commandSegments = @@ -385,15 +388,16 @@ export const acpRegistryManagedBinaryDirectories = (input: { const directory = commandSegments === undefined ? installRoot - : input.path.join(installRoot, ...commandSegments.slice(0, -1)); - if (yield* input.fileSystem.exists(directory).pipe(Effect.orElseSucceed(() => false))) { + : path.join(installRoot, ...commandSegments.slice(0, -1)); + if (yield* fileSystem.exists(directory).pipe(Effect.orElseSucceed(() => false))) { directories.push(directory); } } } } return Array.from(new Set(directories)); - }); + }, +); export interface ResolvedAcpRegistryDistribution { readonly kind: AcpRegistryDistributionKind; diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index f9dc480890d8e91318a3adce376cd15f96316115..872eb099980207ba2cecfe69cc1a19b89a101a2a 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -15,7 +15,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -56,17 +56,14 @@ export interface AcpRegistryAdapterV2Options { readonly instanceId: Parameters[0]["instanceId"]; readonly settings: AcpRegistrySettings; readonly environment: NodeJS.ProcessEnv; - readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; readonly selfInvocation: SelfInvocation; - readonly resolver: Pick; - readonly runtimeCoordinator?: AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator["Service"]; readonly nativeLogging?: Parameters[0]["nativeLogging"]; readonly makeRuntime?: ( input: AcpAdapterV2RuntimeInput, ) => Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, - Crypto.Crypto | Scope.Scope + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope >; readonly assertComplete?: Effect.Effect; } @@ -131,17 +128,20 @@ export function acpRegistryPromptFailure(agentId: string, cause: unknown) { }); } -function makeAcpRegistryRuntime(options: AcpRegistryAdapterV2Options) { +function makeAcpRegistryRuntime( + options: AcpRegistryAdapterV2Options, + catalog: AcpRegistrySupport.AcpRegistryCatalog["Service"], +) { return ( input: AcpAdapterV2RuntimeInput, ): Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, - Crypto.Crypto | Scope.Scope + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope > => Effect.gen(function* () { const { processEnvironment, ...runtimeInput } = input; - const resolved = yield* options.resolver + const resolved = yield* catalog .resolve(options.settings, input.cwd, options.environment) .pipe( Effect.mapError( @@ -163,11 +163,7 @@ function makeAcpRegistryRuntime(options: AcpRegistryAdapterV2Options) { env: { ...resolved.spawn.env, ...processEnvironment }, }, ...(options.settings.authMethodId ? { authMethodId: options.settings.authMethodId } : {}), - }).pipe( - Layer.provide( - Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, options.childProcessSpawner), - ), - ), + }), ); return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe( Effect.provide(context), @@ -178,7 +174,10 @@ function makeAcpRegistryRuntime(options: AcpRegistryAdapterV2Options) { export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(function* ( options: AcpRegistryAdapterV2Options, ) { - const runtimeCoordinator = options.runtimeCoordinator; + const catalog = yield* AcpRegistrySupport.AcpRegistryCatalog; + const runtimeCoordinator = Option.getOrUndefined( + yield* Effect.serviceOption(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator), + ); const registryAgentId = options.settings.source === "local" ? "" : options.settings.agentId; const startupKey = options.settings.source === "local" ? `local:${options.instanceId}` : registryAgentId; @@ -203,7 +202,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu extractSubagentUpdate: extractDevinSubagentUpdate, } : {}), - makeRuntime: options.makeRuntime ?? makeAcpRegistryRuntime(options), + makeRuntime: options.makeRuntime ?? makeAcpRegistryRuntime(options, catalog), ...(runtimeCoordinator === undefined ? {} : { @@ -243,7 +242,6 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu ...(isDevin ? { clientTerminals: { - childProcessSpawner: options.childProcessSpawner, environment: options.environment, shellCommands: true, }, @@ -275,22 +273,12 @@ export const AcpRegistryAdapterV2Driver: ProviderAdapterDriver< function* (input: ProviderAdapterDriverCreateInput) { const hostEnvironment = yield* HostProcessEnvironment; const selfInvocation = yield* resolveSelfInvocation(); - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const providerEventLoggers = yield* ProviderEventLoggers.ProviderEventLoggers; const makeNativeLogger = yield* makeAcpNativeLoggerFactory(); - const resolver = yield* AcpRegistrySupport.AcpRegistryCatalog; - const runtimeCoordinator = yield* Effect.serviceOption( - AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator, - ); return yield* makeAcpRegistryAdapterV2({ instanceId: input.instanceId, settings: { ...input.config, enabled: input.enabled }, environment: mergeProviderInstanceEnvironment(input.environment, hostEnvironment), - childProcessSpawner, - resolver, - ...(Option.isSome(runtimeCoordinator) - ? { runtimeCoordinator: runtimeCoordinator.value } - : {}), selfInvocation, nativeLogging: (threadId) => makeNativeLogger({ diff --git a/packages/provider-acp-registry/src/server/auth.ts b/packages/provider-acp-registry/src/server/auth.ts index 72993a7a5f896977b5eb5ccce612b94984b25743..ded5117c37b52d1f139808a5c6e1b6697457d39d 100644 --- a/packages/provider-acp-registry/src/server/auth.ts +++ b/packages/provider-acp-registry/src/server/auth.ts @@ -12,7 +12,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Scope from "effect/Scope"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import type * as AcpSchema from "effect-acp/compat"; import * as PtyAdapter from "@t3tools/shared/PtyAdapter"; diff --git a/packages/provider-acp-registry/src/server/driver.ts b/packages/provider-acp-registry/src/server/driver.ts index 0e699bee5bf8a07913a1369a236e6bafe46a7a3c..bcfd83af7a399785cd490a28f83082bfb73f84ee 100644 --- a/packages/provider-acp-registry/src/server/driver.ts +++ b/packages/provider-acp-registry/src/server/driver.ts @@ -19,7 +19,7 @@ import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import { AcpRegistryAdapterV2Driver, type AcpRegistryAdapterV2DriverEnv } from "./adapter.ts"; diff --git a/packages/provider-acp-registry/src/server/probe.ts b/packages/provider-acp-registry/src/server/probe.ts index c88dacb8f063dd0271749d074584b04fc268f8ba..e8cead21804faecfd4dac6c777c2c65fe21fc0c4 100644 --- a/packages/provider-acp-registry/src/server/probe.ts +++ b/packages/provider-acp-registry/src/server/probe.ts @@ -19,7 +19,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import type * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpSchema from "effect-acp/compat"; diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index ae3423c6f921c0d0a6dd29a8d8fb8f9b4711f5f8..9891ba50ded22c25f35fa2f0a9b1fbd3360c1bb3 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -1055,11 +1055,11 @@ export function windowsTaskkillResultIsSuccess(exitCode: number, _output?: strin } export const terminateWindowsProcessTreeWithTaskkill = ( - spawner: ChildProcessSpawner.ChildProcessSpawner["Service"], pid: number, -): Effect.Effect => +): Effect.Effect => Effect.scoped( Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const taskkill = yield* spawner.spawn( ChildProcess.make("taskkill", ["/PID", String(pid), "/T", "/F"]), ); @@ -1659,7 +1659,10 @@ export const make = ( ); const terminateWindowsProcessTree = options.windowsProcessTreeTerminator ?? - ((pid: number) => terminateWindowsProcessTreeWithTaskkill(spawner, pid)); + ((pid: number) => + terminateWindowsProcessTreeWithTaskkill(pid).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + )); const terminatePosixProcessTree = (grace: Duration.Input, platformOverride?: NodeJS.Platform) => Effect.gen(function* () { const platform = platformOverride ?? options.processGroupPlatform; @@ -1728,7 +1731,9 @@ export const make = ( const hostPlatform = yield* HostProcessPlatform; const forceTerminateOwnedProcessGroup = hostPlatform === "win32" - ? terminateWindowsProcessTreeWithTaskkill(spawner, Number(child.pid)) + ? terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ) : linuxCgroupLease !== undefined ? terminateLinuxCgroupLease(linuxCgroupLease) : options.ownDescendantProcessGroups === true diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 86b560f0cda30eec856e6fb1927ed7cb09c9c59b..5f5d43e56e5c9453dd7269b9779db4578560afd3 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -50,7 +50,7 @@ import * as Result from "effect/Result"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; -import type { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import type * as EffectAcpSchema from "effect-acp/compat"; @@ -239,7 +239,7 @@ export interface AcpAdapterV2Flavor { ) => Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, - Crypto.Crypto | Scope.Scope + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope >; readonly resolveModelId?: (selection: ModelSelection) => string | undefined; /** @@ -485,11 +485,10 @@ export interface AcpAdapterV2Options { /** * Opts the session into the ACP client `terminal` capability. Agents run * commands themselves unless an adapter sets this; with it, sessions - * advertise `terminal: true` and run agent-created terminals through this - * spawner with the provider instance's environment. Devin sets it. + * advertise `terminal: true` and run agent-created terminals through the + * adapter's spawner with the provider instance's environment. Devin sets it. */ readonly clientTerminals?: { - readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; readonly environment?: NodeJS.ProcessEnv; readonly shellCommands?: boolean; }; @@ -1486,6 +1485,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const crypto = yield* Crypto.Crypto; const fileSystem = yield* FileSystem.FileSystem; + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const idAllocator = yield* IdAllocator.IdAllocatorV2; const host = yield* ProviderHost.ProviderHost; const mcpSessions = yield* McpProviderSessions.McpProviderSessions; @@ -1586,7 +1586,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( options.clientTerminals === undefined ? undefined : yield* makeAcpClientTerminals({ - spawner: options.clientTerminals.childProcessSpawner, defaultCwd: input.runtimePolicy.cwd ?? process.cwd(), environment: options.clientTerminals.environment, shellCommands: options.clientTerminals.shellCommands, @@ -1607,7 +1606,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ? pendingTerminalEnvironment.environment : undefined; }, - }); + }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), + ); if (clientTerminals !== undefined) { yield* Scope.addFinalizer(sessionScope, clientTerminals.disposeAll); } @@ -6018,6 +6019,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( .pipe( Effect.provideService(Scope.Scope, runtimeScope), Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), ); }); @@ -6096,6 +6098,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( .pipe( Effect.provideService(Scope.Scope, replacementScope), Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), ); // Session setup may publish commands before it returns. Buffer those // notifications, but do not expose request or extension handlers diff --git a/packages/provider-acp/src/server/clientTerminals.test.ts b/packages/provider-acp/src/server/clientTerminals.test.ts index 83b5273a21add58c97a5e8cd51faabf33a099735..d2790461b595a8c4b2190f6604931c7e1a1c81ca 100644 --- a/packages/provider-acp/src/server/clientTerminals.test.ts +++ b/packages/provider-acp/src/server/clientTerminals.test.ts @@ -2,7 +2,6 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Result from "effect/Result"; -import { ChildProcessSpawner } from "effect/process"; import { makeAcpClientTerminals, @@ -12,9 +11,7 @@ import { const withTerminals = (use: (terminals: AcpClientTerminals) => Effect.Effect) => Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const terminals = yield* makeAcpClientTerminals({ - spawner, defaultCwd: process.cwd(), }); return yield* use(terminals).pipe(Effect.ensuring(terminals.disposeAll)); @@ -49,9 +46,7 @@ describe("AcpClientTerminals", () => { it.effect("resolves terminal environment from the requesting ACP session", () => Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const terminals = yield* makeAcpClientTerminals({ - spawner, defaultCwd: process.cwd(), environmentForSession: (sessionId) => ({ T3_TEST_ACP_SESSION_TOKEN: sessionId === "session-a" ? "token-a" : "token-b", @@ -78,7 +73,6 @@ describe("AcpClientTerminals", () => { it.effect("runs Devin shell source with arguments, pipes, and session fallback variables", () => Effect.gen(function* () { const terminals = yield* makeAcpClientTerminals({ - spawner: yield* ChildProcessSpawner.ChildProcessSpawner, defaultCwd: process.cwd(), shellCommands: true, environmentForSession: () => ({ T3_ACP_MCP_NODE: "mailbox-probe" }), diff --git a/packages/provider-acp/src/server/clientTerminals.ts b/packages/provider-acp/src/server/clientTerminals.ts index c7743d60daae7739f46f154f03da3c8efd1dbc05..ce6b3d0436ae5b98a4e793f01b826e17c0dec9e8 100644 --- a/packages/provider-acp/src/server/clientTerminals.ts +++ b/packages/provider-acp/src/server/clientTerminals.ts @@ -6,7 +6,8 @@ import * as Fiber from "effect/Fiber"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpSchema from "effect-acp/compat"; import * as NodeBuffer from "node:buffer"; @@ -137,7 +138,6 @@ export interface AcpClientTerminals { export interface AcpClientTerminalsOptions { /** Devin sends shell source in command, rather than an executable plus args. */ readonly shellCommands?: boolean | undefined; - readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; readonly defaultCwd: string; readonly environment?: NodeJS.ProcessEnv | undefined; readonly environmentForSession?: @@ -161,272 +161,261 @@ function acpTerminalCommand(input: { }); } -export const makeAcpClientTerminals = ( +export const makeAcpClientTerminals = Effect.fn("makeAcpClientTerminals")(function* ( options: AcpClientTerminalsOptions, -): Effect.Effect => - Effect.gen(function* () { - const creationPermit = yield* Semaphore.make(1); - const terminals = new Map(); - let nextTerminalNumber = 1; +): Effect.fn.Return { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const creationPermit = yield* Semaphore.make(1); + const terminals = new Map(); + let nextTerminalNumber = 1; - const enforceTotalOutputLimit = (): void => { - let excess = - Array.from(terminals.values()).reduce( - (total, terminal) => total + terminal.buffer.bytes, - 0, - ) - MAX_RETAINED_OUTPUT_BYTES; + const enforceTotalOutputLimit = (): void => { + let excess = + Array.from(terminals.values()).reduce((total, terminal) => total + terminal.buffer.bytes, 0) - + MAX_RETAINED_OUTPUT_BYTES; + if (excess <= 0) return; + for (const terminal of terminals.values()) { + const before = terminal.buffer.bytes; + trimOutputStart(terminal.buffer, excess); + excess -= before - terminal.buffer.bytes; if (excess <= 0) return; - for (const terminal of terminals.values()) { - const before = terminal.buffer.bytes; - trimOutputStart(terminal.buffer, excess); - excess -= before - terminal.buffer.bytes; - if (excess <= 0) return; - } - }; + } + }; - const pruneReleasedSnapshots = (): void => { - const released = Array.from(terminals.values()).filter((terminal) => terminal.released); - let retainedBytes = released.reduce((total, terminal) => total + terminal.buffer.bytes, 0); - while ( - released.length > MAX_RETAINED_TERMINALS || - retainedBytes > MAX_RETAINED_OUTPUT_BYTES - ) { - const terminal = released.shift(); - if (terminal === undefined) break; - retainedBytes -= terminal.buffer.bytes; - terminals.delete(terminal.terminalId); - } - }; + const pruneReleasedSnapshots = (): void => { + const released = Array.from(terminals.values()).filter((terminal) => terminal.released); + let retainedBytes = released.reduce((total, terminal) => total + terminal.buffer.bytes, 0); + while (released.length > MAX_RETAINED_TERMINALS || retainedBytes > MAX_RETAINED_OUTPUT_BYTES) { + const terminal = released.shift(); + if (terminal === undefined) break; + retainedBytes -= terminal.buffer.bytes; + terminals.delete(terminal.terminalId); + } + }; - const requireTerminal = ( - sessionId: string, - terminalId: string, - operation: string, - ): Effect.Effect => { - const terminal = terminals.get(terminalId); - if (terminal === undefined || terminal.sessionId !== sessionId || terminal.released) { - return Effect.fail( - terminalRequestError(`ACP ${operation} received an unknown terminal ID '${terminalId}'.`), - ); - } - return Effect.succeed(terminal); - }; + const requireTerminal = ( + sessionId: string, + terminalId: string, + operation: string, + ): Effect.Effect => { + const terminal = terminals.get(terminalId); + if (terminal === undefined || terminal.sessionId !== sessionId || terminal.released) { + return Effect.fail( + terminalRequestError(`ACP ${operation} received an unknown terminal ID '${terminalId}'.`), + ); + } + return Effect.succeed(terminal); + }; - const create: AcpClientTerminals["create"] = (request) => - creationPermit.withPermit( - Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const liveCount = Array.from(terminals.values()).filter( - (terminal) => !terminal.released && terminal.exitStatus === undefined, - ).length; - const unreleasedCount = Array.from(terminals.values()).filter( - (terminal) => !terminal.released, - ).length; - if (unreleasedCount >= MAX_UNRELEASED_TERMINALS) { - return yield* terminalRequestError( - `ACP terminal/create exceeded the limit of ${MAX_UNRELEASED_TERMINALS} unreleased terminals.`, - ); - } - if (liveCount >= MAX_LIVE_TERMINALS) { - return yield* terminalRequestError( - `ACP terminal/create exceeded the limit of ${MAX_LIVE_TERMINALS} concurrent terminals.`, - ); - } - const requestEnvironment = Object.fromEntries( - (request.env ?? []).map((variable) => [variable.name, variable.value] as const), + const create: AcpClientTerminals["create"] = (request) => + creationPermit.withPermit( + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const liveCount = Array.from(terminals.values()).filter( + (terminal) => !terminal.released && terminal.exitStatus === undefined, + ).length; + const unreleasedCount = Array.from(terminals.values()).filter( + (terminal) => !terminal.released, + ).length; + if (unreleasedCount >= MAX_UNRELEASED_TERMINALS) { + return yield* terminalRequestError( + `ACP terminal/create exceeded the limit of ${MAX_UNRELEASED_TERMINALS} unreleased terminals.`, + ); + } + if (liveCount >= MAX_LIVE_TERMINALS) { + return yield* terminalRequestError( + `ACP terminal/create exceeded the limit of ${MAX_LIVE_TERMINALS} concurrent terminals.`, ); - const sessionEnvironment = options.environmentForSession?.(request.sessionId); - const environment = - options.environment === undefined && - sessionEnvironment === undefined && - (request.env?.length ?? 0) === 0 - ? undefined - : { ...options.environment, ...sessionEnvironment, ...requestEnvironment }; - // Each terminal owns a scope so the spawned process is reliably reaped: - // closing the scope kills a still-running command and frees the handle. - const terminalScope = yield* Scope.make(); - const child = yield* restore( - options.spawner - .spawn( - acpTerminalCommand({ - request, - shellCommands: options.shellCommands, - defaultCwd: options.defaultCwd, - environment, - forceKillAfter: options.forceKillAfter, - }), - ) - .pipe( - Effect.provideService(Scope.Scope, terminalScope), - Effect.mapError((cause) => - terminalRequestError( - `Could not start terminal command '${request.command}'.`, - cause, - ), + } + const requestEnvironment = Object.fromEntries( + (request.env ?? []).map((variable) => [variable.name, variable.value] as const), + ); + const sessionEnvironment = options.environmentForSession?.(request.sessionId); + const environment = + options.environment === undefined && + sessionEnvironment === undefined && + (request.env?.length ?? 0) === 0 + ? undefined + : { ...options.environment, ...sessionEnvironment, ...requestEnvironment }; + // Each terminal owns a scope so the spawned process is reliably reaped: + // closing the scope kills a still-running command and frees the handle. + const terminalScope = yield* Scope.make(); + const child = yield* restore( + spawner + .spawn( + acpTerminalCommand({ + request, + shellCommands: options.shellCommands, + defaultCwd: options.defaultCwd, + environment, + forceKillAfter: options.forceKillAfter, + }), + ) + .pipe( + Effect.provideService(Scope.Scope, terminalScope), + Effect.mapError((cause) => + terminalRequestError( + `Could not start terminal command '${request.command}'.`, + cause, ), ), - ).pipe( - Effect.onExit((exit) => - Exit.isFailure(exit) ? Scope.close(terminalScope, Exit.void) : Effect.void, ), - ); + ).pipe( + Effect.onExit((exit) => + Exit.isFailure(exit) ? Scope.close(terminalScope, Exit.void) : Effect.void, + ), + ); - const terminalId = `t3-term-${nextTerminalNumber}`; - nextTerminalNumber += 1; - const record: ManagedAcpTerminal = { - terminalId, - sessionId: request.sessionId, - commandLine: [request.command, ...(request.args ?? [])].join(" "), - buffer: { - chunks: [], - bytes: 0, - truncated: false, - limit: Math.min( - request.outputByteLimit ?? DEFAULT_OUTPUT_BYTE_LIMIT, - MAX_OUTPUT_BYTE_LIMIT, - ), - }, - exit: yield* Deferred.make(), - kill: child.kill().pipe(Effect.ignore), - dispose: Scope.close(terminalScope, Exit.void).pipe(Effect.ignore), - exitStatus: undefined, - released: false, - }; - terminals.set(terminalId, record); + const terminalId = `t3-term-${nextTerminalNumber}`; + nextTerminalNumber += 1; + const record: ManagedAcpTerminal = { + terminalId, + sessionId: request.sessionId, + commandLine: [request.command, ...(request.args ?? [])].join(" "), + buffer: { + chunks: [], + bytes: 0, + truncated: false, + limit: Math.min( + request.outputByteLimit ?? DEFAULT_OUTPUT_BYTE_LIMIT, + MAX_OUTPUT_BYTE_LIMIT, + ), + }, + exit: yield* Deferred.make(), + kill: child.kill().pipe(Effect.ignore), + dispose: Scope.close(terminalScope, Exit.void).pipe(Effect.ignore), + exitStatus: undefined, + released: false, + }; + terminals.set(terminalId, record); - const pump = (stream: typeof child.stdout) => - stream.pipe( - Stream.runForEach((chunk) => - Effect.sync(() => { - appendOutput(record.buffer, chunk); - enforceTotalOutputLimit(); - }), - ), - Effect.ignore, - ); - const stdoutPump = yield* Effect.forkDetach(pump(child.stdout)); - const stderrPump = yield* Effect.forkDetach(pump(child.stderr)); - yield* Effect.forkDetach( - child.exitCode.pipe( - Effect.match({ - onFailure: (error) => ({ exitCode: null, signal: terminalExitSignal(error) }), - onSuccess: (code) => ({ exitCode: Number(code), signal: null }), + const pump = (stream: typeof child.stdout) => + stream.pipe( + Stream.runForEach((chunk) => + Effect.sync(() => { + appendOutput(record.buffer, chunk); + enforceTotalOutputLimit(); }), - Effect.flatMap((status) => - Effect.all([Fiber.join(stdoutPump), Fiber.join(stderrPump)], { - discard: true, - }).pipe( - Effect.andThen( - Effect.sync(() => { - record.exitStatus = status; - }), - ), - Effect.andThen(Deferred.succeed(record.exit, status)), + ), + Effect.ignore, + ); + const stdoutPump = yield* Effect.forkDetach(pump(child.stdout)); + const stderrPump = yield* Effect.forkDetach(pump(child.stderr)); + yield* Effect.forkDetach( + child.exitCode.pipe( + Effect.match({ + onFailure: (error) => ({ exitCode: null, signal: terminalExitSignal(error) }), + onSuccess: (code) => ({ exitCode: Number(code), signal: null }), + }), + Effect.flatMap((status) => + Effect.all([Fiber.join(stdoutPump), Fiber.join(stderrPump)], { + discard: true, + }).pipe( + Effect.andThen( + Effect.sync(() => { + record.exitStatus = status; + }), ), + Effect.andThen(Deferred.succeed(record.exit, status)), ), ), - ); - return { terminalId }; - }), - ), - ); - - const output: AcpClientTerminals["output"] = (request) => - requireTerminal(request.sessionId, request.terminalId, "terminal/output").pipe( - Effect.map((terminal) => ({ - output: readOutput(terminal.buffer), - truncated: terminal.buffer.truncated, - ...(terminal.exitStatus === undefined ? {} : { exitStatus: terminal.exitStatus }), - })), - ); - - const waitForExit: AcpClientTerminals["waitForExit"] = (request) => - requireTerminal(request.sessionId, request.terminalId, "terminal/wait_for_exit").pipe( - Effect.flatMap((terminal) => Deferred.await(terminal.exit)), - ); - - const kill: AcpClientTerminals["kill"] = (request) => - requireTerminal(request.sessionId, request.terminalId, "terminal/kill").pipe( - Effect.flatMap((terminal) => terminal.kill), - Effect.as({}), - ); - - const release: AcpClientTerminals["release"] = (request) => - Effect.uninterruptible( - requireTerminal(request.sessionId, request.terminalId, "terminal/release").pipe( - Effect.flatMap((terminal) => - Effect.sync(() => { - terminal.released = true; - }).pipe( - Effect.andThen(terminal.dispose), - Effect.andThen(Deferred.await(terminal.exit)), - Effect.andThen(Effect.sync(pruneReleasedSnapshots)), ), - ), - Effect.as({}), - ), - ); + ); + return { terminalId }; + }), + ), + ); - const readOutputSnapshot: AcpClientTerminals["readOutputSnapshot"] = ( - sessionId, - terminalId, - ) => { - const terminal = terminals.get(terminalId); - if (terminal === undefined || terminal.sessionId !== sessionId) return undefined; - return { + const output: AcpClientTerminals["output"] = (request) => + requireTerminal(request.sessionId, request.terminalId, "terminal/output").pipe( + Effect.map((terminal) => ({ output: readOutput(terminal.buffer), truncated: terminal.buffer.truncated, - exitStatus: terminal.exitStatus, - }; - }; + ...(terminal.exitStatus === undefined ? {} : { exitStatus: terminal.exitStatus }), + })), + ); - const readCommandLine: AcpClientTerminals["readCommandLine"] = (terminalId) => - terminals.get(terminalId)?.commandLine; + const waitForExit: AcpClientTerminals["waitForExit"] = (request) => + requireTerminal(request.sessionId, request.terminalId, "terminal/wait_for_exit").pipe( + Effect.flatMap((terminal) => Deferred.await(terminal.exit)), + ); - const disposeAll: AcpClientTerminals["disposeAll"] = Effect.uninterruptible( - Effect.suspend(() => - Effect.forEach( - Array.from(terminals.values()), - (terminal) => - Effect.sync(() => { - terminal.released = true; - }).pipe( - Effect.andThen(terminal.dispose), - Effect.andThen(Deferred.await(terminal.exit)), - ), - { concurrency: "unbounded", discard: true }, - ).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))), - ), + const kill: AcpClientTerminals["kill"] = (request) => + requireTerminal(request.sessionId, request.terminalId, "terminal/kill").pipe( + Effect.flatMap((terminal) => terminal.kill), + Effect.as({}), ); - // Terminal failures surface to the agent as JSON-RPC errors and to the - // user only as a failed tool item, so log them for server-side diagnosis. - const logged = - ( - operation: string, - handler: (request: Request) => Effect.Effect, - ) => - (request: Request) => - handler(request).pipe( - Effect.tapError((error) => - Effect.logWarning("ACP client terminal operation failed", { - operation, - detail: error.message, - }), + const release: AcpClientTerminals["release"] = (request) => + Effect.uninterruptible( + requireTerminal(request.sessionId, request.terminalId, "terminal/release").pipe( + Effect.flatMap((terminal) => + Effect.sync(() => { + terminal.released = true; + }).pipe( + Effect.andThen(terminal.dispose), + Effect.andThen(Deferred.await(terminal.exit)), + Effect.andThen(Effect.sync(pruneReleasedSnapshots)), ), - ); + ), + Effect.as({}), + ), + ); + const readOutputSnapshot: AcpClientTerminals["readOutputSnapshot"] = (sessionId, terminalId) => { + const terminal = terminals.get(terminalId); + if (terminal === undefined || terminal.sessionId !== sessionId) return undefined; return { - create: logged("terminal/create", create), - output: logged("terminal/output", output), - waitForExit: logged("terminal/wait_for_exit", waitForExit), - kill: logged("terminal/kill", kill), - release: logged("terminal/release", release), - readOutputSnapshot, - readCommandLine, - disposeAll, + output: readOutput(terminal.buffer), + truncated: terminal.buffer.truncated, + exitStatus: terminal.exitStatus, }; - }); + }; + + const readCommandLine: AcpClientTerminals["readCommandLine"] = (terminalId) => + terminals.get(terminalId)?.commandLine; + + const disposeAll: AcpClientTerminals["disposeAll"] = Effect.uninterruptible( + Effect.suspend(() => + Effect.forEach( + Array.from(terminals.values()), + (terminal) => + Effect.sync(() => { + terminal.released = true; + }).pipe(Effect.andThen(terminal.dispose), Effect.andThen(Deferred.await(terminal.exit))), + { concurrency: "unbounded", discard: true }, + ).pipe(Effect.andThen(Effect.sync(pruneReleasedSnapshots))), + ), + ); + + // Terminal failures surface to the agent as JSON-RPC errors and to the + // user only as a failed tool item, so log them for server-side diagnosis. + const logged = + ( + operation: string, + handler: (request: Request) => Effect.Effect, + ) => + (request: Request) => + handler(request).pipe( + Effect.tapError((error) => + Effect.logWarning("ACP client terminal operation failed", { + operation, + detail: error.message, + }), + ), + ); + + return { + create: logged("terminal/create", create), + output: logged("terminal/output", output), + waitForExit: logged("terminal/wait_for_exit", waitForExit), + kill: logged("terminal/kill", kill), + release: logged("terminal/release", release), + readOutputSnapshot, + readCommandLine, + disposeAll, + }; +}); /** * Rewrites embedded `terminal` tool-call content into plain text content so diff --git a/packages/provider-grok/src/server/acpCliProbe.test.ts b/packages/provider-grok/src/server/acpCliProbe.test.ts index aa1a7704675825a4ed12565acd3445952bec78d9..eed0c1159342cb58521218e01cd3727bfc1c3733 100644 --- a/packages/provider-grok/src/server/acpCliProbe.test.ts +++ b/packages/provider-grok/src/server/acpCliProbe.test.ts @@ -15,20 +15,15 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Stream from "effect/Stream"; -import { ChildProcessSpawner } from "effect/process"; import { describe, expect } from "vite-plus/test"; import { makeGrokAcpRuntime } from "./acpSupport.ts"; -const makeProbeRuntime = Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - return yield* makeGrokAcpRuntime({ - grokSettings: { binaryPath: "grok" }, - environment: process.env, - childProcessSpawner, - cwd: process.cwd(), - clientInfo: { name: "t3-grok-probe", version: "0.0.0" }, - }); +const makeProbeRuntime = makeGrokAcpRuntime({ + grokSettings: { binaryPath: "grok" }, + environment: process.env, + cwd: process.cwd(), + clientInfo: { name: "t3-grok-probe", version: "0.0.0" }, }); describe.runIf(process.env.T3_GROK_ACP_PROBE === "1")("Grok ACP CLI probe", () => { @@ -78,11 +73,9 @@ describe.runIf(process.env.T3_GROK_ACP_PROBE === "1")("Grok ACP CLI probe", () = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const cwd = yield* fileSystem.makeTempDirectoryScoped(); - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const runtime = yield* makeGrokAcpRuntime({ grokSettings: { binaryPath: "grok" }, environment: process.env, - childProcessSpawner, cwd, runtimeMode: "approval-required", clientInfo: { name: "t3-grok-probe", version: "0.0.0" }, diff --git a/packages/provider-grok/src/server/acpSupport.ts b/packages/provider-grok/src/server/acpSupport.ts index e30b475219ef64934125b946e91332481ef1799a..53f5ae3ffb2be7779fffd8e8c743895a310af75c 100644 --- a/packages/provider-grok/src/server/acpSupport.ts +++ b/packages/provider-grok/src/server/acpSupport.ts @@ -31,7 +31,6 @@ interface GrokAcpRuntimeInput extends Omit< AcpSessionRuntime.AcpSessionRuntimeOptions, "authMethodId" | "clientCapabilities" | "spawn" > { - readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; readonly grokSettings: GrokAcpRuntimeGrokSettings | null | undefined; readonly environment?: NodeJS.ProcessEnv; readonly runtimeMode?: RuntimeMode; @@ -157,7 +156,7 @@ export const makeGrokAcpRuntime = ( ): Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, - Crypto.Crypto | Scope.Scope + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope > => Effect.gen(function* () { const processGroupPlatform = yield* HostProcessPlatform.pipe( @@ -176,11 +175,7 @@ export const makeGrokAcpRuntime = ( cancelMeta: { ...input.cancelMeta, ...GROK_ACP_CANCEL_META }, initializeMeta: GROK_ACP_INITIALIZE_META, ...grokAcpRuntimeProcessOwnership(processGroupPlatform), - }).pipe( - Layer.provide( - Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, input.childProcessSpawner), - ), - ), + }), ); const runtime = yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe( Effect.provide(acpContext), diff --git a/packages/provider-grok/src/server/adapter.ts b/packages/provider-grok/src/server/adapter.ts index 897fe4f0d2a4991f80951422b20f56d43f0f826f..8100246ace186d1e3718466493242ae0da2968fc 100644 --- a/packages/provider-grok/src/server/adapter.ts +++ b/packages/provider-grok/src/server/adapter.ts @@ -20,7 +20,7 @@ import * as Path from "effect/Path"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as EffectAcpErrors from "effect-acp/errors"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -78,6 +78,7 @@ export const GROK_PROVIDER = ProviderDriverKind.make("grok"); const GROK_DRIVER_KIND = GROK_PROVIDER; export const GROK_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(GROK_DRIVER_KIND); const DEFAULT_GROK_SETTINGS = Schema.decodeSync(GrokSettings)({}); +const isAcpRequestError = Schema.is(EffectAcpErrors.AcpRequestError); export const GrokProviderCapabilitiesV2 = { ...AcpProviderCapabilitiesV2, @@ -122,7 +123,7 @@ export interface GrokAdapterV2Options { ) => Effect.Effect< AcpSessionRuntime.AcpSessionRuntime["Service"], EffectAcpErrors.AcpError, - Crypto.Crypto | Scope.Scope + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope >; readonly assertComplete?: Effect.Effect; } @@ -226,11 +227,7 @@ export function grokLaunchRuntimeMode( } /** The flavor runs Grok's launcher through the adapter's spawner. */ -export function makeGrokAcpAdapterFlavor( - options: GrokAdapterV2Options & { - readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; - }, -): AcpAdapterV2Flavor { +export function makeGrokAcpAdapterFlavor(options: GrokAdapterV2Options): AcpAdapterV2Flavor { return { driver: GROK_PROVIDER, runtimeHarness: "Grok", @@ -281,7 +278,6 @@ export function makeGrokAcpAdapterFlavor( interruptPromptOnCancel: input.interruptPromptOnCancel ?? false, grokSettings: options.settings, environment: options.environment, - childProcessSpawner: options.childProcessSpawner, runtimeMode: grokLaunchRuntimeMode(runtimePolicy), })), // In its Auto mode Grok decides routine actions itself and only asks about @@ -292,7 +288,7 @@ export function makeGrokAcpAdapterFlavor( promptFailure: (cause) => makeProviderFailure({ cause, - ...(Schema.is(EffectAcpErrors.AcpRequestError)(cause) + ...(isAcpRequestError(cause) ? { // Grok's own failure text rides on the cause; makeProviderFailure // redacts and bounds it before it reaches the user. @@ -333,11 +329,9 @@ export function makeGrokAcpAdapterFlavor( export const makeGrokAdapterV2 = Effect.fn("makeGrokAdapterV2")(function* ( options: GrokAdapterV2Options, ) { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const flavor = makeGrokAcpAdapterFlavor({ ...options, childProcessSpawner }); return yield* makeAcpAdapterV2({ instanceId: options.instanceId, - flavor, + flavor: makeGrokAcpAdapterFlavor(options), selfInvocation: options.selfInvocation, ...(options.nativeLogging === undefined ? {} : { nativeLogging: options.nativeLogging }), ...(options.continuationRequests === undefined diff --git a/packages/provider-grok/src/server/driver.test.ts b/packages/provider-grok/src/server/driver.test.ts index 9c27b9a2b09232f9552ed15a3e8d64218e92ca03..22b2c64aa0b8ba0400e23f40314767ab6dfc30b1 100644 --- a/packages/provider-grok/src/server/driver.test.ts +++ b/packages/provider-grok/src/server/driver.test.ts @@ -6,7 +6,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; -import { HttpClient } from "effect/http"; +import * as HttpClient from "effect/http/HttpClient"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; diff --git a/packages/provider-grok/src/server/driver.ts b/packages/provider-grok/src/server/driver.ts index fba71962704edf3c60df9d66c8ab02b3b45e702b..9846ba40a90357882874ae02e7574d665dc008fb 100644 --- a/packages/provider-grok/src/server/driver.ts +++ b/packages/provider-grok/src/server/driver.ts @@ -5,8 +5,8 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { HttpClient } from "effect/http"; -import { ChildProcessSpawner } from "effect/process"; +import * as HttpClient from "effect/http/HttpClient"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; @@ -191,7 +191,7 @@ export const GrokDriver: ProviderDriver = { new ProviderDriverError({ driver: DRIVER_KIND, instanceId, - detail: `Failed to build Grok snapshot: ${cause.message ?? String(cause)}`, + detail: "Failed to build Grok snapshot.", cause, }), ), diff --git a/packages/provider-grok/src/server/skills.test.ts b/packages/provider-grok/src/server/skills.test.ts index e05d377e71a6d1d20344807237437aaa2b05e203..73e8ea0edd4667a4af9c74890484bf5eba8d37c8 100644 --- a/packages/provider-grok/src/server/skills.test.ts +++ b/packages/provider-grok/src/server/skills.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { discoverGrokSkills } from "./skills.ts"; diff --git a/packages/provider-grok/src/server/skills.ts b/packages/provider-grok/src/server/skills.ts index 1a65c794357caf09ff9ec48aa785d7a4db90eb9a..ba1b72482c7b9cdb82503fe2c5f899b1b3393b1e 100644 --- a/packages/provider-grok/src/server/skills.ts +++ b/packages/provider-grok/src/server/skills.ts @@ -20,7 +20,7 @@ import type { GrokSettings } from "../settings.ts"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import { ChildProcess } from "effect/process"; +import * as ChildProcess from "effect/process/ChildProcess"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; import { spawnAndCollect } from "@t3tools/provider-core/server/snapshotProbe"; diff --git a/packages/provider-grok/src/server/status.ts b/packages/provider-grok/src/server/status.ts index e9d7caab677aa6852e5909a58bf1a1cbb795f58a..f901e1577899dcb950ea9a1f8ff50f3482aa3305 100644 --- a/packages/provider-grok/src/server/status.ts +++ b/packages/provider-grok/src/server/status.ts @@ -16,8 +16,9 @@ import * as Exit from "effect/Exit"; import * as Option from "effect/Option"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; -import { HttpClient } from "effect/http"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import * as HttpClient from "effect/http/HttpClient"; +import * as ChildProcess from "effect/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { createModelCapabilities } from "@t3tools/shared/model"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; @@ -356,11 +357,9 @@ const discoverGrokMetadataViaAcpInitialize = ( environment: NodeJS.ProcessEnv, ) => Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const acp = yield* makeGrokAcpRuntime({ grokSettings, environment, - childProcessSpawner, cwd: process.cwd(), clientInfo: { name: "t3-code-provider-probe", version: "0.0.0" }, }); diff --git a/packages/provider-grok/src/server/textGeneration.ts b/packages/provider-grok/src/server/textGeneration.ts index b908c7e6e00ce15dac2fdfde7bbf6393fff29e14..df3e6380d7fac8b016f1ca9f484211c745fdd9cb 100644 --- a/packages/provider-grok/src/server/textGeneration.ts +++ b/packages/provider-grok/src/server/textGeneration.ts @@ -3,7 +3,7 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import { ChildProcessSpawner } from "effect/process"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import type * as EffectAcpErrors from "effect-acp/errors"; import { TextGenerationError } from "@t3tools/contracts"; @@ -37,10 +37,12 @@ export const makeGrokTextGeneration = Effect.fn("makeGrokTextGeneration")(functi const runtime = yield* makeGrokAcpRuntime({ grokSettings, environment, - childProcessSpawner: commandSpawner, cwd, clientInfo: { name: "t3-code-git-text", version: "0.0.0" }, - }).pipe(Effect.provideService(Crypto.Crypto, crypto)); + }).pipe( + Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, commandSpawner), + ); yield* runtime.handleSessionUpdate((notification) => { const update = notification.update; diff --git a/packages/provider-grok/src/server/usageLimits.ts b/packages/provider-grok/src/server/usageLimits.ts index f200956b56b17a016411f41dda58c59833183300..83369c062c16510b93cdcfbf7c66b855641a1ecd 100644 --- a/packages/provider-grok/src/server/usageLimits.ts +++ b/packages/provider-grok/src/server/usageLimits.ts @@ -6,7 +6,9 @@ import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; import { clampPercent, makeUnavailableUsageLimits,