Showing the changes to apps/server/src/textGeneration/OpenCodeTextGeneration.ts (1 of 42 files). Show all
Unified · Split
| @@ -1,9 +1,5 @@ | |||
| 1 | import * as Effect from "effect/Effect"; | 1 | import * as Effect from "effect/Effect"; |
| 2 | import * as Exit from "effect/Exit"; | ||
| 3 | import * as Fiber from "effect/Fiber"; | ||
| 4 | import * as Schema from "effect/Schema"; | 2 | import * as Schema from "effect/Schema"; |
| 5 | import * as Scope from "effect/Scope"; | ||
| 6 | import * as Semaphore from "effect/Semaphore"; | ||
| 7 | 3 | ||
| 8 | import { | 4 | import { |
| 9 | NonNegativeInt, | 5 | NonNegativeInt, |
| @@ -31,8 +27,7 @@ import { · 21 unmodified lines | |||
| 31 | sanitizeThreadTitle, | 27 | sanitizeThreadTitle, |
| 32 | } from "./TextGenerationUtils.ts"; | 28 | } from "./TextGenerationUtils.ts"; |
| 33 | import * as OpenCodeRuntime from "../provider/opencodeRuntime.ts"; | 29 | import * as OpenCodeRuntime from "../provider/opencodeRuntime.ts"; |
| 34 | 30 | import * as OpenCodeServerOwner from "../provider/OpenCodeServerOwner.ts"; | |
| 35 | const OPENCODE_TEXT_GENERATION_IDLE_TTL = "30 seconds"; | ||
| 36 | 31 | ||
| 37 | const OpenCodeTextGenerationOperation = Schema.Literals([ | 32 | const OpenCodeTextGenerationOperation = Schema.Literals([ |
| 38 | "generateCommitMessage", | 33 | "generateCommitMessage", |
| @@ -175,188 +170,12 @@ function getOpenCodeTextResponse(parts: ReadonlyArray<unknown> | undefined): str · 136 unmodified lines | |||
| 175 | .trim(); | 170 | .trim(); |
| 176 | } | 171 | } |
| 177 | 172 | ||
| 178 | interface SharedOpenCodeTextGenerationServerState { | ||
| 179 | server: OpenCodeRuntime.OpenCodeServerProcess | null; | ||
| 180 | /** | ||
| 181 | * The scope that owns the shared server's lifetime. Closing this scope | ||
| 182 | * terminates the OpenCode child process and interrupts any fibers the | ||
| 183 | * runtime forked during startup. We don't hold a `close()` function on | ||
| 184 | * the server handle anymore — the scope is the only lifecycle handle. | ||
| 185 | */ | ||
| 186 | serverScope: Scope.Closeable | null; | ||
| 187 | binaryPath: string | null; | ||
| 188 | activeRequests: number; | ||
| 189 | idleCloseFiber: Fiber.Fiber<void, never> | null; | ||
| 190 | } | ||
| 191 | |||
| 192 | export const makeOpenCodeTextGeneration = Effect.fn("makeOpenCodeTextGeneration")(function* ( | 173 | export const makeOpenCodeTextGeneration = Effect.fn("makeOpenCodeTextGeneration")(function* ( |
| 193 | openCodeSettings: OpenCodeSettings, | 174 | openCodeSettings: OpenCodeSettings, |
| 194 | environment?: NodeJS.ProcessEnv, | ||
| 195 | ) { | 175 | ) { |
| 196 | const serverConfig = yield* ServerConfig.ServerConfig; | 176 | const serverConfig = yield* ServerConfig.ServerConfig; |
| 197 | const openCodeRuntime = yield* OpenCodeRuntime.OpenCodeRuntime; | 177 | const openCodeRuntime = yield* OpenCodeRuntime.OpenCodeRuntime; |
| 198 | const resolvedEnvironment = environment ?? process.env; | 178 | const serverOwner = yield* OpenCodeServerOwner.OpenCodeServerOwner; |
| 199 | const idleFiberScope = yield* Effect.acquireRelease(Scope.make(), (scope) => | ||
| 200 | Scope.close(scope, Exit.void), | ||
| 201 | ); | ||
| 202 | const sharedServerMutex = yield* Semaphore.make(1); | ||
| 203 | const sharedServerState: SharedOpenCodeTextGenerationServerState = { | ||
| 204 | server: null, | ||
| 205 | serverScope: null, | ||
| 206 | binaryPath: null, | ||
| 207 | activeRequests: 0, | ||
| 208 | idleCloseFiber: null, | ||
| 209 | }; | ||
| 210 | |||
| 211 | const closeSharedServer = Effect.fn("closeSharedServer")(function* () { | ||
| 212 | const scope = sharedServerState.serverScope; | ||
| 213 | sharedServerState.server = null; | ||
| 214 | sharedServerState.serverScope = null; | ||
| 215 | sharedServerState.binaryPath = null; | ||
| 216 | if (scope !== null) { | ||
| 217 | yield* Scope.close(scope, Exit.void).pipe(Effect.ignore); | ||
| 218 | } | ||
| 219 | }); | ||
| 220 | |||
| 221 | const cancelIdleCloseFiber = Effect.fn("cancelIdleCloseFiber")(function* () { | ||
| 222 | const idleCloseFiber = sharedServerState.idleCloseFiber; | ||
| 223 | sharedServerState.idleCloseFiber = null; | ||
| 224 | if (idleCloseFiber !== null) { | ||
| 225 | yield* Fiber.interrupt(idleCloseFiber).pipe(Effect.ignore); | ||
| 226 | } | ||
| 227 | }); | ||
| 228 | |||
| 229 | const scheduleIdleClose = Effect.fn("scheduleIdleClose")(function* ( | ||
| 230 | server: OpenCodeRuntime.OpenCodeServerProcess, | ||
| 231 | ) { | ||
| 232 | yield* cancelIdleCloseFiber(); | ||
| 233 | const fiber = yield* Effect.sleep(OPENCODE_TEXT_GENERATION_IDLE_TTL).pipe( | ||
| 234 | Effect.andThen( | ||
| 235 | sharedServerMutex.withPermit( | ||
| 236 | Effect.gen(function* () { | ||
| 237 | if (sharedServerState.server !== server || sharedServerState.activeRequests > 0) { | ||
| 238 | return; | ||
| 239 | } | ||
| 240 | sharedServerState.idleCloseFiber = null; | ||
| 241 | yield* closeSharedServer(); | ||
| 242 | }), | ||
| 243 | ), | ||
| 244 | ), | ||
| 245 | Effect.forkIn(idleFiberScope), | ||
| 246 | ); | ||
| 247 | sharedServerState.idleCloseFiber = fiber; | ||
| 248 | }); | ||
| 249 | |||
| 250 | const acquireSharedServer = (input: { | ||
| 251 | readonly binaryPath: string; | ||
| 252 | readonly operation: | ||
| 253 | | "generateCommitMessage" | ||
| 254 | | "generatePrContent" | ||
| 255 | | "generateBranchName" | ||
| 256 | | "generateThreadTitle"; | ||
| 257 | }) => | ||
| 258 | sharedServerMutex.withPermit( | ||
| 259 | Effect.gen(function* () { | ||
| 260 | yield* cancelIdleCloseFiber(); | ||
| 261 | |||
| 262 | const existingServer = sharedServerState.server; | ||
| 263 | if (existingServer !== null) { | ||
| 264 | if ( | ||
| 265 | sharedServerState.binaryPath !== input.binaryPath && | ||
| 266 | sharedServerState.activeRequests === 0 | ||
| 267 | ) { | ||
| 268 | yield* closeSharedServer(); | ||
| 269 | } else { | ||
| 270 | if (sharedServerState.binaryPath !== input.binaryPath) { | ||
| 271 | yield* Effect.logWarning( | ||
| 272 | "OpenCode shared server binary path mismatch: requested " + | ||
| 273 | input.binaryPath + | ||
| 274 | " but active server uses " + | ||
| 275 | sharedServerState.binaryPath + | ||
| 276 | "; reusing existing server because there are active requests", | ||
| 277 | ); | ||
| 278 | } | ||
| 279 | sharedServerState.activeRequests += 1; | ||
| 280 | return existingServer; | ||
| 281 | } | ||
| 282 | } | ||
| 283 | |||
| 284 | // Create a fresh scope that owns this shared server. The runtime | ||
| 285 | // will attach its child-process and fiber finalizers to this scope; | ||
| 286 | // closing it kills the server and interrupts those fibers. | ||
| 287 | // | ||
| 288 | // The `Scope.make` / spawn / record-or-close transitions run inside | ||
| 289 | // `uninterruptibleMask` so an interrupt arriving between any two | ||
| 290 | // steps can't orphan the scope (and the child process attached to | ||
| 291 | // it) before we either close it on failure or hand ownership to | ||
| 292 | // `sharedServerState`. `restore` keeps the actual spawn | ||
| 293 | // interruptible; an interrupt during the spawn is captured by | ||
| 294 | // `Effect.exit` and drives us through the failure branch that | ||
| 295 | // closes the fresh scope. | ||
| 296 | return yield* Effect.uninterruptibleMask((restore) => | ||
| 297 | Effect.gen(function* () { | ||
| 298 | const serverScope = yield* Scope.make(); | ||
| 299 | const startedExit = yield* Effect.exit( | ||
| 300 | restore( | ||
| 301 | openCodeRuntime | ||
| 302 | .startOpenCodeServerProcess({ | ||
| 303 | binaryPath: input.binaryPath, | ||
| 304 | environment: resolvedEnvironment, | ||
| 305 | }) | ||
| 306 | .pipe( | ||
| 307 | Effect.provideService(Scope.Scope, serverScope), | ||
| 308 | Effect.mapError( | ||
| 309 | (cause) => | ||
| 310 | new TextGenerationError({ | ||
| 311 | operation: input.operation, | ||
| 312 | detail: OpenCodeRuntime.openCodeRuntimeErrorDetail(cause), | ||
| 313 | cause, | ||
| 314 | }), | ||
| 315 | ), | ||
| 316 | ), | ||
| 317 | ), | ||
| 318 | ); | ||
| 319 | if (startedExit._tag === "Failure") { | ||
| 320 | yield* Scope.close(serverScope, Exit.void).pipe(Effect.ignore); | ||
| 321 | return yield* Effect.failCause(startedExit.cause); | ||
| 322 | } | ||
| 323 | |||
| 324 | const server = startedExit.value; | ||
| 325 | sharedServerState.server = server; | ||
| 326 | sharedServerState.serverScope = serverScope; | ||
| 327 | sharedServerState.binaryPath = input.binaryPath; | ||
| 328 | sharedServerState.activeRequests = 1; | ||
| 329 | return server; | ||
| 330 | }), | ||
| 331 | ); | ||
| 332 | }), | ||
| 333 | ); | ||
| 334 | |||
| 335 | const releaseSharedServer = (server: OpenCodeRuntime.OpenCodeServerProcess) => | ||
| 336 | sharedServerMutex.withPermit( | ||
| 337 | Effect.gen(function* () { | ||
| 338 | if (sharedServerState.server !== server) { | ||
| 339 | return; | ||
| 340 | } | ||
| 341 | sharedServerState.activeRequests = Math.max(0, sharedServerState.activeRequests - 1); | ||
| 342 | if (sharedServerState.activeRequests === 0) { | ||
| 343 | yield* scheduleIdleClose(server); | ||
| 344 | } | ||
| 345 | }), | ||
| 346 | ); | ||
| 347 | |||
| 348 | // Module-level finalizer: on layer shutdown, cancel the idle close fiber | ||
| 349 | // and close the shared server scope. Consumers therefore cannot leak | ||
| 350 | // the shared OpenCode server by forgetting to call anything. | ||
| 351 | yield* Effect.addFinalizer(() => | ||
| 352 | sharedServerMutex.withPermit( | ||
| 353 | Effect.gen(function* () { | ||
| 354 | yield* cancelIdleCloseFiber(); | ||
| 355 | sharedServerState.activeRequests = 0; | ||
| 356 | yield* closeSharedServer(); | ||
| 357 | }), | ||
| 358 | ), | ||
| 359 | ); | ||
| 360 | 179 | ||
| 361 | const runOpenCodeJson = Effect.fn("runOpenCodeJson")(function* <S extends Schema.Top>(input: { | 180 | const runOpenCodeJson = Effect.fn("runOpenCodeJson")(function* <S extends Schema.Top>(input: { |
| 362 | readonly operation: OpenCodeTextGenerationOperation; | 181 | readonly operation: OpenCodeTextGenerationOperation; |
| @@ -381,13 +200,16 @@ export const makeOpenCodeTextGeneration = Effect.fn("makeOpenCodeTextGeneration" · 18 unmodified lines | |||
| 381 | }); | 200 | }); |
| 382 | 201 | ||
| 383 | const runAgainstServer = Effect.fn("runOpenCodeJson.runAgainstServer")( | 202 | const runAgainstServer = Effect.fn("runOpenCodeJson.runAgainstServer")( |
| 384 | function* (server: Pick<OpenCodeRuntime.OpenCodeServerConnection, "url">) { | 203 | function* ( |
| 204 | server: Pick< | ||
| 205 | OpenCodeRuntime.OpenCodeServerConnection, | ||
| 206 | "url" | "serverPassword" | "version" | ||
| 207 | >, | ||
| 208 | ) { | ||
| 385 | const client = openCodeRuntime.createOpenCodeSdkClient({ | 209 | const client = openCodeRuntime.createOpenCodeSdkClient({ |
| 386 | baseUrl: server.url, | 210 | baseUrl: server.url, |
| 387 | directory: input.cwd, | 211 | directory: input.cwd, |
| 388 | ...(openCodeSettings.serverUrl.length > 0 && openCodeSettings.serverPassword | 212 | ...(server.serverPassword !== undefined ? { serverPassword: server.serverPassword } : {}), |
| 389 | ? { serverPassword: openCodeSettings.serverPassword } | ||
| 390 | : {}), | ||
| 391 | }); | 213 | }); |
| 392 | const session = yield* Effect.tryPromise({ | 214 | const session = yield* Effect.tryPromise({ |
| 393 | try: () => | 215 | try: () => |
| @@ -496,17 +318,31 @@ export const makeOpenCodeTextGeneration = Effect.fn("makeOpenCodeTextGeneration" · 102 unmodified lines | |||
| 496 | }), | 318 | }), |
| 497 | ); | 319 | ); |
| 498 | 320 | ||
| 499 | const rawOutput = | 321 | const serverOutput = |
| 500 | openCodeSettings.serverUrl.length > 0 | 322 | openCodeSettings.serverUrl.length > 0 |
| 501 | ? yield* runAgainstServer({ url: openCodeSettings.serverUrl }) | 323 | ? openCodeRuntime |
| 502 | : yield* Effect.acquireUseRelease( | 324 | .connectToOpenCodeServer({ |
| 503 | acquireSharedServer({ | ||
| 504 | binaryPath: openCodeSettings.binaryPath, | 325 | binaryPath: openCodeSettings.binaryPath, |
| 326 | directory: input.cwd, | ||
| 327 | serverUrl: openCodeSettings.serverUrl, | ||
| 328 | ...(openCodeSettings.serverPassword | ||
| 329 | ? { serverPassword: openCodeSettings.serverPassword } | ||
| 330 | : {}), | ||
| 331 | }) | ||
| 332 | .pipe(Effect.flatMap(runAgainstServer), Effect.scoped) | ||
| 333 | : serverOwner.withServer(runAgainstServer); | ||
| 334 | const rawOutput = yield* serverOutput.pipe( | ||
| 335 | Effect.catchTags({ | ||
| 336 | OpenCodeRuntimeError: (cause) => | ||
| 337 | Effect.fail( | ||
| 338 | new TextGenerationError({ | ||
| 505 | operation: input.operation, | 339 | operation: input.operation, |
| 340 | detail: OpenCodeRuntime.openCodeRuntimeErrorDetail(cause), | ||
| 341 | cause, | ||
| 506 | }), | 342 | }), |
| 507 | runAgainstServer, | 343 | ), |
| 508 | releaseSharedServer, | 344 | }), |
| 509 | ); | 345 | ); |
| 510 | 346 | ||
| 511 | const decodeOutput = Schema.decodeEffect(Schema.fromJsonString(input.outputSchemaJson)); | 347 | const decodeOutput = Schema.decodeEffect(Schema.fromJsonString(input.outputSchemaJson)); |
| 512 | return yield* decodeOutput(extractJsonObject(rawOutput)).pipe( | 348 | return yield* decodeOutput(extractJsonObject(rawOutput)).pipe( |