Repository: stateful/vscode-runme
Pull request #227 (2023-02-21 23:08:51)

Trim stop character from annotation


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/schema.ts b/src/schema.ts
index 37e2a6b7..61575c34 100644
--- a/src/schema.ts
+++ b/src/schema.ts
@@ -2,7 +2,14 @@ import { z } from 'zod'
 
 const CELL_RANDOM_NAME = `Cell #${Math.random().toString().slice(2)}`
 
+const cleanAnnotation = (value: string, character: string): string => {
+    return value.trim().replaceAll(/[ ]{2,}/g, '').replaceAll(character, '')
+}
+
 const falseyBoolean = (defaultValue: boolean) => z.preprocess((subject) => {
+    if (typeof subject === 'string') {
+        subject = cleanAnnotation(subject, ',')
+    }
     if (typeof subject === 'boolean') {
         return subject
     }
@@ -20,6 +27,9 @@ const boolify = (defaultValue: boolean, invalidTypeError: string = 'expected a b
         if (!subject) {
             return defaultValue
         }
+        if (typeof subject === 'string') {
+            subject = cleanAnnotation(subject, ',')
+        }
         if (typeof subject === 'string' && subject.toLowerCase() === 'false') {
             return false
         }
@@ -34,7 +44,12 @@ export const AnnotationSchema = {
     background: boolify(false),
     interactive: boolify(true),
     closeTerminalOnSuccess: boolify(true),
-    name: z.string().default(CELL_RANDOM_NAME),
+    name: z.preprocess(
+        (value) =>
+            typeof value === 'string' ?
+                cleanAnnotation(value, ',') :
+                value,
+        z.string().default(CELL_RANDOM_NAME)),
     mimeType: z
         .string()
         .refine((subject) => {
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
diff --git a/tests/extension/schema.test.ts b/tests/extension/schema.test.ts
index d38c3478..0920dc91 100644
--- a/tests/extension/schema.test.ts
+++ b/tests/extension/schema.test.ts
@@ -45,6 +45,48 @@ suite('AnnotationSchema', () => {
             const parseResult = AnnotationSchema.name.safeParse('')
             expect(parseResult.success).toBeTruthy()
         })
+
+        test('it should clean the annotation separator for true values', () => {
+            ['true,', 'true    ,', 'true   '].forEach(el => {
+                const parseResult = AnnotationSchema.name.safeParse(el)
+                expect(parseResult.success).toBeTruthy()
+            })
+
+        })
+
+        test('it should clean the annotation separator for false values', () => {
+            ['false,', 'false    ,', 'false   '].forEach(el => {
+                const parseResult = AnnotationSchema.name.safeParse(el)
+                expect(parseResult.success).toBeTruthy()
+            })
+        })
+
+        test('it should clean the annotation separator for string values', () => {
+            [{
+                value: 'annotation',
+                expectedValue: 'annotation'
+            },
+            {
+                value: 'annotation,',
+                expectedValue: 'annotation'
+            },
+            {
+                value: 'annotation      ,',
+                expectedValue: 'annotation'
+            },
+            {
+                value: 'annotation      ',
+                expectedValue: 'annotation'
+            }
+            ].forEach(({ value, expectedValue }) => {
+                const parseResult = AnnotationSchema.name.safeParse(value)
+                expect(parseResult.success).toBeTruthy()
+                if (parseResult.success) {
+                    expect(parseResult.data).toStrictEqual(expectedValue)
+                }
+            })
+        })
+
     })
 
     suite('SafeCellAnnotationsSchema', () => {