Use CDP (if available) to inject instrumentation to avoid conflicts (#63)

Author: mattiamanzati

.changeset/sharp-sides-relax.md

@@ -0,0 +1,5 @@
+---
+"effect-vscode": minor
+---
+
+Use CDP (if available) to inject instrumentation to avoid scope conflicts

src/DebugChannel.ts

@@ -1,13 +1,20 @@
+import * as ChannelSchema from "@effect/platform/ChannelSchema"
+import * as Socket from "@effect/platform/Socket"
 import * as Array from "effect/Array"
 import * as Data from "effect/Data"
 import * as Effect from "effect/Effect"
+import * as Function from "effect/Function"
+import * as Mailbox from "effect/Mailbox"
 import * as Option from "effect/Option"
 import type * as ParseResult from "effect/ParseResult"
+import * as Predicate from "effect/Predicate"
 import * as Schema from "effect/Schema"
 import * as SchemaAST from "effect/SchemaAST"
+import * as Stream from "effect/Stream"
+import * as ws from "ws"
 import * as VsCode from "./VsCode"
 
-export class DebugChannelError extends Data.TaggedError("VariableExtractError")<{
+export class DebugChannelError extends Data.TaggedError("DebugChannelError")<{
   readonly message: string
 }> {}
 
@@ -31,8 +38,44 @@ export class DebugChannel extends Effect.Tag("effect-vscode/DebugChannel")<Debug
       threadId: number | undefined
     }
   ) => Effect.Effect<VariableReference, DebugChannelError, never>
+  evaluateOnEveryExecutionContext: (
+    opts: {
+      expression: string
+    }
+  ) => Effect.Effect<void, DebugChannelError, never>
 }>() {}
 
+export interface CdpProxyClient {
+  readonly queue: Mailbox.ReadonlyMailbox<unknown>
+  readonly request: (_: unknown) => Effect.Effect<void>
+}
+
+export const run = Effect.fnUntraced(
+  function*<R, E, _>(socket: Socket.Socket, handle: (client: CdpProxyClient) => Effect.Effect<_, E, R>) {
+    const responses = yield* Mailbox.make<unknown>()
+    const requests = yield* Mailbox.make<unknown>()
+
+    const client: CdpProxyClient = {
+      queue: requests,
+      request: (res) => responses.offer(res)
+    }
+
+    yield* Mailbox.toStream(responses).pipe(
+      Stream.pipeThroughChannel(
+        ChannelSchema.duplexUnknown(Socket.toChannelString(socket), {
+          inputSchema: Schema.parseJson(Schema.Unknown),
+          outputSchema: Schema.parseJson(Schema.Unknown)
+        })
+      ),
+      Stream.runForEach((req) => requests.offer(req)),
+      Effect.ensuring(Effect.zipRight(responses.shutdown, requests.shutdown)),
+      Effect.forkScoped
+    )
+
+    yield* handle(client)
+  }
+)
+
 interface DapVariableReference {
   readonly name: string
   readonly value: string
@@ -142,7 +185,12 @@ export const makeVsCodeDebugSession = (debugSession: VsCode.VsCodeDebugSession["
                 })
               }
               const elementAst = index < ast.elements.length ? ast.elements[index] : ast.rest[0]
-              return extractValue(indexProperty, elementAst.type)
+              return extractValue(indexProperty, elementAst.type).pipe(
+                Effect.catchTag(
+                  "DebugChannelError",
+                  (e) => new DebugChannelError({ message: "at index " + index + " " + e.message })
+                )
+              )
             })
             return yield* Effect.all(elements, { concurrency: "unbounded" })
           }
@@ -173,6 +221,14 @@ export const makeVsCodeDebugSession = (debugSession: VsCode.VsCodeDebugSession["
               result[propertySignature.name] = extractValue(
                 propertyVariableReference,
                 propertySignature.type
+              ).pipe(
+                Effect.catchTag(
+                  "DebugChannelError",
+                  (e) =>
+                    new DebugChannelError({
+                      message: "in property " + String(propertySignature.name) + " " + e.message
+                    })
+                )
               )
             }
             return yield* Effect.all(result, { concurrency: "unbounded" })
@@ -207,6 +263,66 @@ export const makeVsCodeDebugSession = (debugSession: VsCode.VsCodeDebugSession["
     }
 
     return DebugChannel.of({
+      evaluateOnEveryExecutionContext: (_opts) =>
+        Effect.gen(function*() {
+          const addr = yield* VsCode.executeCommand<{ host: string; port: number; path?: string }>(
+            "extension.js-debug.requestCDPProxy",
+            debugSession.id
+          )
+          const uri = `ws://${addr.host}:${addr.port}${addr.path || ""}`
+
+          const cdpSocket = yield* Socket.fromWebSocket(
+            Effect.sync(() => {
+              const wss = new ws.WebSocket(uri, {
+                perMessageDeflate: false,
+                maxPayload: 256 * 1024 * 1024
+              })
+
+              return wss as any
+            })
+          )
+
+          yield* run(
+            cdpSocket,
+            Effect.fn(function*(client) {
+              // we enable reporting of execution contexts
+              yield* client.request({
+                method: "Runtime.enable"
+              })
+
+              while (true) {
+                const response = yield* client.queue.take
+                if (
+                  Predicate.hasProperty(response, "method") && response.method === "Runtime.executionContextCreated" &&
+                  Predicate.hasProperty(response, "params") && Predicate.hasProperty(response.params, "context") &&
+                  Predicate.hasProperty(response.params.context, "id")
+                ) {
+                  const contextId = response.params.context.id
+
+                  // we send as notification and silent because we don't want to pollute the output
+                  yield* client.request({
+                    method: "Runtime.evaluate",
+                    params: {
+                      expression: _opts.expression,
+                      contextId,
+                      silent: true
+                    }
+                  })
+                }
+              }
+            })
+          )
+        }).pipe(
+          Effect.scoped,
+          Effect.timeoutTo({
+            onTimeout: Function.constUndefined,
+            onSuccess: Function.identity,
+            duration: 1000
+          }),
+          Effect.catchAll((socketError) => {
+            return new DebugChannelError({ message: String(socketError) })
+          })
+        ),
       evaluate: (opts) =>
         Effect.gen(function*() {
           let request: any = {
@@ -224,7 +340,7 @@ export const makeVsCodeDebugSession = (debugSession: VsCode.VsCodeDebugSession["
               const stackTraces = yield* debugRequest<DapStackTracesResponse>("stackTrace", {
                 threadId
               })
-              const stackTrace = stackTraces.stackFrames[stackTraces.stackFrames.length - 1]
+              const stackTrace = stackTraces.stackFrames[0]
               if (stackTrace) {
                 request = {
                   expression: opts.expression,

src/DebugEnv.ts

@@ -99,16 +99,29 @@ export class DebugEnv extends Context.Tag("effect-vscode/DebugEnv")<
 
 // --
 
-export const ensureInstrumentationInjected = (guessFrameId: boolean, threadId?: number) =>
+export const ensureInstrumentationInjected = (
+  guessFrameId: boolean,
+  threadId?: number
+) =>
   Effect.gen(function*() {
     const result = yield* DebugChannel.DebugChannel.evaluate({
-      expression: `globalThis && "effect/devtools/instrumentation" in globalThis`,
+      expression: `(globalThis && "effect/devtools/instrumentation" in globalThis)`,
       guessFrameId,
       threadId
     })
     const isInjected = yield* result.parse(Schema.Boolean)
     if (!isInjected) {
-      yield* DebugChannel.DebugChannel.evaluate({ expression: compiledInstrumentationString, guessFrameId, threadId })
+      yield* DebugChannel.DebugChannel.evaluateOnEveryExecutionContext({
+        expression: compiledInstrumentationString
+      }).pipe(
+        Effect.orElse(() =>
+          DebugChannel.DebugChannel.evaluate({
+            expression: compiledInstrumentationString,
+            guessFrameId,
+            threadId
+          })
+        )
+      )
     }
   })
 
@@ -169,75 +182,83 @@ const SpanSchema = Schema.Struct({
 })
 
 const ExternalSpanSchema = Schema.Struct({
-  _tag: Schema.Literal("External"),
+  _tag: Schema.Literal("ExternalSpan"),
   spanId: Schema.String,
   traceId: Schema.String
 })
 
-const SpanStackSchema = Schema.Array(Schema.Union(SpanSchema, ExternalSpanSchema))
+const AnySpanSchema = Schema.Union(SpanSchema, ExternalSpanSchema)
+export type AnySpanSchema = Schema.Schema.Type<typeof AnySpanSchema>
+
+function spanEntryToSpanStackEntry(entry: AnySpanSchema | undefined): Array<SpanStackEntry> {
+  const spans: Array<SpanStackEntry> = []
+  if (!entry) return []
+  switch (entry._tag) {
+    case "Span": {
+      let match = false
+      for (let stackIndex = 0; stackIndex < entry.stack.length; stackIndex++) {
+        const stackLine = entry.stack[stackIndex]!
+        match = true
+        spans.push(
+          new SpanStackEntry({
+            ...stackLine,
+            ...entry,
+            stackIndex
+          })
+        )
+      }
+
+      if (!match) {
+        spans.push(new SpanStackEntry({ ...entry, stackIndex: -1, line: 0, column: 0 }))
+      }
+      break
+    }
+    case "ExternalSpan": {
+      spans.push(
+        new SpanStackEntry({
+          ...entry,
+          name: "<external span " + entry.spanId + ">",
+          stackIndex: -1,
+          line: 0,
+          column: 0,
+          attributes: []
+        })
+      )
+      break
+    }
+  }
+  return spans
+}
+
+const FiberCurrentSpanResponseSchema = Schema.Array(Schema.Union(SpanSchema, ExternalSpanSchema))
 
-function getFiberCurrentSpan(currentFiberExpression: string, threadId: number | undefined) {
+function getFiberCurrentSpan(currentFiberExpression: string, maxDepth: number, threadId: number | undefined) {
   return Effect.gen(function*() {
     yield* ensureInstrumentationInjected(true, threadId)
     const result = yield* DebugChannel.DebugChannel.evaluate({
-      expression: `globalThis["effect/devtools/instrumentation"].getFiberCurrentSpanStack(${currentFiberExpression})`,
+      expression:
+        `globalThis["effect/devtools/instrumentation"].getFiberCurrentSpanStack(${currentFiberExpression}, ${maxDepth})`,
       guessFrameId: true,
       threadId
     })
-    return yield* result.parse(SpanStackSchema)
+    return yield* result.parse(FiberCurrentSpanResponseSchema)
   }).pipe(
     Effect.tapError(Effect.logError),
     Effect.orElseSucceed(() => []),
     Effect.map((stack) => {
       // now, a single span can have a stack with multiple locations
       // so we need to duplicate the span for each location
-      const spans: Array<SpanStackEntry> = []
-      const stackEntries = [...stack]
-      while (stackEntries.length > 0) {
-        const entry = stackEntries.shift()!
-        switch (entry._tag) {
-          case "Span": {
-            let match = false
-            for (let stackIndex = 0; stackIndex < entry.stack.length; stackIndex++) {
-              const stackLine = entry.stack[stackIndex]!
-              match = true
-              spans.push(
-                new SpanStackEntry({
-                  ...stackLine,
-                  ...entry,
-                  stackIndex
-                })
-              )
-            }
-
-            if (!match) {
-              spans.push(new SpanStackEntry({ ...entry, stackIndex: -1, line: 0, column: 0 }))
-            }
-            break
-          }
-          case "External": {
-            spans.push(
-              new SpanStackEntry({
-                ...entry,
-                name: "<external span " + entry.spanId + ">",
-                stackIndex: -1,
-                line: 0,
-                column: 0,
-                attributes: []
-              })
-            )
-            break
-          }
-        }
+      let spans: Array<SpanStackEntry> = []
+      for (const entry of stack) {
+        spans = [...spans, ...spanEntryToSpanStackEntry(entry)]
       }
-
       return spans
     })
   )
 }
 
 export const getCurrentSpanStack = (threadId: number | undefined) =>
-  getFiberCurrentSpan(`globalThis["effect/FiberCurrent"]`, threadId)
+  getFiberCurrentSpan(`globalThis["effect/FiberCurrent"]`, 0, threadId)
 
 // --
 
@@ -271,9 +292,11 @@ const getCurrentFibers = (threadId: number | undefined) =>
       Effect.all(
         fibers.map((fiber, idx) =>
           Effect.map(
-            getFiberCurrentSpan(`(globalThis["effect/devtools/instrumentation"].fibers || [])[${idx}]`, threadId),
+            getFiberCurrentSpan(`(globalThis["effect/devtools/instrumentation"].fibers || [])[${idx}]`, 1, threadId),
             (stack) => new FiberEntry({ ...fiber, stack })
-          ), { concurrency: "unbounded" })
+          )
+        ),
+        { concurrency: "unbounded" }
       )
     )
   )

src/VsCode.ts

@@ -32,8 +32,8 @@ export const dismissable = <A>(
   f: () => Thenable<A | undefined>
 ): Effect.Effect<A, Cause.NoSuchElementException> => thenable(f).pipe(Effect.flatMap(Effect.fromNullable))
 
-export const executeCommand = (command: string, ...args: Array<any>) =>
-  thenable(() => vscode.commands.executeCommand(command, ...args))
+export const executeCommand = <A = unknown>(command: string, ...args: Array<any>) =>
+  thenable(() => vscode.commands.executeCommand<A>(command, ...args))
 
 export const registerCommand = <R, E, A>(
   command: string,