openAgent

1// Regression coverage for issue #26526's claim that promptAsync's
2// Effect.forkIn loses the request's InstanceRef/WorkspaceRef. It does not —
3// forkIn preserves Context.Reference values via standard fiber inheritance.
4//
5// The companion claim that the streaming prompt handler "captures and
6// provides" those services is true and load-bearing: Stream.fromEffect's
7// body runs detached from the request fiber's context, so the explicit
8// Effect.provideService calls there are required, not defensive duplication.
9 
10import { NodeHttpServer, NodeServices } from "@effect/platform-node"
11import { describe, expect } from "bun:test"
12import { Deferred, Effect, Layer, Schema, Scope } from "effect"
13import * as Stream from "effect/Stream"
14import { HttpClient, HttpRouter, HttpServerResponse } from "effect/unstable/http"
15import * as Socket from "effect/unstable/socket/Socket"
16import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/unstable/httpapi"
17import { mkdir } from "node:fs/promises"
18import { registerAdapter } from "../../src/control-plane/adapters"
19import type { WorkspaceAdapter } from "../../src/control-plane/types"
20import { Workspace } from "../../src/control-plane/workspace"
21import { InstanceRef, WorkspaceRef } from "../../src/effect/instance-ref"
22import { Project } from "../../src/project/project"
23import { Session } from "../../src/session/session"
24import {
25  InstanceContextMiddleware,
26  instanceContextLayer,
27} from "../../src/server/routes/instance/httpapi/middleware/instance-context"
28import {
29  WorkspaceRoutingMiddleware,
30  WorkspaceRoutingQuery,
31  workspaceRoutingLayer,
32} from "../../src/server/routes/instance/httpapi/middleware/workspace-routing"
33import { resetDatabase } from "../fixture/db"
34import { disposeAllInstances, tmpdirScoped } from "../fixture/fixture"
35import { workspaceLayerWithRuntimeFlags } from "../fixture/workspace"
36import { testEffect } from "../lib/effect"
37 
38const testStateLayer = Layer.effectDiscard(
39  Effect.gen(function* () {
40    yield* Effect.promise(() => resetDatabase())
41    yield* Effect.addFinalizer(() =>
42      Effect.promise(async () => {
43        await disposeAllInstances()
44        await resetDatabase()
45      }),
46    )
47  }),
48)
49 
50const workspaceLayer = workspaceLayerWithRuntimeFlags({ experimentalWorkspaces: true })
51 
52const it = testEffect(Layer.mergeAll(testStateLayer, NodeHttpServer.layerTest, NodeServices.layer, workspaceLayer))
53 
54const instanceContextTestLayer = Layer.mergeAll(
55  instanceContextLayer,
56  workspaceRoutingLayer.pipe(Layer.provide(Socket.layerWebSocketConstructorGlobal)),
57)
58 
59const localAdapter = (directory: string): WorkspaceAdapter => ({
60  name: "Local Test",
61  description: "Create a local test workspace",
62  configure: (info) => ({ ...info, name: "local-test", directory }),
63  create: async () => {
64    await mkdir(directory, { recursive: true })
65  },
66  async remove() {},
67  target: () => ({ type: "local" as const, directory }),
68})
69 
70const setupWorkspace = (kind: string) =>
71  Effect.gen(function* () {
72    const dir = yield* tmpdirScoped({ git: true })
73    yield* Project.use.fromDirectory(dir)
74    const projectID = yield* Project.Service.use((svc) => svc.fromDirectory(dir).pipe(Effect.map((p) => p.project.id)))
75    registerAdapter(projectID, kind, localAdapter(dir))
76    const workspace = yield* Workspace.Service.use((svc) =>
77      svc.create({ type: kind, branch: null, extra: null, projectID }),
78    )
79    return { dir, workspace }
80  })
81 
82type Capture = { directory?: string; workspaceID?: string }
83 
84const captureInstance = Effect.gen(function* () {
85  const instance = yield* InstanceRef
86  const workspaceID = yield* WorkspaceRef
87  return { directory: instance?.directory, workspaceID } satisfies Capture
88})
89 
90const ProbeApi = HttpApi.make("handler-context-probe").add(
91  HttpApiGroup.make("probe")
92    .add(
93      HttpApiEndpoint.post("fork", "/fork-probe", { query: WorkspaceRoutingQuery, success: Schema.Boolean }),
94      HttpApiEndpoint.post("streamWithout", "/stream-probe-without", {
95        query: WorkspaceRoutingQuery,
96        success: Schema.String.pipe(HttpApiSchema.asText({ contentType: "application/json" })),
97      }),
98      HttpApiEndpoint.post("streamWith", "/stream-probe-with", {
99        query: WorkspaceRoutingQuery,
100        success: Schema.String.pipe(HttpApiSchema.asText({ contentType: "application/json" })),
101      }),
102    )
103    .middleware(InstanceContextMiddleware)
104    .middleware(WorkspaceRoutingMiddleware),
105)
106 
107const serveProbes = (input: {
108  fork?: Effect.Effect<boolean, never, Scope.Scope>
109  streamWithout?: Effect.Effect<HttpServerResponse.HttpServerResponse>
110  streamWith?: Effect.Effect<HttpServerResponse.HttpServerResponse>
111}) =>
112  HttpApiBuilder.layer(ProbeApi).pipe(
113    Layer.provide(
114      HttpApiBuilder.group(ProbeApi, "probe", (handlers) =>
115        handlers
116          .handle("fork", () => input.fork ?? Effect.succeed(false))
117          .handleRaw(
118            "streamWithout",
119            () => input.streamWithout ?? Effect.succeed(HttpServerResponse.empty({ status: 404 })),
120          )
121          .handleRaw("streamWith", () => input.streamWith ?? Effect.succeed(HttpServerResponse.empty({ status: 404 }))),
122      ),
123    ),
124    Layer.provide(instanceContextTestLayer),
125    Layer.provide(Layer.mock(Session.Service)({})),
126    HttpRouter.serve,
127    Layer.build,
128  )
129 
130describe("HttpApi handler context inheritance", () => {
131  // Mirrors handlers/session.ts:281 promptAsync. The forked fiber inherits
132  // the request's Context — including InstanceRef and WorkspaceRef provided
133  // by InstanceContextMiddleware — without any explicit re-provide.
134  it.live("Effect.forkIn preserves InstanceRef/WorkspaceRef across the fork", () =>
135    Effect.gen(function* () {
136      const { dir, workspace } = yield* setupWorkspace("local-fork")
137      const capture = yield* Deferred.make<Capture>()
138 
139      yield* serveProbes({
140        fork: Effect.gen(function* () {
141          const scope = yield* Scope.Scope
142          yield* Effect.gen(function* () {
143            yield* Deferred.succeed(capture, yield* captureInstance)
144          }).pipe(Effect.forkIn(scope, { startImmediately: true }))
145          return true
146        }),
147      })
148 
149      const response = yield* HttpClient.post(
150        `/fork-probe?directory=${encodeURIComponent(dir)}&workspace=${encodeURIComponent(workspace.id)}`,
151      )
152      expect(response.status).toBe(200)
153 
154      const observed = yield* Deferred.await(capture).pipe(Effect.timeout("2 seconds"))
155      expect(observed.directory).toBe(dir)
156      expect(observed.workspaceID).toBe(workspace.id)
157    }),
158  )
159 
160  // Mirrors handlers/session.ts:255 prompt — the streaming handler reads
161  // InstanceRef/WorkspaceRef in the request fiber and re-provides them to
162  // the Stream.fromEffect body. This test locks in why the explicit
163  // provides are required: without them the stream body sees undefined.
164  it.live("Stream.fromEffect body needs explicit provides — inheritance does not carry through", () =>
165    Effect.gen(function* () {
166      const { dir, workspace } = yield* setupWorkspace("local-stream")
167      const withoutCapture = yield* Deferred.make<Capture>()
168      const withCapture = yield* Deferred.make<Capture>()
169 
170      yield* serveProbes({
171        streamWithout: Effect.gen(function* () {
172          return HttpServerResponse.stream(
173            Stream.fromEffect(
174              Effect.gen(function* () {
175                yield* Deferred.succeed(withoutCapture, yield* captureInstance)
176                return ""
177              }),
178            ).pipe(Stream.encodeText),
179            { contentType: "application/json" },
180          )
181        }),
182        streamWith: Effect.gen(function* () {
183          const instance = yield* InstanceRef
184          const workspaceID = yield* WorkspaceRef
185          return HttpServerResponse.stream(
186            Stream.fromEffect(
187              Effect.gen(function* () {
188                yield* Deferred.succeed(withCapture, yield* captureInstance)
189                return ""
190              }).pipe(Effect.provideService(InstanceRef, instance), Effect.provideService(WorkspaceRef, workspaceID)),
191            ).pipe(Stream.encodeText),
192            { contentType: "application/json" },
193          )
194        }),
195      })
196 
197      const queryString = `directory=${encodeURIComponent(dir)}&workspace=${encodeURIComponent(workspace.id)}`
198      const responseWithout = yield* HttpClient.post(`/stream-probe-without?${queryString}`)
199      yield* responseWithout.text
200      const responseWith = yield* HttpClient.post(`/stream-probe-with?${queryString}`)
201      yield* responseWith.text
202 
203      const without = yield* Deferred.await(withoutCapture).pipe(Effect.timeout("2 seconds"))
204      expect(without.directory).toBeUndefined()
205      expect(without.workspaceID).toBeUndefined()
206 
207      const withProvide = yield* Deferred.await(withCapture).pipe(Effect.timeout("2 seconds"))
208      expect(withProvide.directory).toBe(dir)
209      expect(withProvide.workspaceID).toBe(workspace.id)
210    }),
211  )
212})
213