Repository: stateful/vscode-runme
Pull request #205 (2023-02-14 21:47:57)

Integrate gRPC Runner Service
Solves #202 

![2023-02-14 15-13-55](https://user-images.githubusercontent.com/16108792/218866900-43a92dfe-c0c9-4ed6-86a8-6f3388895a0d.gif)

Currently facing the following issues:

@adambabik 

 - `InputData` sent in the `sendRawInput` function (located in `runner.ts > GrpcRunnerProgramSession`) is completely empty by the time it gets to `runnerService.Execute` on the server. It's getting totally dropped, and this is a big blocker for interactive support. Do you have any idea why this might be? @sourishkrout please test and let me know if you have the same issue/any thoughts here, I have absolutely no idea what could be causing this...
 - I get the following trace when running Execute. I think what's happening here is that the program has exited, so the stream has ended, but `srv.Recv()` still gets called one last time, resulting in this "rpc Canceled" error. If this is not a client-side issue, I think we should make an issue about it on the `runme` repo.
```
2023-02-14T15:23:11.950-0600	INFO	runner/service.go:122	running Execute in runnerService
2023-02-14T15:23:11.969-0600	INFO	runner/service.go:204	finished command execution	{"exitCode": 0}
2023-02-14T15:23:11.969-0600	INFO	runner/service.go:173	failed to receive requests; stopping the command	{"error": "rpc error: code = Canceled desc = context canceled"}
2023-02-14T15:23:11.969-0600	INFO	runner/service.go:175	failed to stop the command	{"error": "os: process already finished", "errorVerbose": "os: process already finished\ngithub.com/stateful/runme/internal/runner.(*command).Stop\n\t/Users/mxs/Projects/runme/internal/runner/command.go:343\ngithub.com/stateful/runme/internal/runner.(*runnerService).Execute.func1\n\t/Users/mxs/Projects/runme/internal/runner/service.go:174\nruntime.goexit\n\t/opt/homebrew/Cellar/go/1.19.5/libexec/src/runtime/asm_arm64.s:1172"}
```
 - Is there any way to request a program exit other than sending an interrupt? Currently if the user requests a command stop,  the client sends an `EOF`, but then the program continues running. I think maybe that the program should try to stop if its still running during an `EOF`, since an interrupt can be cancelled by the program, but something like `cmd.Stop()` cannot - what do you think?

@sourishkrout 

 - I added the main functionality to `kernel.ts` as an inline async function. Much of this is copy/pasted and/or adapted from `taskExecutor` and `shellExecutor`. I want to bring this to its own file/method, but its not clear exactly how. I basically see two ways to proceed:
   1. Add a new `executor` function which runs the given runner. This is basically what I've done already, but results in lots of duplication/furthering the aforementioned issue of cementing the `executor` model.
   2. Create an abstraction - my idea is something like an abstract class `ExecutionNode`, so for the runner it will be `RunnerExecutionNode`. For now it will be implemented rather simply (maybe it flatly produces some kind of "`ExecutionResult`" interface), but I think this is a good step towards things like multiple output stages, cleaner input utilization, chaining, etc. 
 - Basically, if I understood correctly, in our 1:1 you were describing a desire for a kind of "middle ground" solution between a simplistic function (executor) and a full-blown solution. I think I can implement a class (`ExecutionNode`) which satisfies this desire and also can be built upon going forward. Maybe the exact abstraction described in (ii) isn't the one you want to go with, but in any case I don't think it will be a significant development effort to implement it. But I need your approval before going this way, as we could easily just do (i) and then work towards (ii) later/in follow up PRs, and I know this feature is time-sensitive.

Other issues:

 - Currently there is one environment per `Kernel`. We need to set-up a system that puts the session-id on the notebook metadata if we want per-notebook environment. @sourishkrout I could use your advice here - where in the code is the mechanism to manage notebook metadata? This metadata in particular should probably not be sent to the serializer since its irrelevant for serialization.
 - Environment is maybe not deleted on exit. I tried to make this happen as part of the `dispose` cycle, but it's still janky and needs further testing

sourishkrout: > * `InputData` sent in the `sendRawInput` function (located in `runner.ts > GrpcRunnerProgramSession`) is completely empty by the time it gets to `runnerService.Execute` on the server. It's getting totally dropped, and this is a big blocker for interactive support. Do you have any idea why this might be? @sourishkrout please test and let me know if you have the same issue/any thoughts here, I have absolutely no idea what could be causing this...

I think i see the problem and have a hunch why but I'm not sure if I'm looking at the right "issue". What are the steps to reproduce for you @mxsdev?
mxsdev: @sourishkrout 

I included a line that printed the `InputData` as-is in the `runme` go code, and it was always empty - didn't matter what I ran. Just tested `node ./scripts/stdin.js` and have the same issue.

I use `zsh` on my system, if that's relevant.
sourishkrout: > @sourishkrout
> 
> I included a line that printed the `InputData` as-is in the `runme` go code, and it was always empty - didn't matter what I ran. Just tested `node ./scripts/stdin.js` and have the same issue. Also have the same issue with a `read` command in a bash script.
> 
> I use `zsh` on my system, if that's relevant.

I believe the problem you're running up against here is that the Pty is not fully implemented. In other words, individual keystrokes result in individual calls of input data (and likely individual requests) and aren't being buffered until, let's say RETURN is being pressed what would tell the Pty to send the buffered input stream off to the server.

That's the fun of implementing the Pseudoterminal, now one has to define and implement the different user interactions with a underlying terminal.
adambabik: > I get the following trace when running Execute. I think what's happening here is that the program has exited, so the stream has ended, but srv.Recv() still gets called one last time, resulting in this "rpc Canceled" error. If this is not a client-side issue, I think we should make an issue about it on the runme repo.

Ideally, you'd like to notify the runner service that you want to drop the connection and let it do it. For example, when running a long running process like an HTTP server, try to kill it first by sending `0x03` byte (you likely need to start it with TTY (`ExecuteRequest.tty`). Then, the server should close the connection and send the exit code of the process. Alternatively, you can handle the `rpc Canceled` error if that's what you expect.

> Is there any way to request a program exit other than sending an interrupt? Currently if the user requests a command stop, the client sends an EOF, but then the program continues running. I think maybe that the program should try to stop if its still running during an EOF, since an interrupt can be cancelled by the program, but something like cmd.Stop() cannot - what do you think?

Can you send the command and how you execute it (with TTY or not), please? Typically, the command should exit on its own. If you want to interrupt it, currently, you must start it with TTY and send `0x03` and `0x04` bytes (at least for bash).

Regarding the second part, when a program handles signals and ignores them, the client can drop the connection and the program will exit. You should try that with https://github.com/stateful/runme/pull/127 as it might have been broken.

Another solution would be to add a field to `ExecuteRequest` which would indicate a force stop.
adambabik: > Alternatively, you can handle the rpc Canceled error if that's what you expect.

There is actually another way that is preferable. I don't know exactly how the TS interface looks like but when you start a bi-directional stream, you can close the send direction with `stream.CloseSend()` in Go. This will notify the server that the client no longer wants to send anything. When this happens, the runner service will close the process' stdin, which usually terminates the whole process.

Check out [`service_test.go`](https://github.com/stateful/runme/blob/6c45d2824168ef8428d2f5784b8b85fa58a8eba3/internal/runner/service_test.go) to see different variants. Overall, the following should always hold:
- If you start an execution with TTY, try to interrupt the running command and close the send direction.
- If you start an execution without TTY, the only reliable way is to cancel, unless the program exits when the stdin is closed (`bash` works that way, for example), then closing the send direction is enough.
- If the executing command is unresponsive, cancelling will force it to stop.
sourishkrout: > * I added the main functionality to `kernel.ts` as an inline async function. Much of this is copy/pasted and/or adapted from `taskExecutor` and `shellExecutor`. I want to bring this to its own file/method, but its not clear exactly how. I basically see two ways to proceed:
>   
>   1. Add a new `executor` function which runs the given runner. This is basically what I've done already, but results in lots of duplication/furthering the aforementioned issue of cementing the `executor` model.
>   2. Create an abstraction - my idea is something like an abstract class `ExecutionNode`, so for the runner it will be `RunnerExecutionNode`. For now it will be implemented rather simply (maybe it flatly produces some kind of "`ExecutionResult`" interface), but I think this is a good step towards things like multiple output stages, cleaner input utilization, chaining, etc.
> * Basically, if I understood correctly, in our 1:1 you were describing a desire for a kind of "middle ground" solution between a simplistic function (executor) and a full-blown solution. I think I can implement a class (`ExecutionNode`) which satisfies this desire and also can be built upon going forward. Maybe the exact abstraction described in (ii) isn't the one you want to go with, but in any case I don't think it will be a significant development effort to implement it. But I need your approval before going this way, as we could easily just do (i) and then work towards (ii) later/in follow up PRs, and I know this feature is time-sensitive.

Just a quick comment on this one: I'd like to keep us focused on getting this to a place where it's shippable. Meaning, not try to do too many things in one PR. So my advice would be to potentially keep it simple and see if we can design a test suite that will help us with a refactor later. If there's an incremental step forward we can make, feel free to go ahead. From what I understand, I believe your thoughts are on the right track. Again, I'd suggest to keep it simple for now and capture notes for another PR to evolve executors down the line.
mxsdev: Should be good on my end, basically waiting for [#131](https://github.com/stateful/runme/issues/131) and [#130](https://github.com/stateful/runme/issues/130) before we can review/merge 👍 
sourishkrout: > PR should be basically ready, contingent on runme [#134](https://github.com/stateful/runme/pull/134) being merged.

cool 🤘. i'll start my review in a little bit. hopefully will finish before EoD.
mxsdev: > cool 🤘. i'll start my review in a little bit. hopefully will finish before EoD.

FYI, if you want to do testing, you'll need to link the newest generated files from that PR, since there's been a change to the proto defs.
sourishkrout: Overall this is a big step in the right direction. Let's figure out what the definition of "done" is for this PR.
sourishkrout: Let's look into `message: 'gRPC Server closed output stream unexpectedly!'` after interactive execution completes.

List of incomplete functionality outside of this PR:
- [ ] Environment Variables inherited from host-process
- [ ] Use correct $SHELL
- [ ] User-prompts for ENV variables
- [ ] Deep integrations: Vercel/deno
- [ ] Terminal dimensions (not high priority)
mxsdev: @sourishkrout Anything more preventing this merge?

Patch:
diff --git a/package-lock.json b/package-lock.json
index 95bf01a2..a63ff108 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -10,7 +10,7 @@
       "hasInstallScript": true,
       "license": "Apache-2.0",
       "dependencies": {
-        "@buf/stateful_runme.community_timostamm-protobuf-ts": "^2.8.2-20230119154408-6077dbdcb8b5.1",
+        "@buf/stateful_runme.community_timostamm-protobuf-ts": "2.8.2-20230217155517-b1ae5526a72e.1",
         "@grpc/grpc-js": "^1.8.4",
         "@protobuf-ts/grpc-transport": "^2.8.2",
         "@protobuf-ts/runtime": "^2.8.2",
@@ -1711,8 +1711,8 @@
       "dev": true
     },
     "node_modules/@buf/stateful_runme.community_timostamm-protobuf-ts": {
-      "version": "2.8.2-20230119154408-6077dbdcb8b5.1",
-      "resolved": "https://buf.build/gen/npm/v1/@buf/stateful_runme.community_timostamm-protobuf-ts/2.8.2-20230119154408-6077dbdcb8b5.1/tarball",
+      "version": "2.8.2-20230217155517-b1ae5526a72e.1",
+      "resolved": "https://buf.build/gen/npm/v1/@buf/stateful_runme.community_timostamm-protobuf-ts/2.8.2-20230217155517-b1ae5526a72e.1/tarball",
       "peerDependencies": {
         "@protobuf-ts/runtime": "^2.8.2",
         "@protobuf-ts/runtime-rpc": "^2.8.2"
@@ -23832,8 +23832,8 @@
       "dev": true
     },
     "@buf/stateful_runme.community_timostamm-protobuf-ts": {
-      "version": "2.8.2-20230119154408-6077dbdcb8b5.1",
-      "resolved": "https://buf.build/gen/npm/v1/@buf/stateful_runme.community_timostamm-protobuf-ts/2.8.2-20230119154408-6077dbdcb8b5.1/tarball",
+      "version": "2.8.2-20230217155517-b1ae5526a72e.1",
+      "resolved": "https://buf.build/gen/npm/v1/@buf/stateful_runme.community_timostamm-protobuf-ts/2.8.2-20230217155517-b1ae5526a72e.1/tarball",
       "requires": {}
     },
     "@cspotcode/source-map-support": {
diff --git a/package.json b/package.json
index cba2ef23..fe33e28b 100644
--- a/package.json
+++ b/package.json
@@ -241,6 +241,12 @@
             "default": false,
             "markdownDescription": "If set to `true`, the notebook serialization will use the GRPC interface (experimental)."
           },
+          "runme.experiments.grpcRunner": {
+            "type": "boolean",
+            "scope": "machine",
+            "default": false,
+            "markdownDescription": "If set to `true`, the notebook runner will use the GRPC interface (experimental)."
+          },
           "runme.checkout.projectDir": {
             "type": "string",
             "scope": "machine",
@@ -416,7 +422,7 @@
     "webpack-cli": "^5.0.1"
   },
   "dependencies": {
-    "@buf/stateful_runme.community_timostamm-protobuf-ts": "^2.8.2-20230119154408-6077dbdcb8b5.1",
+    "@buf/stateful_runme.community_timostamm-protobuf-ts": "2.8.2-20230217155517-b1ae5526a72e.1",
     "@grpc/grpc-js": "^1.8.4",
     "@protobuf-ts/grpc-transport": "^2.8.2",
     "@protobuf-ts/runtime": "^2.8.2",
diff --git a/src/extension/executors/runner.ts b/src/extension/executors/runner.ts
new file mode 100644
index 00000000..0d7906c5
--- /dev/null
+++ b/src/extension/executors/runner.ts
@@ -0,0 +1,168 @@
+import path from 'node:path'
+
+import {
+  NotebookCellOutputItem,
+  NotebookCellOutput,
+  Task,
+  TaskScope,
+  CustomExecution,
+  TaskRevealKind,
+  TaskPanelKind,
+  tasks,
+  NotebookCellExecution,
+  TextDocument
+} from 'vscode'
+
+import { OutputType } from '../../constants'
+import { CellOutputPayload } from '../../types'
+import { PLATFORM_OS } from '../constants'
+import { IRunner, IRunnerEnvironment } from '../runner'
+import { getAnnotations, getCmdShellSeq } from '../utils'
+
+import { closeTerminalByEnvID } from './task'
+import { getShellPath } from './utils'
+
+const LABEL_LIMIT = 15
+const MIME_TYPES_WITH_CUSTOM_RENDERERS = ['text/plain']
+
+export async function executeRunner(
+  runner: IRunner,
+  exec: NotebookCellExecution,
+  runningCell: TextDocument,
+  execKey: 'bash'|'sh',
+  environment?: IRunnerEnvironment
+) {
+  const cwd = path.dirname(runningCell.uri.fsPath)
+
+  const RUNME_ID = `${runningCell.fileName}:${exec.cell.index}`
+
+  const cellText = exec.cell.document.getText()
+  const script = getCmdShellSeq(cellText, PLATFORM_OS)
+
+  const annotations = getAnnotations(exec.cell)
+  const { interactive, mimeType, background } = annotations
+
+  const program = await runner.createProgramSession({
+    programName: getShellPath(execKey),
+    environment,
+    script: {
+      type: 'script', script
+    },
+    envs: [
+      `RUNME_ID=${RUNME_ID}`
+    ],
+    cwd,
+    tty: interactive
+  })
+
+  if(!interactive) {
+    const output: Buffer[] = []
+
+    const mime = mimeType || 'text/plain' as const
+
+    // adapted from `shellExecutor` in `shell.ts`
+    const handleOutput = (data: Uint8Array) => {
+      output.push(Buffer.from(data))
+
+      let item = new NotebookCellOutputItem(Buffer.concat(output), mime)
+
+      // hacky for now, maybe inheritence is a fitting pattern
+      if (script.trim().endsWith('vercel')) {
+        // TODO: vercel (see `shellExecutor`)
+      } else if (MIME_TYPES_WITH_CUSTOM_RENDERERS.includes(mime)) {
+        item = NotebookCellOutputItem.json(<CellOutputPayload<OutputType.outputItems>>{
+          type: OutputType.outputItems,
+          output: {
+            content: Buffer.concat(output).toString('base64'),
+            mime
+          }
+        }, OutputType.outputItems)
+      }
+
+      exec.clearOutput() // bug? https://github.com/microsoft/vscode/issues/173577
+      exec.replaceOutput([ new NotebookCellOutput([ item ]) ])
+    }
+
+    program.onStdoutRaw(handleOutput)
+    program.onStderrRaw(handleOutput)
+
+    exec.token.onCancellationRequested(() => {
+      program.close()
+    })
+
+    await program.run()
+  } else {
+    const taskExecution = new Task(
+      { type: 'shell', name: `Runme Task (${RUNME_ID})` },
+      TaskScope.Workspace,
+      cellText.length > LABEL_LIMIT
+        ? `${cellText.slice(0, LABEL_LIMIT)}...`
+        : cellText,
+      'exec',
+      new CustomExecution(async () => program)
+    )
+
+    taskExecution.isBackground = background
+    taskExecution.presentationOptions = {
+      focus: true,
+      reveal: background ? TaskRevealKind.Never : TaskRevealKind.Always,
+      panel: background ? TaskPanelKind.Dedicated : TaskPanelKind.Shared
+    }
+
+    const execution = await tasks.executeTask(taskExecution)
+
+    exec.token.onCancellationRequested(() => {
+      try {
+        // runs `program.close()` implicitly
+        execution.terminate()
+        closeTerminalByEnvID(RUNME_ID)
+      } catch (err: any) {
+        // console.error(`[Runme] Failed to terminate task: ${(err as Error).message}`)
+        throw new Error(`[Runme] Failed to terminate task: ${(err as Error).message}`)
+      }
+    })
+
+    tasks.onDidEndTaskProcess((e) => {
+      const taskId = (e.execution as any)['_id']
+      const executionId = (execution as any)['_id']
+
+      /**
+       * ignore if
+       */
+      if (
+        /**
+         * VS Code is running a different task
+         */
+        taskId !== executionId ||
+        /**
+         * we don't have an exit code
+         */
+        typeof e.exitCode === 'undefined') {
+        return
+      }
+
+      /**
+       * only close terminal if execution passed and desired by user
+       */
+      if (e.exitCode === 0 && annotations.closeTerminalOnSuccess) {
+        closeTerminalByEnvID(RUNME_ID)
+      }
+    })
+  }
+
+  return await new Promise<boolean>((resolve) => {
+    program.onDidClose((code) => {
+      resolve(code === 0)
+    })
+
+    program.onInternalErr((e) => {
+      console.error('Internal failure executing runner', e)
+      resolve(false)
+    })
+
+    if(program.hasExited()) {
+      // unexpected early return, likely an error
+      resolve(false)
+    }
+  })
+}
diff --git a/src/extension/executors/utils.ts b/src/extension/executors/utils.ts
index 5e542892..02a839ff 100644
--- a/src/extension/executors/utils.ts
+++ b/src/extension/executors/utils.ts
@@ -114,3 +114,14 @@ export async function retrieveShellCommand (exec: NotebookCellExecution) {
   }
   return cellText
 }
+
+/**
+ * Try to get shell path from environment (`$SHELL`)
+ *
+ * @param execKey Used as fallback in case `$SHELL` is not present
+ */
+export function getShellPath(): string|undefined
+export function getShellPath(execKey: 'bash'|'sh'): string
+export function getShellPath(execKey?: string): string|undefined {
+  return process.env.SHELL ?? execKey
+}
diff --git a/src/extension/extension.ts b/src/extension/extension.ts
index e80e6a1f..209e1e2a 100644
--- a/src/extension/extension.ts
+++ b/src/extension/extension.ts
@@ -24,6 +24,7 @@ import { WasmSerializer, GrpcSerializer } from './serializer'
 import { RunmeLauncherProvider } from './provider/launcher'
 import { RunmeUriHandler } from './handler/uri'
 import { BOOTFILE } from './constants'
+import { GrpcRunner } from './runner'
 
 export class RunmeExtension {
   async initialize(context: ExtensionContext) {
@@ -58,6 +59,7 @@ export class RunmeExtension {
     const serializer = grpcSerializer ? new GrpcSerializer(context) : new WasmSerializer(context)
     const treeViewer = new RunmeLauncherProvider(getDefaultWorkspace())
     const uriHandler = new RunmeUriHandler()
+    kernel.hasExperimentEnabled('grpcRunner') && kernel.useRunner(new GrpcRunner())
 
     context.subscriptions.push(
       kernel,
diff --git a/src/extension/grpc/client.ts b/src/extension/grpc/client.ts
index 56a86191..85451ec1 100644
--- a/src/extension/grpc/client.ts
+++ b/src/extension/grpc/client.ts
@@ -1,4 +1,6 @@
 // eslint-disable-next-line max-len
 import { ParserServiceClient } from '@buf/stateful_runme.community_timostamm-protobuf-ts/runme/parser/v1/parser_pb.client'
+// eslint-disable-next-line max-len
+import { RunnerServiceClient } from '@buf/stateful_runme.community_timostamm-protobuf-ts/runme/runner/v1/runner_pb.client'
 
-export { ParserServiceClient }
+export { ParserServiceClient, RunnerServiceClient }
diff --git a/src/extension/grpc/runnerTypes.ts b/src/extension/grpc/runnerTypes.ts
new file mode 100644
index 00000000..b084cf62
--- /dev/null
+++ b/src/extension/grpc/runnerTypes.ts
@@ -0,0 +1,1 @@
+export * from '@buf/stateful_runme.community_timostamm-protobuf-ts/runme/runner/v1/runner_pb'
\ No newline at end of file
diff --git a/src/extension/grpc/types.ts b/src/extension/grpc/serializerTypes.ts
similarity index 100%
rename from src/extension/grpc/types.ts
rename to src/extension/grpc/serializerTypes.ts
diff --git a/src/extension/kernel.ts b/src/extension/kernel.ts
index 2246f875..311e6e24 100644
--- a/src/extension/kernel.ts
+++ b/src/extension/kernel.ts
@@ -1,7 +1,16 @@
 import {
-  Disposable, notebooks, window, workspace, ExtensionContext,
-  NotebookEditor, NotebookCell, NotebookCellKind, NotebookCellExecution, WorkspaceEdit, NotebookEdit, NotebookDocument
-} from 'vscode'
+  Disposable,
+  notebooks,
+  window,
+  workspace,
+  ExtensionContext,
+  NotebookEditor,
+  NotebookCell,
+  NotebookCellKind,
+  NotebookCellExecution,
+  WorkspaceEdit,
+  NotebookEdit,
+  NotebookDocument} from 'vscode'
 import { TelemetryReporter } from 'vscode-telemetry'
 
 import type { ClientMessage } from '../types'
@@ -12,8 +21,9 @@ import executor, { runme } from './executors'
 import { ExperimentalTerminal } from './terminal/terminal'
 import { ENV_STORE, DENO_ACCESS_TOKEN_KEY } from './constants'
 import { resetEnv, getKey, getAnnotations, hashDocumentUri } from './utils'
-
 import './wasm/wasm_exec.js'
+import { IRunner, IRunnerEnvironment } from './runner'
+import { executeRunner } from './executors/runner'
 
 enum ConfirmationItems {
   Yes = 'Yes',
@@ -36,10 +46,16 @@ export class Kernel implements Disposable {
   )
   protected messaging = notebooks.createRendererMessaging('runme-renderer')
 
-  constructor(protected context: ExtensionContext) {
+  protected runner?: IRunner
+  protected environment?: IRunnerEnvironment
+
+  constructor(
+    protected context: ExtensionContext,
+  ) {
     const config = workspace.getConfiguration('runme.experiments')
     this.#experiments.set('pseudoterminal', config.get<boolean>('pseudoterminal', false))
     this.#experiments.set('grpcSerializer', config.get<boolean>('grpcSerializer', false))
+    this.#experiments.set('grpcRunner', config.get<boolean>('grpcRunner', false))
 
     this.#controller.supportedLanguages = Object.keys(executor)
     this.#controller.supportsExecutionOrder = false
@@ -237,14 +253,35 @@ export class Kernel implements Disposable {
     exec.start(Date.now())
     let execKey = getKey(runningCell)
 
-    /**
-     * check if user is running experiment to execute shell via runme cli
-     */
     const hasPsuedoTerminalExperimentEnabled = this.hasExperimentEnabled('pseudoterminal')
     const terminal = this.#terminals.get(cell.document.uri.fsPath)
-    const successfulCellExecution = (hasPsuedoTerminalExperimentEnabled && terminal)
-      ? await runme.call(this, exec, terminal)
-      : await executor[execKey].call(this, exec, runningCell)
+
+    let successfulCellExecution: boolean
+
+    if(
+      this.runner &&
+      !(hasPsuedoTerminalExperimentEnabled && terminal) &&
+      (execKey === 'bash' || execKey === 'sh')
+    ) {
+      successfulCellExecution = await executeRunner(
+        this.runner,
+        exec,
+        runningCell,
+        execKey,
+        this.environment
+      )
+        .catch((e) => {
+          console.error(e)
+          return false
+        })
+    } else {
+      /**
+       * check if user is running experiment to execute shell via runme cli
+       */
+      successfulCellExecution = (hasPsuedoTerminalExperimentEnabled && terminal)
+        ? await runme.call(this, exec, terminal)
+        : await executor[execKey].call(this, exec, runningCell)
+    }
     TelemetryReporter.sendTelemetryEvent('cell.endExecute', { 'cell.success': successfulCellExecution?.toString() })
     exec.end(successfulCellExecution)
   }
@@ -265,4 +302,17 @@ export class Kernel implements Disposable {
     const runmeTerminal = new ExperimentalTerminal(editor.notebook)
     this.#terminals.set(editor.notebook.uri.fsPath, runmeTerminal)
   }
+
+  useRunner(runner: IRunner) {
+    if(this.#experiments.get('grpcRunner') && runner) {
+      this.runner = runner
+
+      this.#disposables.push(runner)
+      runner.createEnvironment(
+        // copy env from process naively for now
+        // later we might want a more sophisticated approach/to bring this serverside
+        Object.entries(process.env).map(([k, v]) => `${k}=${v || ''}`)
+      ).then(env => { this.environment = env })
+    }
+  }
 }
diff --git a/src/extension/runner.ts b/src/extension/runner.ts
new file mode 100644
index 00000000..6df37f7d
--- /dev/null
+++ b/src/extension/runner.ts
@@ -0,0 +1,412 @@
+import { ChannelCredentials } from '@grpc/grpc-js'
+import { GrpcTransport } from '@protobuf-ts/grpc-transport'
+import { DuplexStreamingCall } from '@protobuf-ts/runtime-rpc/build/types/duplex-streaming-call'
+import {
+  Pseudoterminal,
+  Event,
+  EventEmitter,
+  Disposable,
+} from 'vscode'
+import type { UInt32Value } from '@buf/stateful_runme.community_timostamm-protobuf-ts/google/protobuf/wrappers_pb'
+import { RpcError } from '@protobuf-ts/runtime-rpc'
+
+import type { DisposableAsync } from '../types'
+
+import {
+  CreateSessionRequest,
+  ExecuteRequest,
+  ExecuteResponse,
+  ExecuteStop,
+  Session,
+} from './grpc/runnerTypes'
+import { RunnerServiceClient } from './grpc/client'
+
+type ExecuteDuplex = DuplexStreamingCall<ExecuteRequest, ExecuteResponse>
+
+export interface RunProgramOptions {
+  programName: string
+  args?: string[]
+  cwd?: string
+  envs?: string[]
+  script?:
+    |{
+      type: 'commands'
+      commands: string[]
+    }
+    |{
+      type: 'script'
+      script: string
+    }
+  tty?: boolean
+  environment?: IRunnerEnvironment
+}
+
+export interface IRunner extends Disposable {
+  close(): void
+
+  createEnvironment(
+    envs?: string[],
+    metadata?: { [index: string]: string }
+  ): Promise<IRunnerEnvironment>
+
+  createProgramSession(opts: RunProgramOptions): Promise<IRunnerProgramSession>
+}
+
+export interface IRunnerEnvironment extends DisposableAsync { }
+
+export interface IRunnerProgramSession extends DisposableAsync, Pseudoterminal {
+  /**
+   * Called when an unrecoverable error occurs, for instance a failure over gRPC
+   * transport
+   *
+   * Note that this is different from `onDidErr`, which is for stderr
+   */
+  readonly onInternalErr: Event<Error>
+
+  readonly onDidErr: Event<string>
+  readonly onDidClose: Event<number>
+
+  /**
+   * Implementers should **still call `onDidWrite`** to stay compatible with
+   * VSCode pseudoterminal interface
+   */
+  readonly onStdoutRaw: Event<Uint8Array>
+  /**
+   * Implementers should **still call `onDidErr`** to stay compatible with
+   * VSCode pseudoterminal interface
+   */
+  readonly onStderrRaw: Event<Uint8Array>
+
+  handleInput(message: string): Promise<void>
+
+  setRunOptions(opts: RunProgramOptions): void
+  run(): Promise<void>
+  hasExited(): boolean
+}
+
+export class GrpcRunner implements IRunner {
+  protected readonly client: RunnerServiceClient
+  protected transport: GrpcTransport
+
+  private children: WeakRef<DisposableAsync>[] = []
+
+  constructor() {
+    this.transport = new GrpcTransport({
+      host: 'unix:///tmp/runme.sock',
+      channelCredentials: ChannelCredentials.createInsecure(),
+    })
+
+    this.client = new RunnerServiceClient(this.transport)
+  }
+
+  async createProgramSession(
+    opts: RunProgramOptions
+  ): Promise<IRunnerProgramSession> {
+    const session = new GrpcRunnerProgramSession(
+      this.client,
+      opts
+    )
+
+    this.registerChild(session)
+
+    return session
+  }
+
+  createEnvironment(
+    envs?: string[],
+    metadata?: { [index: string]: string }
+  ) {
+    const request = CreateSessionRequest.create({
+      metadata, envs
+    })
+
+    try {
+      return this.client
+        .createSession(request)
+        .then(({ response: { session } }) => {
+          if(!session) {
+            throw new Error('Did not receive session!!')
+          }
+
+          const environment = new GrpcRunnerEnvironment(
+            this.client,
+            session
+          )
+
+          this.registerChild(environment)
+          return environment
+        })
+        .catch((e) => {
+          throw e
+        })
+    } catch (err: any) {
+      console.error(err)
+      this.close()
+      throw err
+    }
+  }
+
+  close(): void {
+    this.transport.close()
+  }
+
+  async dispose(): Promise<void> {
+    await Promise.all(
+      this.children.map(c => c.deref()?.dispose())
+    ).finally(() => this.close())
+  }
+
+  /**
+   * Register disposable child weakref, so that it can still be GC'ed even if
+   * there is a reference to it in this object and its already been disposed of
+   */
+  protected registerChild(d: DisposableAsync) {
+    this.children.push(new WeakRef(d))
+  }
+}
+
+export class GrpcRunnerProgramSession implements IRunnerProgramSession {
+  private disposables: Disposable[] = []
+
+  readonly _onInternalErr = this.register(new EventEmitter<Error>())
+  readonly _onDidWrite    = this.register(new EventEmitter<string>())
+  readonly _onDidErr      = this.register(new EventEmitter<string>())
+  readonly _onDidClose    = this.register(new EventEmitter<number>())
+  readonly _onStdoutRaw   = this.register(new EventEmitter<Uint8Array>())
+  readonly _onStderrRaw   = this.register(new EventEmitter<Uint8Array>())
+
+  readonly onDidWrite = this._onDidWrite.event
+  readonly onDidErr = this._onDidErr.event
+  readonly onDidClose = this._onDidClose.event
+  readonly onStdoutRaw = this._onStdoutRaw.event
+  readonly onStderrRaw = this._onStderrRaw.event
+  readonly onInternalErr = this._onInternalErr.event
+
+  private readonly session: ExecuteDuplex
+
+  private exitCode: UInt32Value|undefined
+
+  private isDisposed = false
+
+  protected initialized = false
+
+  protected buffer = ''
+
+  constructor(
+    private readonly client: RunnerServiceClient,
+    protected opts: RunProgramOptions
+  ) {
+    this.session = client.execute()
+
+    this.register(
+      this._onStdoutRaw.event((data) => {
+        // TODO: web compat
+        const stdout = Buffer.from(data).toString('utf-8')
+        this._onDidWrite.fire(stdout)
+      })
+    )
+
+    this.register(
+      this._onStderrRaw.event((data) => {
+        // TODO: web compat
+        const stderr = Buffer.from(data).toString('utf-8')
+        this._onDidErr.fire(stderr)
+      })
+    )
+
+    this.session.responses.onMessage(({ stderrData, stdoutData, exitCode }) => {
+      if(stdoutData.length > 0) {
+        this._onStdoutRaw.fire(stdoutData)
+      }
+
+      if(stderrData.length > 0) {
+        this._onStderrRaw.fire(stderrData)
+      }
+
+      if(exitCode) {
+        this._onDidClose.fire(exitCode.value)
+        this.exitCode = exitCode
+        this.dispose()
+      }
+    })
+
+    this.session.responses.onComplete(() => {
+      if(!this.hasExited()) {
+        this.error(new Error('gRPC Server closed output stream unexpectedly!'))
+      }
+    })
+
+    this.session.responses.onError((error) => {
+      if(error instanceof RpcError) {
+        console.error('RpcError occurred!', {
+          // duping here since `Error` types are uninspectable in console
+          code: error.code,
+          message: error.message,
+          method: error.methodName,
+          meta: error.meta,
+          service: error.serviceName,
+          name: error.name,
+        }, error)
+      } else {
+        console.error('Unexpected error!!', error)
+      }
+
+      this.error(error)
+    })
+  }
+
+  protected async init(opts?: RunProgramOptions) {
+    if(this.initialized) { throw new Error('Already initialized!') }
+    if(opts) { this.opts = opts }
+
+    this.initialized = true
+
+    await this.session.requests.send(GrpcRunnerProgramSession.runOptionsToExecuteRequest(this.opts))
+  }
+
+  setRunOptions(opts: RunProgramOptions): void {
+    this.opts = opts
+  }
+
+  async run(opts?: RunProgramOptions): Promise<void> {
+    await this.init(opts)
+  }
+
+  isPseudoterminal(): boolean {
+    return !!this.opts.tty
+  }
+
+  async handleInput(data: string): Promise<void> {
+    if(this.hasExited()) { throw new Error('Cannot write to closed program session!') }
+    this.sendRawInput(data)
+
+    // if(this.isPseudoterminal()) {
+    //   switch (data) {
+    //     case '\r': // Enter
+    //       const command = this.buffer + '\n'
+
+    //       // lastCommand = buffer
+    //       console.log("Sending command", command)
+
+    //       this.sendRawInput(command)
+    //         // .then(() => console.log('sent input', command))
+    //       break
+    //     case '\u007F': // Backspace (DEL)
+    //       if (this.buffer.length > 0) {
+    //         this._onDidWrite.fire('\b \b')
+    //         if (this.buffer.length > 0) {
+    //           this.buffer = this.buffer.slice(0, this.buffer.length - 1)
+    //         }
+    //       }
+    //       break
+    //     case '\u0003': // Ctrl+C
+    //       this._onDidWrite.fire('^C')
+    //       break
+    //     default:
+    //       if (
+    //         (data >= String.fromCharCode(0x20) &&
+    //           data <= String.fromCharCode(0x7e)) ||
+    //         data >= '\u00a0'
+    //       ) {
+    //         this.buffer += data
+    //         this._onDidWrite.fire(data)
+    //       }
+    //   }
+    // } else {
+    //   this.sendRawInput(data)
+    // }
+  }
+
+  protected async sendRawInput(data: string) {
+    const inputData = Buffer.from(data)
+
+    this.session.requests.send(ExecuteRequest.create({
+      inputData: inputData
+    }))
+  }
+
+  open(): void {
+    // in pty, we wait for open to run
+    this.run()
+  }
+
+  async dispose() {
+    if(this.isDisposed) { return }
+    this.isDisposed = true
+
+    await this.session.requests.complete()
+  }
+
+  /**
+   * Manually closed by the user
+   *
+   * Implemented for compatibility with VSCode's `Pseudoterminal` interface;
+   * please use `_close` internally
+   */
+  close() {
+    if(this.hasExited()) { return }
+
+    this.session.requests.send(ExecuteRequest.create({
+      stop: this.isPseudoterminal() ? ExecuteStop.INTERRUPT : ExecuteStop.KILL
+    }))
+  }
+
+  /**
+   * Unrecoverable internal error
+   */
+  private error(error: Error) {
+    this._onInternalErr.fire(error)
+    this.dispose()
+  }
+
+  hasExited() {
+    return this.exitCode !== undefined || this.isDisposed
+  }
+
+  protected register<T extends Disposable>(disposable: T): T {
+    this.disposables.push(disposable)
+    return disposable
+  }
+
+  static runOptionsToExecuteRequest(
+    { programName, args, cwd, environment, script, tty, envs }: RunProgramOptions
+  ): ExecuteRequest {
+    if(environment && !(environment instanceof GrpcRunnerEnvironment)) {
+      throw new Error('Expected gRPC environment!')
+    }
+
+    return ExecuteRequest.create({
+      arguments: args,
+      envs,
+      directory: cwd,
+      tty,
+      sessionId: environment?.getSessionId(),
+      programName,
+      ...script?.type === 'commands' && { commands: script.commands },
+      ...script?.type === 'script' && { script: script.script }
+    })
+  }
+}
+
+export class GrpcRunnerEnvironment implements IRunnerEnvironment {
+  constructor(
+    private readonly client: RunnerServiceClient,
+    private readonly session: Session
+  ) { }
+
+  getRunmeSession(): Session {
+    return this.session
+  }
+
+  getSessionId(): string {
+    return this.session.id
+  }
+
+  async dispose() {
+    await this.delete()
+  }
+
+  private async delete() {
+    return await this.client.deleteSession({ id: this.getSessionId() })
+  }
+}
diff --git a/src/extension/serializer.ts b/src/extension/serializer.ts
index 410b8f35..772bc026 100644
--- a/src/extension/serializer.ts
+++ b/src/extension/serializer.ts
@@ -22,7 +22,7 @@ import { Serializer } from '../types'
 import { SERVER_ADDRESS } from '../constants'
 import { getPortNumber } from '../utils/configuration'
 
-import { DeserializeRequest, SerializeRequest, Notebook } from './grpc/types'
+import { DeserializeRequest, SerializeRequest, Notebook } from './grpc/serializerTypes'
 import { ParserServiceClient } from './grpc/client'
 import Languages from './languages'
 import { PLATFORM_OS } from './constants'
diff --git a/src/types.ts b/src/types.ts
index 745ce91c..27fa52d0 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -121,3 +121,7 @@ export type CellAnnotationsErrorResult = {
   errors?: CellAnnotationErrorKey
   originalAnnotations: CellAnnotations
 }
+
+export interface DisposableAsync {
+  dispose(): Promise<void>
+}


Test patch:
diff --git a/tests/extension/extension.test.ts b/tests/extension/extension.test.ts
index 063f367a..b04ed4a3 100644
--- a/tests/extension/extension.test.ts
+++ b/tests/extension/extension.test.ts
@@ -8,8 +8,11 @@ vi.mock('vscode-telemetry')
 
 vi.mock('../../src/extension/grpc/client', () => ({
   ParserServiceClient: vi.fn(),
+  RunnerServiceClient: vi.fn(),
 }))
 
+vi.mock('../../src/extension/grpc/runnerTypes', () => ({}))
+
 test('initializes all providers', async () => {
   const configValues = {
     binaryPath: '.bin/runme'
diff --git a/tests/extension/kernel.test.ts b/tests/extension/kernel.test.ts
index bba32443..b2dadc35 100644
--- a/tests/extension/kernel.test.ts
+++ b/tests/extension/kernel.test.ts
@@ -16,6 +16,7 @@ vi.mock('../../src/extension/utils', () => ({
 vi.mock('../../src/extension/executors/index.js', () => ({
   default: { foobar: vi.fn() }
 }))
+vi.mock('../../src/extension/runner', () => ({ }))
 
 const getCells = (cnt: number) => ([...new Array(cnt)]).map((_, i) => ({
   document: { getText: vi.fn().mockReturnValue(`Cell #${i}`) },
diff --git a/tests/extension/provider/annotations.test.ts b/tests/extension/provider/annotations.test.ts
index 4a98edf4..b8bf8442 100644
--- a/tests/extension/provider/annotations.test.ts
+++ b/tests/extension/provider/annotations.test.ts
@@ -30,6 +30,8 @@ vi.mock('../../../src/extension/utils', () => ({
   validateAnnotations: vi.fn()
 }))
 
+vi.mock('../../../src/extension/runner', () => ({ }))
+
 describe('Runme Annotations', () => {
   const kernel = new Kernel({} as any)
   it('should register a command when initializing', () => {
diff --git a/tests/extension/runner.test.ts b/tests/extension/runner.test.ts
new file mode 100644
index 00000000..99e0050c
--- /dev/null
+++ b/tests/extension/runner.test.ts
@@ -0,0 +1,260 @@
+import path from 'node:path'
+
+import { vi, suite, test, expect } from 'vitest'
+import type { ExecuteResponse } from '@buf/stateful_runme.community_timostamm-protobuf-ts/runme/runner/v1/runner_pb'
+
+import { 
+  GrpcRunner, 
+  GrpcRunnerEnvironment, 
+  GrpcRunnerProgramSession, 
+  RunProgramOptions 
+} from '../../src/extension/runner'
+
+vi.mock('vscode', () => ({ 
+  ...import(path.join(process.cwd(), '__mocks__', 'vscode')) ,
+  EventEmitter: getEventEmitterClass()
+}))
+
+vi.mock('@protobuf-ts/grpc-transport', () => ({
+  GrpcTransport: class {
+    constructor() {}
+
+    close() { }
+  }
+}))
+
+let id = 0
+
+const resetId = () => { id = 0 }
+
+const createSession = vi.fn(() => ({
+  id: (id++).toString()
+}))
+
+const deleteSession = vi.fn(async () => ({
+
+}))
+
+function getEventEmitterClass() {
+  return class EventEmitter<T> {
+    listeners: MessageCallback<T>[] = []
+    
+    event: Event<T> = (listener) => {
+      this.listeners.push(listener)
+  
+      return () => {
+        this.listeners = this.listeners.filter(x => x !== listener)
+      }
+    }
+  
+    fire(data: T) {
+      this.listeners.forEach(l => l(data))
+    }
+  }
+}
+
+type MessageCallback<T> = (message: T) => void
+type Event<T> = (listener: MessageCallback<T>) => (() => void)
+
+const EventEmitter = getEventEmitterClass()
+
+class MockedDuplexClientStream {
+  constructor() {}
+
+  _onMessage = new EventEmitter<ExecuteResponse>()
+  _onComplete = new EventEmitter<void>()
+  _onError = new EventEmitter<Error>()
+
+  responses = {
+    onMessage: this._onMessage.event,
+    onComplete: this._onComplete.event,
+    onError: this._onError.event
+  }
+
+  requests = {
+    send: vi.fn(async () => {}),
+    complete: vi.fn()
+  }
+}
+
+vi.mock('../../src/extension/grpc/client', () => ({
+  RunnerServiceClient: class { 
+    constructor() {}
+
+    async createSession() { 
+      return {
+        response: {
+          session: createSession()
+        }
+      }
+    }
+
+    async deleteSession() {
+      return {
+        response: deleteSession()
+      }
+    }
+
+    execute() {
+      return new MockedDuplexClientStream()
+    }
+  }
+}))
+
+vi.mock('@buf/stateful_runme.community_timostamm-protobuf-ts/runme/runner/v1/runner_pb', () => ({
+  default: { },
+  CreateSessionRequest: {
+    create: vi.fn((x: any) => x)
+  }
+}))
+
+suite('grpc Runner', () => {
+  test('environment dispose is called on runner dispose', async () => {
+    resetId()
+    deleteSession.mockClear()
+    
+    const runner = new GrpcRunner()
+    const environment = (await runner.createEnvironment()) as GrpcRunnerEnvironment
+
+    const oldEnvDispose = environment.dispose
+
+    const environmentDispose = vi.fn(async () => { 
+      await oldEnvDispose.call(environment)
+    })
+
+    environment.dispose = environmentDispose
+
+    await runner.dispose()
+
+    expect(environmentDispose).toBeCalledTimes(1)
+    expect(deleteSession).toBeCalledTimes(1)
+  })
+
+  suite('grpc program session', () => {
+    test('session dispose is called on runner dispose', async () => {
+      const { runner, session, duplex } = await createNewSession()
+      
+      const oldSessionDispose = session.dispose
+  
+      const sessionDispose = vi.fn(async () => {
+        await oldSessionDispose.call(session)
+      })
+  
+      session.dispose = sessionDispose
+  
+      await runner.dispose()
+  
+      expect(sessionDispose).toBeCalledTimes(1)
+      expect(duplex.requests.complete).toBeCalledTimes(1)
+    })
+
+    test('duplex onMessage calls stdout raw', async () => {
+      const { duplex, stdoutListener, stderrListener } = await createNewSession()
+
+      duplex._onMessage.fire({
+        stdoutData: Buffer.from('test'),
+        stderrData: Buffer.from('')
+      })
+
+      expect(stdoutListener).toBeCalledTimes(1)
+      expect(stdoutListener).toBeCalledWith(Buffer.from('test'))
+
+      expect(stderrListener).not.toBeCalled()
+    })
+
+    test('duplex onMessage calls stderr raw', async () => {
+      const { duplex, stdoutListener, stderrListener } = await createNewSession()
+
+      duplex._onMessage.fire({
+        stdoutData: Buffer.from(''),
+        stderrData: Buffer.from('test')
+      })
+
+      expect(stdoutListener).not.toBeCalled()
+      
+      expect(stderrListener).toBeCalledTimes(1)
+      expect(stderrListener).toBeCalledWith(Buffer.from('test'))
+    })
+
+    test('duplex onMessage calls onDidWrite', async () => {
+      const { duplex, writeListener } = await createNewSession()
+
+      duplex._onMessage.fire({
+        stdoutData: Buffer.from('test'),
+        stderrData: Buffer.from('')
+      })
+
+      expect(writeListener).toBeCalledTimes(1)
+      expect(writeListener).toBeCalledWith('test')
+    })
+
+    test('duplex onMessage calls onDidErr', async () => {
+      const { duplex, errListener } = await createNewSession()
+
+      duplex._onMessage.fire({
+        stdoutData: Buffer.from(''),
+        stderrData: Buffer.from('test')
+      })
+
+      expect(errListener).toBeCalledTimes(1)
+      expect(errListener).toBeCalledWith('test')
+    })
+
+    test('duplex onMessage calls close', async () => {
+      const { 
+        duplex, 
+        stdoutListener, 
+        stderrListener, 
+        closeListener 
+      } = await createNewSession()
+
+      duplex._onMessage.fire({
+        stdoutData: Buffer.from(''),
+        stderrData: Buffer.from(''),
+        exitCode: {
+          value: 1
+        }
+      })
+
+      expect(stdoutListener).not.toBeCalled()
+      expect(stderrListener).not.toBeCalled()
+
+      expect(closeListener).toBeCalledTimes(1)
+      expect(closeListener).toBeCalledWith(1)
+    })
+  })
+})
+
+function getMockedDuplex(session: GrpcRunnerProgramSession): MockedDuplexClientStream {
+  return session['session'] as unknown as MockedDuplexClientStream
+}
+
+async function createNewSession(options: Partial<RunProgramOptions> = {}, runner?: GrpcRunner) {
+  runner ??= new GrpcRunner()
+  const session = (await runner.createProgramSession({
+    programName: 'sh',
+    ...options
+  })) as GrpcRunnerProgramSession
+
+  const stdoutListener = vi.fn()
+  session.onStdoutRaw(stdoutListener)
+  const stderrListener = vi.fn()
+  session.onStderrRaw(stderrListener)
+  const closeListener = vi.fn()
+  session.onDidClose(closeListener)
+  const writeListener = vi.fn()
+  session.onDidWrite(writeListener)
+  const errListener = vi.fn()
+  session.onDidErr(errListener)
+
+  return { 
+    runner, 
+    session, 
+    duplex: getMockedDuplex(session),
+    stdoutListener,
+    stderrListener,
+    closeListener,
+    writeListener,
+    errListener,
+  }
+}
\ No newline at end of file