diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 63b2b644b77d4..e237cbe4ed10d 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -937,21 +937,20 @@ const program = Effect.gen(function* () { }, }); }); - yield* Effect.sleep("1200 millis") - .pipe( - Effect.andThen( - Effect.sync(() => { - writeJsonRpcNotification("_x.ai/task_completed", { - sessionId: cancelledSessionId, - update: { - sessionUpdate: "task_completed", - task_snapshot: { task_id: "task-bg-1", command: "sleep 30" }, - }, - }); - }), - ), - ) - .pipe(Effect.forkDetach); + yield* Effect.sleep("1200 millis").pipe( + Effect.andThen( + Effect.sync(() => { + writeJsonRpcNotification("_x.ai/task_completed", { + sessionId: cancelledSessionId, + update: { + sessionUpdate: "task_completed", + task_snapshot: { task_id: "task-bg-1", command: "sleep 30" }, + }, + }); + }), + ), + Effect.forkDetach, + ); } }), ); diff --git a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts index fcd9f8bfef5cd..ca37361d244c1 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts @@ -220,10 +220,10 @@ const make = Effect.gen(function* () { if (threadId === undefined) { return yield* new PullRequestThreadRequiredError(); } - const thread = yield* engine - .getThreadShell(threadId) - .pipe(Effect.map(Option.fromNullishOr)) - .pipe(Effect.mapError((cause) => new Failure({ cause }))); + const thread = yield* engine.getThreadShell(threadId).pipe( + Effect.map(Option.fromNullishOr), + Effect.mapError((cause) => new Failure({ cause })), + ); if (Option.isNone(thread) || thread.value.deletedAt !== null) { return yield* new PullRequestThreadNotFoundError({ threadId }); } diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 5dfbffa595908..64891a4444d46 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1359,12 +1359,11 @@ describe("ClaudeAdapterV2 resume compaction", () => { assert.equal(planEvent.value.plan.markdown, longPlan.trim()); } } - const requestEvent = yield* runtime.events - .pipe( - Stream.filter((event) => event.type === "runtime_request.updated"), - Stream.runHead, - ) - .pipe(Effect.forkScoped); + const requestEvent = yield* runtime.events.pipe( + Stream.filter((event) => event.type === "runtime_request.updated"), + Stream.runHead, + Effect.forkScoped, + ); const controller = new AbortController(); const dialog = yield* Effect.promise(() => callback!( diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 18500427a103d..132c9ad668035 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -724,28 +724,27 @@ const make = Effect.gen(function* () { fields, filter, ) => - ensureProjectionTranscript(input.threadId) - .pipe(Effect.andThen(orchestrator.getThreadRecords(input.threadId, fields, filter))) - .pipe( - Effect.mapError( - (cause) => - new ThreadManagementProjectionLoadError({ - projectId: input.projectId, - threadId: input.threadId, - cause, - }), - ), - Effect.flatMap((projection) => - projection.thread.projectId === input.projectId && projection.thread.deletedAt === null - ? Effect.succeed(projection) - : Effect.fail( - new ThreadManagementThreadNotFoundError({ - projectId: input.projectId, - threadId: input.threadId, - }), - ), - ), - ); + ensureProjectionTranscript(input.threadId).pipe( + Effect.andThen(orchestrator.getThreadRecords(input.threadId, fields, filter)), + Effect.mapError( + (cause) => + new ThreadManagementProjectionLoadError({ + projectId: input.projectId, + threadId: input.threadId, + cause, + }), + ), + Effect.flatMap((projection) => + projection.thread.projectId === input.projectId && projection.thread.deletedAt === null + ? Effect.succeed(projection) + : Effect.fail( + new ThreadManagementThreadNotFoundError({ + projectId: input.projectId, + threadId: input.threadId, + }), + ), + ), + ); const listProjectThreads: ThreadManagementServiceShape["listProjectThreads"] = (input) => orchestrator.getShellSnapshot({ projectId: input.projectId, location: "active" }).pipe( diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 356eb1f5fd912..e804e809da9f6 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -623,30 +623,27 @@ export const make = Effect.gen(function* () { const scans = Effect.forEach( scanReaders, ({ driver, reader }) => - usageInstances(driver, settings) - .pipe( - Effect.flatMap((instances) => - reader.scan({ instances, settings, windowStartMs, retentionCutoffMs, awaitRefresh }), - ), - ) - .pipe( - Effect.flatMap((sources) => - Effect.forEach(sources, ({ volumeId, ...source }) => - Effect.map( - volumeId === undefined - ? Effect.promise(() => readDirectoryVolumeId(source.dir)) - : Effect.succeed(volumeId), - (resolved): ScannedDir => ({ - ...source, - files: source.files?.map((file) => ({ ...file, malformedRecords: 0 })) ?? null, - provider: reader.provider, - volumeId: resolved, - }), - ), + usageInstances(driver, settings).pipe( + Effect.flatMap((instances) => + reader.scan({ instances, settings, windowStartMs, retentionCutoffMs, awaitRefresh }), + ), + Effect.flatMap((sources) => + Effect.forEach(sources, ({ volumeId, ...source }) => + Effect.map( + volumeId === undefined + ? Effect.promise(() => readDirectoryVolumeId(source.dir)) + : Effect.succeed(volumeId), + (resolved): ScannedDir => ({ + ...source, + files: source.files?.map((file) => ({ ...file, malformedRecords: 0 })) ?? null, + provider: reader.provider, + volumeId: resolved, + }), ), ), - Effect.provideContext(readerContext), ), + Effect.provideContext(readerContext), + ), { concurrency: "unbounded" }, ); diff --git a/tsconfig.base.json b/tsconfig.base.json index 07c4224cf8d25..68c43f51318df 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -24,6 +24,7 @@ "namespaceImportPackages": ["@effect/platform-node", "effect"], "effectFn": ["span", "inferred-span", "suggested-span"], "diagnosticSeverity": { + "unnecessaryPipeChain": "error", "effectFnOpportunity": "suggestion", "anyUnknownInErrorContext": "error", "unsafeEffectTypeAssertion": "error",