From 9b1020106d20368519b343d5d1b16b3dc498a487 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 09:38:50 +0000 Subject: [PATCH 1/9] refactor(sess): separate runtime and transport lifecycles --- sess/R/dispatch.R | 8 +- sess/R/hooks.R | 69 ++++++--- sess/R/rstudioapi.R | 2 +- sess/R/runtime.R | 263 ++++++++++++++++++++++++++++++++++ sess/R/server.R | 103 +++++++++---- sess/README.md | 8 +- sess/inst/tinytest/test-ipc.R | 237 +++++++++++++++++++++++++++++- 7 files changed, 629 insertions(+), 61 deletions(-) create mode 100644 sess/R/runtime.R diff --git a/sess/R/dispatch.R b/sess/R/dispatch.R index 1d9e6759..af825d4e 100644 --- a/sess/R/dispatch.R +++ b/sess/R/dispatch.R @@ -20,6 +20,7 @@ ipc_write <- function(data) { }, error = function(e) { warning("[sess] Failed to send IPC message: ", e$message) + .transport_disconnect(silent = TRUE) invisible(FALSE) } ) @@ -49,18 +50,21 @@ rpc_send <- function(method, params = list(), request = FALSE) { msg$id <- req_id } - ipc_write(msg) + sent <- ipc_write(msg) + if (!isTRUE(sent)) return(invisible(FALSE)) if (!request) { invisible(TRUE) } else { # NON-BLOCKING WAIT: # Run later callbacks (which include poll_connection) while waiting for a response. - while (is.null(.sess_env$pending_responses[[req_id]])) { + while (!is.null(.sess_env$con) && is.null(.sess_env$pending_responses[[req_id]])) { later::run_now() Sys.sleep(0.01) } + if (is.null(.sess_env$con)) return(invisible(FALSE)) + response <- .sess_env$pending_responses[[req_id]] .sess_env$pending_responses[[req_id]] <- NULL diff --git a/sess/R/hooks.R b/sess/R/hooks.R index 0cb2dd90..9eb5c0b5 100644 --- a/sess/R/hooks.R +++ b/sess/R/hooks.R @@ -1,13 +1,27 @@ -#' Register hooks for the client IPC +#' Register VS Code runtime integrations #' #' @param use_rstudioapi Logical. Enable rstudioapi emulation. #' @param use_httpgd Logical. Enable httpgd plot device if available. #' @param use_jgd Logical. Enable jgd plot device if available. #' @export register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) { + runtime_start(use_rstudioapi, use_httpgd, use_jgd) +} + +#' Start the VS Code runtime integration (internal) +#' +#' @keywords internal +runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) { + state <- .runtime_state() + if (isTRUE(state$active)) runtime_stop() + state <- .runtime_state() + state$active <- TRUE + completed <- FALSE + on.exit(if (!completed) try(runtime_stop(), silent = TRUE), add = TRUE) + # 1. Override View() to serve table data via paged RPC. if (is.null(.sess_env$dataview_registry)) { - .sess_env$dataview_registry <- new.env(parent = emptyenv()) + .runtime_set_field("dataview_registry", new.env(parent = emptyenv())) } show_dataview <- function(x, title = deparse(substitute(x))) { @@ -58,7 +72,7 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F )) } } - rebind("View", show_dataview, ns = "utils") + .runtime_rebind("View", show_dataview, ns = "utils") # 2. Browser & Webview Options make_viewer <- function(method) { @@ -86,12 +100,10 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F } } - options( - browser = make_viewer("browser"), - viewer = make_viewer("webview"), - page_viewer = make_viewer("page_viewer"), - help_type = "html" - ) + .runtime_set_option("browser", make_viewer("browser")) + .runtime_set_option("viewer", make_viewer("webview")) + .runtime_set_option("page_viewer", make_viewer("page_viewer")) + .runtime_set_option("help_type", "html") # 3. Help System Interception sess_print.help_files_with_topic <- function(x, ...) { @@ -108,7 +120,7 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F } invisible(x) } - registerS3method( + .runtime_register_s3( "print", "help_files_with_topic", sess_print.help_files_with_topic, envir = asNamespace("utils") ) @@ -125,11 +137,11 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F } invisible(x) } - # 4. Plot device: JGD > httpgd > Standard if (use_jgd && nzchar(Sys.getenv("JGD_SOCKET")) && requireNamespace("jgd", quietly = TRUE)) { - options(device = function(...) { + .runtime_set_option("device", function(...) { jgd::jgd() + .runtime_track_device() }) # On reattach (e.g. after a VS Code window reload) the renderer starts a new @@ -145,7 +157,13 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F grDevices::dev.set(devs[names(devs) == "jgd"][[1]]) recorded <- tryCatch(grDevices::recordPlot(), error = function(e) NULL) tryCatch(grDevices::dev.off(), error = function(e) NULL) + before_reopen <- grDevices::dev.list() tryCatch(jgd::jgd(), error = function(e) NULL) + after_reopen <- grDevices::dev.list() + if (!is.null(after_reopen)) { + opened <- if (is.null(before_reopen)) after_reopen else setdiff(after_reopen, before_reopen) + if (length(opened)) .runtime_track_device(opened[[1L]]) + } if (!is.null(recorded)) { tryCatch(grDevices::replayPlot(recorded), error = function(e) NULL) } @@ -153,8 +171,9 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F } reconnect_jgd_device() } else if (use_httpgd && requireNamespace("httpgd", quietly = TRUE)) { - options(device = function(...) { + .runtime_set_option("device", function(...) { httpgd::hgd(silent = TRUE) + .runtime_track_device() notify_client("httpgd", list(url = httpgd::hgd_url())) }) } else { @@ -197,9 +216,10 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F } } - options(device = function(...) { + .runtime_set_option("device", function(...) { grDevices::pdf(NULL, width = 7, height = 7, bg = "white") - options(sess.null_dev = grDevices::dev.cur()) + .runtime_track_device() + .runtime_set_option("sess.null_dev", grDevices::dev.cur()) grDevices::dev.control(displaylist = "enable") }) @@ -215,7 +235,7 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F if (plot_updated || curr_length != last_plot_record_length) { plot_updated <<- FALSE last_plot_record_length <<- curr_length - .sess_env$latest_plot_record <- record + .runtime_set_field("latest_plot_record", record) notify_client("plot_updated") } } @@ -228,18 +248,20 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F TRUE } - setHook("plot.new", new_plot, "replace") - setHook("grid.newpage", new_plot, "replace") + .runtime_set_hook("plot.new", new_plot, "replace") + .runtime_set_hook("grid.newpage", new_plot, "replace") update_plot() - addTaskCallback(update_plot, name = "sess.plot") + .runtime_add_task_callback(update_plot, name = "sess.plot") } # 5. rstudioapi hooks if (use_rstudioapi) { - setHook(packageEvent("rstudioapi", "onLoad"), function(...) { + rstudioapi_hook <- function(...) { patch_rstudioapi() - }, action = "append") + } + .runtime_set_hook(packageEvent("rstudioapi", "onLoad"), + rstudioapi_hook, action = "append") if ("rstudioapi" %in% loadedNamespaces()) { patch_rstudioapi() @@ -249,11 +271,12 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F # 6. Workspace Update Callback # This notifies the client whenever a top-level command is completed, # suggesting that the Global Environment might have changed. - removeTaskCallback("sess.workspace") - addTaskCallback(function(...) { + .runtime_add_task_callback(function(...) { + if (!isTRUE(.runtime_state()$active)) return(FALSE) notify_client("workspace_updated") TRUE }, name = "sess.workspace") + completed <- TRUE invisible(NULL) } diff --git a/sess/R/rstudioapi.R b/sess/R/rstudioapi.R index 5b73e772..89f02563 100644 --- a/sess/R/rstudioapi.R +++ b/sess/R/rstudioapi.R @@ -451,7 +451,7 @@ patch_rstudioapi <- function() { for (name in names(overrides)) { if (exists(name, envir = asNamespace("rstudioapi"), inherits = FALSE)) { - rebind(name, overrides[[name]], "rstudioapi") + .runtime_rebind(name, overrides[[name]], "rstudioapi") } } } diff --git a/sess/R/runtime.R b/sess/R/runtime.R new file mode 100644 index 00000000..3a2413b7 --- /dev/null +++ b/sess/R/runtime.R @@ -0,0 +1,263 @@ +# Runtime state is deliberately independent from the IPC connection state. +.runtime_state <- function() { + if (is.null(.sess_env$runtime)) { + state <- new.env(parent = emptyenv()) + state$active <- FALSE + state$options <- list() + state$bindings <- list() + state$hooks <- list() + state$s3_methods <- list() + state$task_callbacks <- list() + state$devices <- list() + state$fields <- list() + .sess_env$runtime <- state + } + .sess_env$runtime +} + +.runtime_set_field <- function(name, value) { + state <- .runtime_state() + index <- which(vapply(state$fields, function(entry) identical(entry$name, name), logical(1))) + if (!length(index)) { + exists_original <- exists(name, envir = .sess_env, inherits = FALSE) + original <- if (exists_original) get(name, envir = .sess_env, inherits = FALSE) else NULL + state$fields[[length(state$fields) + 1L]] <- list( + name = name, + original_exists = exists_original, + original = original + ) + index <- length(state$fields) + } + assign(name, value, envir = .sess_env) + state$fields[[index[[1L]]]]$installed_exists <- exists(name, envir = .sess_env, inherits = FALSE) + state$fields[[index[[1L]]]]$installed <- if (state$fields[[index[[1L]]]]$installed_exists) { + get(name, envir = .sess_env, inherits = FALSE) + } else { + NULL + } + invisible(value) +} + +.runtime_set_option <- function(name, value) { + state <- .runtime_state() + if (is.null(state$options[[name]])) { + state$options[[name]] <- list(original = getOption(name)) + } + do.call(options, setNames(list(value), name)) + state$options[[name]]$installed <- getOption(name) + invisible(value) +} + +.runtime_assign_binding <- function(sym, value, env) { + locked <- bindingIsLocked(sym, env) + if (locked) unlockBinding(sym, env) + on.exit({ + if (locked && exists(sym, envir = env, inherits = FALSE) && + !bindingIsLocked(sym, env)) lockBinding(sym, env) + }, add = TRUE) + assign(sym, value, envir = env) + invisible(value) +} + +.runtime_rebind <- function(sym, value, ns) { + envs <- if (is.character(ns)) { + namespace <- asNamespace(ns) + attached <- paste0("package:", ns) + if (attached %in% search()) c(list(namespace), list(as.environment(attached))) else list(namespace) + } else if (is.environment(ns)) { + list(ns) + } else { + stop("ns must be a string or environment") + } + + state <- .runtime_state() + for (env in envs) { + if (!exists(sym, envir = env, inherits = FALSE)) next + index <- which(vapply(state$bindings, function(entry) { + identical(entry$env, env) && identical(entry$name, sym) + }, logical(1))) + if (!length(index)) { + state$bindings[[length(state$bindings) + 1L]] <- list( + env = env, + name = sym, + original = get(sym, envir = env, inherits = FALSE), + installed = value, + locked = bindingIsLocked(sym, env) + ) + } else { + state$bindings[[index[[1L]]]]$installed <- value + } + + .runtime_assign_binding(sym, value, env) + } + invisible(value) +} + +.runtime_set_hook <- function(name, value, action = "replace") { + state <- .runtime_state() + current <- getHook(name) + index <- which(vapply(state$hooks, function(entry) identical(entry$name, name), logical(1))) + if (!length(index)) { + state$hooks[[length(state$hooks) + 1L]] <- list(name = name, original = current) + index <- length(state$hooks) + } + setHook(name, value, action = action) + state$hooks[[index[[1L]]]]$installed <- getHook(name) + state$hooks[[index[[1L]]]]$added <- value + invisible(NULL) +} + +.runtime_register_s3 <- function(generic, class, method, envir) { + state <- .runtime_state() + original <- utils::getS3method(generic, class, envir = envir, optional = TRUE) + original_namespace_methods <- if (isNamespace(envir)) { + getNamespaceInfo(envir, "S3methods") + } else { + NULL + } + registerS3method(generic, class, method, envir = envir) + state$s3_methods[[length(state$s3_methods) + 1L]] <- list( + generic = generic, + class = class, + envir = envir, + original = original, + installed = method, + original_namespace_methods = original_namespace_methods, + installed_namespace_methods = if (isNamespace(envir)) getNamespaceInfo(envir, "S3methods") else NULL + ) + invisible(method) +} + +.runtime_add_task_callback <- function(fun, name) { + state <- .runtime_state() + id <- addTaskCallback(fun, name = name) + state$task_callbacks[[length(state$task_callbacks) + 1L]] <- id + invisible(id) +} + +.runtime_track_device <- function(id = grDevices::dev.cur()) { + state <- .runtime_state() + devices <- grDevices::dev.list() + if (is.null(devices)) return(invisible(NULL)) + name <- names(devices)[match(id, devices)] + if (length(name) && !is.na(name)) { + state$devices[[length(state$devices) + 1L]] <- list(id = unname(id), name = name) + } + invisible(id) +} + +.runtime_restore <- function() { + state <- .runtime_state() + + for (id in rev(state$task_callbacks)) { + try(removeTaskCallback(id), silent = TRUE) + } + state$task_callbacks <- list() + + for (entry in rev(state$s3_methods)) { + current <- utils::getS3method(entry$generic, entry$class, + envir = entry$envir, optional = TRUE) + if (identical(current, entry$installed) && !is.null(entry$original)) { + try(registerS3method(entry$generic, entry$class, entry$original, + envir = entry$envir), silent = TRUE) + } else if (identical(current, entry$installed) && is.null(entry$original)) { + generic <- try(get(entry$generic, envir = entry$envir), silent = TRUE) + dispatch_env <- if (!inherits(generic, "try-error") && is.function(generic) && + !is.null(environment(generic))) { + environment(generic) + } else { + asNamespace("base") + } + table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) + method_name <- paste(entry$generic, entry$class, sep = ".") + if (is.environment(table) && exists(method_name, envir = table, inherits = FALSE) && + identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { + rm(list = method_name, envir = table) + } + if (isNamespace(entry$envir) && + identical(getNamespaceInfo(entry$envir, "S3methods"), entry$installed_namespace_methods)) { + try(setNamespaceInfo(entry$envir, "S3methods", entry$original_namespace_methods), silent = TRUE) + } + } + } + state$s3_methods <- list() + + for (entry in rev(state$hooks)) { + current <- getHook(entry$name) + if (identical(current, entry$installed)) { + try(setHook(entry$name, entry$original, action = "replace"), silent = TRUE) + } else if (!is.null(entry$added)) { + # Keep hooks installed by other code while removing only our own callback. + current <- Filter(function(hook) !identical(hook, entry$added), current) + try(setHook(entry$name, current, action = "replace"), silent = TRUE) + } + } + state$hooks <- list() + + for (entry in rev(state$bindings)) { + env <- entry$env + name <- entry$name + if (exists(name, envir = env, inherits = FALSE) && + identical(get(name, envir = env, inherits = FALSE), entry$installed)) { + try(.runtime_assign_binding(name, entry$original, env), silent = TRUE) + } + } + state$bindings <- list() + + # Devices opened through sess's device factory are closed so plotting resumes + # on the device that was active before the factory was used. + original_device <- grDevices::dev.cur() + for (entry in rev(state$devices)) { + devices <- grDevices::dev.list() + index <- if (is.null(devices)) NA_integer_ else match(entry$id, devices) + if (!is.na(index) && identical(names(devices)[[index]], entry$name)) { + try(grDevices::dev.off(which = entry$id), silent = TRUE) + } + } + devices <- grDevices::dev.list() + if (!is.null(devices) && original_device %in% devices) { + try(grDevices::dev.set(original_device), silent = TRUE) + } + state$devices <- list() + + for (entry in rev(state$fields)) { + exists_current <- exists(entry$name, envir = .sess_env, inherits = FALSE) + current <- if (exists_current) get(entry$name, envir = .sess_env, inherits = FALSE) else NULL + if (identical(exists_current, entry$installed_exists) && + (!exists_current || identical(current, entry$installed))) { + if (entry$original_exists) { + assign(entry$name, entry$original, envir = .sess_env) + } else if (exists_current) { + rm(list = entry$name, envir = .sess_env) + } + } + } + state$fields <- list() + + for (name in names(state$options)) { + entry <- state$options[[name]] + if (identical(getOption(name), entry$installed)) { + try(do.call(options, setNames(list(entry$original), name)), silent = TRUE) + } + } + state$options <- list() + state$active <- FALSE + invisible(NULL) +} + +#' Stop the VS Code runtime integration (internal) +#' +#' Removes runtime callbacks and restores R state installed by runtime_start(). +#' @keywords internal +runtime_stop <- function() { + state <- .runtime_state() + if (!isTRUE(state$active) && + !length(state$options) && !length(state$bindings) && !length(state$hooks) && + !length(state$s3_methods) && !length(state$task_callbacks) && + !length(state$devices) && !length(state$fields)) { + return(invisible(NULL)) + } + # Prevent callbacks from sending new notifications during cleanup. + state$active <- FALSE + .runtime_restore() +} diff --git a/sess/R/server.R b/sess/R/server.R index fe5c8eba..608e1ddb 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -8,6 +8,8 @@ #' @param use_jgd Logical. Use jgd for plotting if available. Defaults to FALSE. #' @export connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) { + # Invalidate poll callbacks and restore a previous runtime before reconnecting. + .transport_disconnect(silent = TRUE) .sess_env$con <- NULL .sess_env$pending_responses <- list() .sess_env$read_buffer <- "" @@ -67,7 +69,7 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, NULL } ) - if (is.null(con)) return() + if (is.null(con)) return(FALSE) .sess_env$con <- con @@ -84,19 +86,43 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, ) )) + if (is.null(.sess_env$con)) return(FALSE) + print_async_msg("[sess] Connected to VS Code") # Start the polling loop - poll_connection() + poll_connection(.sess_env$transport_generation) + TRUE } - do_connect() + connected <- do_connect() if (is.na(use_rstudioapi)) use_rstudioapi <- TRUE if (is.na(use_httpgd)) use_httpgd <- TRUE if (is.na(use_jgd)) use_jgd <- FALSE - register_hooks(use_rstudioapi = use_rstudioapi, use_httpgd = use_httpgd, use_jgd = use_jgd) + if (isTRUE(connected) && !is.null(.sess_env$con)) { + runtime_start(use_rstudioapi = use_rstudioapi, + use_httpgd = use_httpgd, + use_jgd = use_jgd) + } + + invisible(NULL) +} + +.transport_disconnect <- function(silent = FALSE) { + con <- .sess_env$con + .sess_env$con <- NULL + .sess_env$transport_generation <- if (is.null(.sess_env$transport_generation)) { + 1L + } else { + .sess_env$transport_generation + 1L + } + .sess_env$read_buffer <- "" + .sess_env$pending_responses <- list() + if (!is.null(con)) try(close(con), silent = TRUE) + runtime_stop() + if (!silent && !is.null(con)) message("[sess] Disconnected from VS Code") invisible(NULL) } @@ -104,52 +130,67 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, #' #' Runs as a recurring later callback; dispatches NDJSON messages from vscode. #' @keywords internal -poll_connection <- function() { +poll_connection <- function(generation = .sess_env$transport_generation) { con <- .sess_env$con - if (is.null(con)) return() + if (is.null(con) || !identical(generation, .sess_env$transport_generation)) return() # Non-blocking poll: 0 ms timeout ready <- tryCatch( processx::poll(list(con), 0L), - error = function(e) NULL + error = function(e) { + .transport_disconnect(silent = TRUE) + NULL + } ) + if (is.null(ready) || is.null(.sess_env$con) || + !identical(generation, .sess_env$transport_generation)) return() + if (!is.null(ready) && length(ready) > 0 && identical(ready[[1]], "ready")) { chunk <- tryCatch( processx::conn_read_chars(con), error = function(e) { - .sess_env$con <- NULL + .transport_disconnect(silent = TRUE) NULL } ) - if (!is.null(chunk) && nzchar(chunk)) { - .sess_env$read_buffer <- paste0(.sess_env$read_buffer, chunk) - parts <- strsplit(.sess_env$read_buffer, "\n", fixed = TRUE)[[1]] - - n <- length(parts) - # Keep any trailing partial line in the buffer - if (endsWith(.sess_env$read_buffer, "\n")) { - .sess_env$read_buffer <- "" - } else { - .sess_env$read_buffer <- parts[n] - parts <- parts[-n] - } + if (is.null(.sess_env$con)) return() + if (is.null(chunk) || !nzchar(chunk)) { + .transport_disconnect(silent = TRUE) + return() + } - for (line in parts) { - line <- trimws(line) - if (!nzchar(line)) next - tryCatch( - dispatch_message(line), - error = function(e) { - warning("[sess] Error dispatching message: ", e$message) - } - ) - } + .sess_env$read_buffer <- paste0(.sess_env$read_buffer, chunk) + parts <- strsplit(.sess_env$read_buffer, "\n", fixed = TRUE)[[1]] + + n <- length(parts) + # Keep any trailing partial line in the buffer + if (endsWith(.sess_env$read_buffer, "\n")) { + .sess_env$read_buffer <- "" + } else { + .sess_env$read_buffer <- parts[n] + parts <- parts[-n] } + + for (line in parts) { + line <- trimws(line) + if (!nzchar(line)) next + tryCatch( + dispatch_message(line), + error = function(e) { + warning("[sess] Error dispatching message: ", e$message) + } + ) + } + } else if (length(ready) > 0 && ready[[1]] %in% c("closed", "error")) { + .transport_disconnect(silent = TRUE) + return() } - later::later(poll_connection, 0.01) + if (!is.null(.sess_env$con) && identical(generation, .sess_env$transport_generation)) { + later::later(function() poll_connection(generation), 0.01) + } } #' Dispatch a single NDJSON line as a JSON-RPC message (internal) diff --git a/sess/README.md b/sess/README.md index 60ca932e..f9a64165 100644 --- a/sess/README.md +++ b/sess/README.md @@ -223,7 +223,13 @@ Response example: ## 7. Hook Registration and Options -`connect()` initializes runtime hooks via `register_hooks()`. +After the IPC connection succeeds, `connect()` starts the same VS Code runtime +integration managed by `register_hooks()`. If polling or writing detects that the +transport has closed, the runtime is stopped automatically. Runtime shutdown +removes its task callbacks and restores the options, bindings, S3 method, and +plot hooks it installed when their current values have not been changed by +other code. Calling `register_hooks()` again replaces its previous runtime +installation instead of accumulating callbacks. Intercepted features include: diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index 1482e106..9cda1fb4 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -147,9 +147,122 @@ local({ expect_equal(sorted$rows[[3]][["1"]], "10") }) -# NDJSON framing round-trips correctly through a socket pair. -# Kept last: tinytest runs files as flat scripts, so an unrecoverable socket -# error here must not mask the blocks above. Socket support is +# Runtime startup and shutdown are reversible and idempotent. +local({ + .sess_env <- sess:::.sess_env + old_plot_path <- .sess_env$latest_plot_path + .sess_env$latest_plot_path <- tempfile(fileext = ".png") + on.exit({ + sess:::runtime_stop() + unlink(.sess_env$latest_plot_path) + .sess_env$latest_plot_path <- old_plot_path + }, add = TRUE) + + utils_ns <- asNamespace("utils") + old_view <- get("View", utils_ns, inherits = FALSE) + old_options <- lapply(c("browser", "viewer", "page_viewer", "help_type", "device"), getOption) + names(old_options) <- c("browser", "viewer", "page_viewer", "help_type", "device") + old_plot_hook <- getHook("plot.new") + old_grid_hook <- getHook("grid.newpage") + old_help_method <- utils::getS3method("print", "help_files_with_topic", + envir = utils_ns) + + sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + expect_true(isTRUE(sess:::.runtime_state()$active)) + expect_false(identical(get("View", utils_ns, inherits = FALSE), old_view)) + expect_true(is.function(getOption("viewer"))) + expect_false(identical(utils::getS3method("print", "help_files_with_topic", + envir = utils_ns), old_help_method)) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) + expect_false(identical(getHook("plot.new"), old_plot_hook)) + expect_false(identical(getHook("grid.newpage"), old_grid_hook)) + + callbacks_after_first_start <- getTaskCallbackNames() + sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) + expect_equal(length(grep("^sess.plot$", getTaskCallbackNames())), + length(grep("^sess.plot$", callbacks_after_first_start))) + + sess:::runtime_stop() + expect_false(isTRUE(sess:::.runtime_state()$active)) + expect_identical(get("View", utils_ns, inherits = FALSE), old_view) + expect_identical(getOption("browser"), old_options$browser) + expect_identical(getOption("viewer"), old_options$viewer) + expect_identical(getOption("page_viewer"), old_options$page_viewer) + expect_identical(getOption("help_type"), old_options$help_type) + expect_identical(getOption("device"), old_options$device) + expect_identical(getHook("plot.new"), old_plot_hook) + expect_identical(getHook("grid.newpage"), old_grid_hook) + expect_identical(utils::getS3method("print", "help_files_with_topic", + envir = utils_ns), old_help_method) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) + expect_equal(length(grep("^sess.plot$", getTaskCallbackNames())), 0L) +}) + +# Cleanup preserves options and bindings changed by user code after startup. +local({ + .sess_env <- sess:::.sess_env + old_plot_path <- .sess_env$latest_plot_path + .sess_env$latest_plot_path <- tempfile(fileext = ".png") + utils_ns <- asNamespace("utils") + original_view <- get("View", utils_ns, inherits = FALSE) + binding_was_locked <- bindingIsLocked("View", utils_ns) + original_viewer <- getOption("viewer") + user_view <- function(...) "user view" + user_viewer <- function(...) "user viewer" + on.exit({ + sess:::runtime_stop() + if (binding_was_locked) unlockBinding("View", utils_ns) + assign("View", original_view, envir = utils_ns) + if (binding_was_locked) lockBinding("View", utils_ns) + options(viewer = original_viewer) + unlink(.sess_env$latest_plot_path) + .sess_env$latest_plot_path <- old_plot_path + }, add = TRUE) + + sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + options(viewer = user_viewer) + if (binding_was_locked) unlockBinding("View", utils_ns) + assign("View", user_view, envir = utils_ns) + if (binding_was_locked) lockBinding("View", utils_ns) + sess:::runtime_stop() + + expect_identical(getOption("viewer"), user_viewer) + expect_identical(get("View", utils_ns, inherits = FALSE), user_view) +}) + +# rstudioapi overrides and its package-load hook are restored at runtime stop. +local({ + if (!requireNamespace("rstudioapi", quietly = TRUE)) return(invisible(NULL)) + rstudioapi_ns <- asNamespace("rstudioapi") + if (!exists("isAvailable", envir = rstudioapi_ns, inherits = FALSE)) { + return(invisible(NULL)) + } + + .sess_env <- sess:::.sess_env + old_plot_path <- .sess_env$latest_plot_path + .sess_env$latest_plot_path <- tempfile(fileext = ".png") + original_is_available <- get("isAvailable", rstudioapi_ns, inherits = FALSE) + rstudioapi_hook_name <- packageEvent("rstudioapi", "onLoad") + original_load_hook <- getHook(rstudioapi_hook_name) + on.exit({ + sess:::runtime_stop() + unlink(.sess_env$latest_plot_path) + .sess_env$latest_plot_path <- old_plot_path + }, add = TRUE) + + sess:::runtime_start(use_rstudioapi = TRUE, use_httpgd = FALSE, use_jgd = FALSE) + expect_false(identical(get("isAvailable", rstudioapi_ns, inherits = FALSE), + original_is_available)) + expect_false(identical(getHook(rstudioapi_hook_name), original_load_hook)) + + sess:::runtime_stop() + expect_identical(get("isAvailable", rstudioapi_ns, inherits = FALSE), + original_is_available) + expect_identical(getHook(rstudioapi_hook_name), original_load_hook) +}) + +# NDJSON framing round-trips correctly through a socket pair. Socket support is # environment-sensitive (some processx builds/platforms fail to accept or read # the loopback connection), so any infrastructure error becomes a silent skip # rather than a failure. A genuine framing/protocol bug yields wrong captured @@ -201,3 +314,121 @@ local({ expect_equal(res$method, "ping") expect_equal(res$value, 42L) }) + +# A peer disappearing while a request is waiting stops the runtime, and a new +# connection can start a fresh runtime without duplicating callbacks. +local({ + .sess_env <- sess:::.sess_env + if (!requireNamespace("processx", quietly = TRUE) || .Platform$OS.type == "windows") { + return(invisible(NULL)) + } + + listener <- function() { + path <- tempfile(fileext = ".sock") + server <- tryCatch(processx::conn_create_unix_socket(path, encoding = ""), + error = function(e) NULL) + if (is.null(server)) return(NULL) + list(path = path, server = server) + } + accept_peer <- function(server) { + ready <- tryCatch(processx::poll(list(server), 1000L), error = function(e) NULL) + if (is.null(ready) || !ready[[1]] %in% c("connect", "ready")) return(NULL) + tryCatch(processx::conn_accept_unix_socket(server), error = function(e) NULL) + } + + first <- listener() + if (is.null(first)) return(invisible(NULL)) + utils_ns <- asNamespace("utils") + original_view <- get("View", utils_ns, inherits = FALSE) + original_options <- lapply(c("browser", "viewer", "page_viewer", "help_type", "device"), getOption) + names(original_options) <- c("browser", "viewer", "page_viewer", "help_type", "device") + original_help_method <- utils::getS3method("print", "help_files_with_topic", + envir = utils_ns) + original_plot_hook <- getHook("plot.new") + original_grid_hook <- getHook("grid.newpage") + second <- NULL + first_peer <- NULL + second_peer <- NULL + on.exit({ + sess:::.transport_disconnect() + for (con in list(first_peer, second_peer, first$server, + if (!is.null(second)) second$server else NULL)) { + if (!is.null(con)) try(close(con), silent = TRUE) + } + unlink(c(first$path, if (!is.null(second)) second$path else character())) + }, add = TRUE) + + connected <- tryCatch({ + sess::connect(first$path, use_rstudioapi = FALSE, + use_httpgd = FALSE, use_jgd = FALSE) + first_peer <- accept_peer(first$server) + !is.null(first_peer) && !is.null(.sess_env$con) + }, error = function(e) FALSE) + if (!isTRUE(connected)) return(invisible(NULL)) + + close(first_peer) + result <- suppressWarnings(sess::request_client("test/disconnect_wait")) + expect_false(isTRUE(result)) + expect_null(.sess_env$con) + expect_false(isTRUE(sess:::.runtime_state()$active)) + expect_identical(get("View", utils_ns, inherits = FALSE), original_view) + expect_identical(getOption("browser"), original_options$browser) + expect_identical(getOption("viewer"), original_options$viewer) + expect_identical(getOption("page_viewer"), original_options$page_viewer) + expect_identical(getOption("help_type"), original_options$help_type) + expect_identical(getOption("device"), original_options$device) + expect_identical(utils::getS3method("print", "help_files_with_topic", + envir = utils_ns), original_help_method) + expect_identical(getHook("plot.new"), original_plot_hook) + expect_identical(getHook("grid.newpage"), original_grid_hook) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) + expect_equal(length(grep("^sess.plot$", getTaskCallbackNames())), 0L) + + second <- listener() + if (is.null(second)) return(invisible(NULL)) + sess::connect(second$path, use_rstudioapi = FALSE, + use_httpgd = FALSE, use_jgd = FALSE) + second_peer <- accept_peer(second$server) + if (is.null(second_peer) || is.null(.sess_env$con)) return(invisible(NULL)) + expect_true(isTRUE(sess:::.runtime_state()$active)) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) + sess:::.transport_disconnect() + expect_null(.sess_env$con) + expect_false(isTRUE(sess:::.runtime_state()$active)) +}) + +# A transport write failure also stops the runtime promptly. +local({ + .sess_env <- sess:::.sess_env + if (!requireNamespace("processx", quietly = TRUE) || .Platform$OS.type == "windows") { + return(invisible(NULL)) + } + + path <- tempfile(fileext = ".sock") + server <- tryCatch(processx::conn_create_unix_socket(path, encoding = ""), + error = function(e) NULL) + if (is.null(server)) return(invisible(NULL)) + peer <- NULL + on.exit({ + sess:::.transport_disconnect() + if (!is.null(peer)) try(close(peer), silent = TRUE) + try(close(server), silent = TRUE) + unlink(path) + }, add = TRUE) + + connected <- tryCatch({ + sess::connect(path, use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + ready <- processx::poll(list(server), 1000L) + if (ready[[1]] %in% c("connect", "ready")) { + peer <- processx::conn_accept_unix_socket(server) + } + !is.null(peer) && !is.null(.sess_env$con) + }, error = function(e) FALSE) + if (!isTRUE(connected)) return(invisible(NULL)) + + close(.sess_env$con) + sent <- suppressWarnings(sess:::ipc_write(list(method = "test/write_failure"))) + expect_false(isTRUE(sent)) + expect_null(.sess_env$con) + expect_false(isTRUE(sess:::.runtime_state()$active)) +}) From 2c6466d57c5340c8a191a6b8113cc4704d7bf524 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 10:03:20 +0000 Subject: [PATCH 2/9] fix(sess): keep polling after transient empty reads --- .github/workflows/main.yml | 3 ++ sess/DESCRIPTION | 2 +- sess/R/runtime.R | 58 ++++++++++++++++++------------ sess/R/server.R | 66 ++++++++++++++++++++++------------- sess/README.md | 6 ++-- sess/inst/tinytest/test-ipc.R | 65 ++++++++++++++++++++++++++++++++-- sess/man/poll_connection.Rd | 2 +- sess/man/register_hooks.Rd | 4 +-- sess/man/runtime_start.Rd | 12 +++++++ sess/man/runtime_stop.Rd | 12 +++++++ 10 files changed, 174 insertions(+), 56 deletions(-) create mode 100644 sess/man/runtime_start.Rd create mode 100644 sess/man/runtime_stop.Rd diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index fefdb4ad..ecb59c9f 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -32,6 +32,9 @@ jobs: run: install.packages("remotes") shell: Rscript {0} - run: npm run build + - name: Run sess tests + run: tinytest::test_package("sess") + shell: Rscript {0} - name: Run tests (Linux) if: runner.os == 'Linux' run: xvfb-run -a npm run test diff --git a/sess/DESCRIPTION b/sess/DESCRIPTION index cc022f52..aba1828f 100644 --- a/sess/DESCRIPTION +++ b/sess/DESCRIPTION @@ -20,4 +20,4 @@ Suggests: jgd, svglite, tinytest -Config/roxygen2/version: 8.0.0 +Config/roxygen2/version: 8.1.0 diff --git a/sess/R/runtime.R b/sess/R/runtime.R index 3a2413b7..eddb81ae 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -53,7 +53,9 @@ if (locked) unlockBinding(sym, env) on.exit({ if (locked && exists(sym, envir = env, inherits = FALSE) && - !bindingIsLocked(sym, env)) lockBinding(sym, env) + !bindingIsLocked(sym, env)) { + lockBinding(sym, env) + } }, add = TRUE) assign(sym, value, envir = env) invisible(value) @@ -63,7 +65,11 @@ envs <- if (is.character(ns)) { namespace <- asNamespace(ns) attached <- paste0("package:", ns) - if (attached %in% search()) c(list(namespace), list(as.environment(attached))) else list(namespace) + if (attached %in% search()) { + c(list(namespace), list(as.environment(attached))) + } else { + list(namespace) + } } else if (is.environment(ns)) { list(ns) } else { @@ -123,16 +129,20 @@ original = original, installed = method, original_namespace_methods = original_namespace_methods, - installed_namespace_methods = if (isNamespace(envir)) getNamespaceInfo(envir, "S3methods") else NULL + installed_namespace_methods = if (isNamespace(envir)) { + getNamespaceInfo(envir, "S3methods") + } else { + NULL + } ) invisible(method) } .runtime_add_task_callback <- function(fun, name) { state <- .runtime_state() - id <- addTaskCallback(fun, name = name) - state$task_callbacks[[length(state$task_callbacks) + 1L]] <- id - invisible(id) + addTaskCallback(fun, name = name) + state$task_callbacks[[length(state$task_callbacks) + 1L]] <- name + invisible(name) } .runtime_track_device <- function(id = grDevices::dev.cur()) { @@ -141,7 +151,10 @@ if (is.null(devices)) return(invisible(NULL)) name <- names(devices)[match(id, devices)] if (length(name) && !is.na(name)) { - state$devices[[length(state$devices) + 1L]] <- list(id = unname(id), name = name) + state$devices[[length(state$devices) + 1L]] <- list( + id = unname(id), + name = name + ) } invisible(id) } @@ -149,8 +162,8 @@ .runtime_restore <- function() { state <- .runtime_state() - for (id in rev(state$task_callbacks)) { - try(removeTaskCallback(id), silent = TRUE) + for (name in rev(state$task_callbacks)) { + try(removeTaskCallback(name), silent = TRUE) } state$task_callbacks <- list() @@ -162,7 +175,8 @@ envir = entry$envir), silent = TRUE) } else if (identical(current, entry$installed) && is.null(entry$original)) { generic <- try(get(entry$generic, envir = entry$envir), silent = TRUE) - dispatch_env <- if (!inherits(generic, "try-error") && is.function(generic) && + dispatch_env <- if (!inherits(generic, "try-error") && + is.function(generic) && !is.null(environment(generic))) { environment(generic) } else { @@ -170,13 +184,18 @@ } table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) method_name <- paste(entry$generic, entry$class, sep = ".") - if (is.environment(table) && exists(method_name, envir = table, inherits = FALSE) && + if (is.environment(table) && + exists(method_name, envir = table, inherits = FALSE) && identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { rm(list = method_name, envir = table) } if (isNamespace(entry$envir) && - identical(getNamespaceInfo(entry$envir, "S3methods"), entry$installed_namespace_methods)) { - try(setNamespaceInfo(entry$envir, "S3methods", entry$original_namespace_methods), silent = TRUE) + identical(getNamespaceInfo(entry$envir, "S3methods"), + entry$installed_namespace_methods)) { + try( + setNamespaceInfo(entry$envir, "S3methods", entry$original_namespace_methods), + silent = TRUE + ) } } } @@ -204,9 +223,8 @@ } state$bindings <- list() - # Devices opened through sess's device factory are closed so plotting resumes - # on the device that was active before the factory was used. - original_device <- grDevices::dev.cur() + # Close devices opened through sess's device factory so plotting resumes on + # the device R selects after the runtime-owned device is removed. for (entry in rev(state$devices)) { devices <- grDevices::dev.list() index <- if (is.null(devices)) NA_integer_ else match(entry$id, devices) @@ -214,10 +232,6 @@ try(grDevices::dev.off(which = entry$id), silent = TRUE) } } - devices <- grDevices::dev.list() - if (!is.null(devices) && original_device %in% devices) { - try(grDevices::dev.set(original_device), silent = TRUE) - } state$devices <- list() for (entry in rev(state$fields)) { @@ -251,8 +265,8 @@ #' @keywords internal runtime_stop <- function() { state <- .runtime_state() - if (!isTRUE(state$active) && - !length(state$options) && !length(state$bindings) && !length(state$hooks) && + if (!isTRUE(state$active) && !length(state$options) && + !length(state$bindings) && !length(state$hooks) && !length(state$s3_methods) && !length(state$task_callbacks) && !length(state$devices) && !length(state$fields)) { return(invisible(NULL)) diff --git a/sess/R/server.R b/sess/R/server.R index 608e1ddb..4ea2efc9 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -126,6 +126,17 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, invisible(NULL) } +.transport_empty_read_is_eof <- function(con) { + # A ready poll can race with a read that finds no bytes. Treat only a + # connection known to have reached EOF as disconnected; processx documents + # conn_is_incomplete() as FALSE once no more data can arrive. + incomplete <- tryCatch( + processx::conn_is_incomplete(con), + error = function(e) TRUE + ) + identical(incomplete, FALSE) +} + #' Poll the IPC connection for incoming messages (internal) #' #' Runs as a recurring later callback; dispatches NDJSON messages from vscode. @@ -144,7 +155,9 @@ poll_connection <- function(generation = .sess_env$transport_generation) { ) if (is.null(ready) || is.null(.sess_env$con) || - !identical(generation, .sess_env$transport_generation)) return() + !identical(generation, .sess_env$transport_generation)) { + return() + } if (!is.null(ready) && length(ready) > 0 && identical(ready[[1]], "ready")) { chunk <- tryCatch( @@ -156,32 +169,35 @@ poll_connection <- function(generation = .sess_env$transport_generation) { ) if (is.null(.sess_env$con)) return() - if (is.null(chunk) || !nzchar(chunk)) { - .transport_disconnect(silent = TRUE) - return() - } - - .sess_env$read_buffer <- paste0(.sess_env$read_buffer, chunk) - parts <- strsplit(.sess_env$read_buffer, "\n", fixed = TRUE)[[1]] - - n <- length(parts) - # Keep any trailing partial line in the buffer - if (endsWith(.sess_env$read_buffer, "\n")) { - .sess_env$read_buffer <- "" + has_data <- !is.null(chunk) && length(chunk) > 0L && any(nzchar(chunk)) + if (!has_data) { + if (.transport_empty_read_is_eof(con)) { + .transport_disconnect(silent = TRUE) + return() + } } else { - .sess_env$read_buffer <- parts[n] - parts <- parts[-n] - } + .sess_env$read_buffer <- paste0(.sess_env$read_buffer, paste0(chunk, collapse = "")) + parts <- strsplit(.sess_env$read_buffer, "\n", fixed = TRUE)[[1]] + + n <- length(parts) + # Keep any trailing partial line in the buffer + if (endsWith(.sess_env$read_buffer, "\n")) { + .sess_env$read_buffer <- "" + } else { + .sess_env$read_buffer <- parts[n] + parts <- parts[-n] + } - for (line in parts) { - line <- trimws(line) - if (!nzchar(line)) next - tryCatch( - dispatch_message(line), - error = function(e) { - warning("[sess] Error dispatching message: ", e$message) - } - ) + for (line in parts) { + line <- trimws(line) + if (!nzchar(line)) next + tryCatch( + dispatch_message(line), + error = function(e) { + warning("[sess] Error dispatching message: ", e$message) + } + ) + } } } else if (length(ready) > 0 && ready[[1]] %in% c("closed", "error")) { .transport_disconnect(silent = TRUE) diff --git a/sess/README.md b/sess/README.md index f9a64165..788af751 100644 --- a/sess/README.md +++ b/sess/README.md @@ -228,8 +228,10 @@ integration managed by `register_hooks()`. If polling or writing detects that th transport has closed, the runtime is stopped automatically. Runtime shutdown removes its task callbacks and restores the options, bindings, S3 method, and plot hooks it installed when their current values have not been changed by -other code. Calling `register_hooks()` again replaces its previous runtime -installation instead of accumulating callbacks. +other code. Runtime-owned graphics devices are also closed so later plotting no +longer targets a stale VS Code connection. A plot held only by a `jgd` device +may not survive a disconnect/reload. Calling `register_hooks()` again replaces +its previous runtime installation instead of accumulating callbacks. Intercepted features include: diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index 9cda1fb4..a75b186e 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -177,6 +177,10 @@ local({ expect_false(identical(getHook("plot.new"), old_plot_hook)) expect_false(identical(getHook("grid.newpage"), old_grid_hook)) + grDevices::pdf(NULL) + sess:::.runtime_track_device() + runtime_device <- grDevices::dev.cur() + callbacks_after_first_start <- getTaskCallbackNames() sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) @@ -193,6 +197,7 @@ local({ expect_identical(getOption("device"), old_options$device) expect_identical(getHook("plot.new"), old_plot_hook) expect_identical(getHook("grid.newpage"), old_grid_hook) + expect_false(runtime_device %in% grDevices::dev.list()) expect_identical(utils::getS3method("print", "help_files_with_topic", envir = utils_ns), old_help_method) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) @@ -262,6 +267,59 @@ local({ expect_identical(getHook(rstudioapi_hook_name), original_load_hook) }) +# An empty ready-read is EOF only after processx confirms no more data can +# arrive. Errors from the EOF check are treated conservatively as still open. +local({ + if (!requireNamespace("processx", quietly = TRUE)) return(invisible(NULL)) + cons <- tryCatch(processx::conn_create_pipepair(), error = function(e) NULL) + if (is.null(cons)) return(invisible(NULL)) + on.exit({ + try(close(cons[[1L]]), silent = TRUE) + try(close(cons[[2L]]), silent = TRUE) + }, add = TRUE) + + expect_false(sess:::.transport_empty_read_is_eof(cons[[2L]])) + expect_false(sess:::.transport_empty_read_is_eof(NULL)) + close(cons[[1L]]) + chunk <- processx::conn_read_chars(cons[[2L]]) + expect_true(is.null(chunk) || !length(chunk) || !any(nzchar(chunk))) + expect_true(sess:::.transport_empty_read_is_eof(cons[[2L]])) +}) + +# EOF observed by the polling loop stops the runtime and releases the transport +# on every platform without depending on a Unix-domain socket. +local({ + if (!requireNamespace("processx", quietly = TRUE)) return(invisible(NULL)) + cons <- tryCatch(processx::conn_create_pipepair(), error = function(e) NULL) + if (is.null(cons)) return(invisible(NULL)) + + .sess_env <- sess:::.sess_env + old_plot_path <- .sess_env$latest_plot_path + .sess_env$latest_plot_path <- tempfile(fileext = ".png") + on.exit({ + sess:::.transport_disconnect(silent = TRUE) + try(close(cons[[1L]]), silent = TRUE) + try(close(cons[[2L]]), silent = TRUE) + unlink(.sess_env$latest_plot_path) + .sess_env$latest_plot_path <- old_plot_path + }, add = TRUE) + + .sess_env$con <- cons[[2L]] + .sess_env$transport_generation <- if (is.null(.sess_env$transport_generation)) { + 1L + } else { + .sess_env$transport_generation + 1L + } + sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + expect_true(isTRUE(sess:::.runtime_state()$active)) + + close(cons[[1L]]) + sess:::poll_connection(.sess_env$transport_generation) + expect_null(.sess_env$con) + expect_false(isTRUE(sess:::.runtime_state()$active)) + expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) +}) + # NDJSON framing round-trips correctly through a socket pair. Socket support is # environment-sensitive (some processx builds/platforms fail to accept or read # the loopback connection), so any infrastructure error becomes a silent skip @@ -272,7 +330,7 @@ local({ local({ if (!requireNamespace("processx", quietly = TRUE) || .Platform$OS.type == "windows") { - # Windows named pipe paths are tested separately. + # This check specifically exercises Unix-domain socket framing. return(invisible(NULL)) } @@ -340,8 +398,9 @@ local({ if (is.null(first)) return(invisible(NULL)) utils_ns <- asNamespace("utils") original_view <- get("View", utils_ns, inherits = FALSE) - original_options <- lapply(c("browser", "viewer", "page_viewer", "help_type", "device"), getOption) - names(original_options) <- c("browser", "viewer", "page_viewer", "help_type", "device") + option_names <- c("browser", "viewer", "page_viewer", "help_type", "device") + original_options <- lapply(option_names, getOption) + names(original_options) <- option_names original_help_method <- utils::getS3method("print", "help_files_with_topic", envir = utils_ns) original_plot_hook <- getHook("plot.new") diff --git a/sess/man/poll_connection.Rd b/sess/man/poll_connection.Rd index cd257fc3..4e2c9a7b 100644 --- a/sess/man/poll_connection.Rd +++ b/sess/man/poll_connection.Rd @@ -4,7 +4,7 @@ \alias{poll_connection} \title{Poll the IPC connection for incoming messages (internal)} \usage{ -poll_connection() +poll_connection(generation = .sess_env$transport_generation) } \description{ Runs as a recurring later callback; dispatches NDJSON messages from vscode. diff --git a/sess/man/register_hooks.Rd b/sess/man/register_hooks.Rd index 26235125..8fd0882b 100644 --- a/sess/man/register_hooks.Rd +++ b/sess/man/register_hooks.Rd @@ -2,7 +2,7 @@ % Please edit documentation in R/hooks.R \name{register_hooks} \alias{register_hooks} -\title{Register hooks for the client IPC} +\title{Register VS Code runtime integrations} \usage{ register_hooks(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) } @@ -14,5 +14,5 @@ register_hooks(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) \item{use_jgd}{Logical. Enable jgd plot device if available.} } \description{ -Register hooks for the client IPC +Register VS Code runtime integrations } diff --git a/sess/man/runtime_start.Rd b/sess/man/runtime_start.Rd new file mode 100644 index 00000000..37c1172a --- /dev/null +++ b/sess/man/runtime_start.Rd @@ -0,0 +1,12 @@ +% Generated by roxygen2: do not edit by hand +% Please edit documentation in R/hooks.R +\name{runtime_start} +\alias{runtime_start} +\title{Start the VS Code runtime integration (internal)} +\usage{ +runtime_start(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) +} +\description{ +Start the VS Code runtime integration (internal) +} +\keyword{internal} diff --git a/sess/man/runtime_stop.Rd b/sess/man/runtime_stop.Rd new file mode 100644 index 00000000..f16d8eb1 --- /dev/null +++ b/sess/man/runtime_stop.Rd @@ -0,0 +1,12 @@ +% Generated by roxygen2: do not edit by hand +% Please edit documentation in R/runtime.R +\name{runtime_stop} +\alias{runtime_stop} +\title{Stop the VS Code runtime integration (internal)} +\usage{ +runtime_stop() +} +\description{ +Removes runtime callbacks and restores R state installed by runtime_start(). +} +\keyword{internal} From 250587e58bb30fe31f2ec2b09bb957e52c22a234 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 10:17:27 +0000 Subject: [PATCH 3/9] fix(sess): clear viewer state on disconnect --- sess/R/runtime.R | 30 ++++++++++++------- sess/R/server.R | 2 +- sess/inst/tinytest/test-ipc.R | 54 +++++++++++++++++++++++++++++------ 3 files changed, 66 insertions(+), 20 deletions(-) diff --git a/sess/R/runtime.R b/sess/R/runtime.R index eddb81ae..16cc2a29 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -15,6 +15,14 @@ .sess_env$runtime } +.runtime_clear_viewer_state <- function() { + # Open dataviews and title-to-id mappings belong to one runtime connection. + # Keep an empty registry available, but never restore data from a prior run. + .sess_env$dataviews <- list() + .sess_env$dataview_registry <- new.env(parent = emptyenv()) + invisible(NULL) +} + .runtime_set_field <- function(name, value) { state <- .runtime_state() index <- which(vapply(state$fields, function(entry) identical(entry$name, name), logical(1))) @@ -53,7 +61,7 @@ if (locked) unlockBinding(sym, env) on.exit({ if (locked && exists(sym, envir = env, inherits = FALSE) && - !bindingIsLocked(sym, env)) { + !bindingIsLocked(sym, env)) { lockBinding(sym, env) } }, add = TRUE) @@ -176,8 +184,8 @@ } else if (identical(current, entry$installed) && is.null(entry$original)) { generic <- try(get(entry$generic, envir = entry$envir), silent = TRUE) dispatch_env <- if (!inherits(generic, "try-error") && - is.function(generic) && - !is.null(environment(generic))) { + is.function(generic) && + !is.null(environment(generic))) { environment(generic) } else { asNamespace("base") @@ -185,13 +193,13 @@ table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) method_name <- paste(entry$generic, entry$class, sep = ".") if (is.environment(table) && - exists(method_name, envir = table, inherits = FALSE) && - identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { + exists(method_name, envir = table, inherits = FALSE) && + identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { rm(list = method_name, envir = table) } if (isNamespace(entry$envir) && - identical(getNamespaceInfo(entry$envir, "S3methods"), - entry$installed_namespace_methods)) { + identical(getNamespaceInfo(entry$envir, "S3methods"), + entry$installed_namespace_methods)) { try( setNamespaceInfo(entry$envir, "S3methods", entry$original_namespace_methods), silent = TRUE @@ -217,7 +225,7 @@ env <- entry$env name <- entry$name if (exists(name, envir = env, inherits = FALSE) && - identical(get(name, envir = env, inherits = FALSE), entry$installed)) { + identical(get(name, envir = env, inherits = FALSE), entry$installed)) { try(.runtime_assign_binding(name, entry$original, env), silent = TRUE) } } @@ -238,7 +246,7 @@ exists_current <- exists(entry$name, envir = .sess_env, inherits = FALSE) current <- if (exists_current) get(entry$name, envir = .sess_env, inherits = FALSE) else NULL if (identical(exists_current, entry$installed_exists) && - (!exists_current || identical(current, entry$installed))) { + (!exists_current || identical(current, entry$installed))) { if (entry$original_exists) { assign(entry$name, entry$original, envir = .sess_env) } else if (exists_current) { @@ -255,6 +263,7 @@ } } state$options <- list() + .runtime_clear_viewer_state() state$active <- FALSE invisible(NULL) } @@ -266,9 +275,10 @@ runtime_stop <- function() { state <- .runtime_state() if (!isTRUE(state$active) && !length(state$options) && - !length(state$bindings) && !length(state$hooks) && + !length(state$bindings) && !length(state$hooks) && !length(state$s3_methods) && !length(state$task_callbacks) && !length(state$devices) && !length(state$fields)) { + .runtime_clear_viewer_state() return(invisible(NULL)) } # Prevent callbacks from sending new notifications during cleanup. diff --git a/sess/R/server.R b/sess/R/server.R index 4ea2efc9..89402921 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -155,7 +155,7 @@ poll_connection <- function(generation = .sess_env$transport_generation) { ) if (is.null(ready) || is.null(.sess_env$con) || - !identical(generation, .sess_env$transport_generation)) { + !identical(generation, .sess_env$transport_generation)) { return() } diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index a75b186e..67c06afb 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -93,9 +93,10 @@ local({ filterModel = list() )) - expect_equal(length(page_res$rows), 2) - expect_equal(page_res$rows[[1]][["1"]], "3") - expect_equal(page_res$rows[[2]][["1"]], "1") + expect_true(is.data.frame(page_res$rows)) + expect_equal(nrow(page_res$rows), 2) + expect_equal(page_res$rows[["1"]][[1L]], 3) + expect_equal(page_res$rows[["1"]][[2L]], 1) disposed <- sess:::handle_dataview_dispose(list(view_id = registration$view_id)) expect_true(isTRUE(disposed)) @@ -127,9 +128,9 @@ local({ )) expect_equal(filtered$totalRows, 2) - expect_equal(length(filtered$rows), 2) - expect_equal(filtered$rows[[1]][["2"]], "banana") - expect_equal(filtered$rows[[2]][["2"]], "berry") + expect_equal(nrow(filtered$rows), 2) + expect_equal(filtered$rows[["2"]][[1L]], "banana") + expect_equal(filtered$rows[["2"]][[2L]], "berry") sorted <- sess:::handle_dataview_page(list( view_id = registration$view_id, @@ -142,20 +143,24 @@ local({ )) expect_equal(sorted$totalRows, 3) - expect_equal(sorted$rows[[1]][["1"]], "30") - expect_equal(sorted$rows[[2]][["1"]], "20") - expect_equal(sorted$rows[[3]][["1"]], "10") + expect_equal(sorted$rows[["1"]][[1L]], 30) + expect_equal(sorted$rows[["1"]][[2L]], 20) + expect_equal(sorted$rows[["1"]][[3L]], 10) }) # Runtime startup and shutdown are reversible and idempotent. local({ .sess_env <- sess:::.sess_env old_plot_path <- .sess_env$latest_plot_path + old_dataviews <- .sess_env$dataviews + old_dataview_registry <- .sess_env$dataview_registry .sess_env$latest_plot_path <- tempfile(fileext = ".png") on.exit({ sess:::runtime_stop() unlink(.sess_env$latest_plot_path) .sess_env$latest_plot_path <- old_plot_path + .sess_env$dataviews <- old_dataviews + .sess_env$dataview_registry <- old_dataview_registry }, add = TRUE) utils_ns <- asNamespace("utils") @@ -167,7 +172,17 @@ local({ old_help_method <- utils::getS3method("print", "help_files_with_topic", envir = utils_ns) + stale_registry <- new.env(parent = emptyenv()) + assign("stale view", "stale_view_id", envir = stale_registry) + .sess_env$dataviews <- list(stale_view_id = list()) + .sess_env$dataview_registry <- stale_registry + sess:::runtime_stop() + expect_equal(.sess_env$dataviews, list()) + expect_length(ls(.sess_env$dataview_registry, all.names = TRUE), 0L) + sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + expect_equal(.sess_env$dataviews, list()) + expect_length(ls(.sess_env$dataview_registry, all.names = TRUE), 0L) expect_true(isTRUE(sess:::.runtime_state()$active)) expect_false(identical(get("View", utils_ns, inherits = FALSE), old_view)) expect_true(is.function(getOption("viewer"))) @@ -177,17 +192,38 @@ local({ expect_false(identical(getHook("plot.new"), old_plot_hook)) expect_false(identical(getHook("grid.newpage"), old_grid_hook)) + dataview_data <- data.frame(value = 1:2) + assign("lifecycle dataview", "runtime_view_before_restart", + envir = .sess_env$dataview_registry) + utils::View(dataview_data, title = "lifecycle dataview") + first_view_id <- get("lifecycle dataview", envir = .sess_env$dataview_registry) + expect_identical(first_view_id, "runtime_view_before_restart") + expect_true(first_view_id %in% names(.sess_env$dataviews)) + grDevices::pdf(NULL) sess:::.runtime_track_device() runtime_device <- grDevices::dev.cur() callbacks_after_first_start <- getTaskCallbackNames() sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) + expect_equal(.sess_env$dataviews, list()) + expect_length(ls(.sess_env$dataview_registry, all.names = TRUE), 0L) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) expect_equal(length(grep("^sess.plot$", getTaskCallbackNames())), length(grep("^sess.plot$", callbacks_after_first_start))) + utils::View(dataview_data, title = "lifecycle dataview") + second_view_id <- get("lifecycle dataview", envir = .sess_env$dataview_registry) + expect_false(identical(first_view_id, second_view_id)) + expect_true(second_view_id %in% names(.sess_env$dataviews)) + sess:::runtime_stop() + expect_equal(.sess_env$dataviews, list()) + expect_length(ls(.sess_env$dataview_registry, all.names = TRUE), 0L) + expect_error( + sess:::handle_dataview_init(list(view_id = second_view_id)), + "Unknown dataview id" + ) expect_false(isTRUE(sess:::.runtime_state()$active)) expect_identical(get("View", utils_ns, inherits = FALSE), old_view) expect_identical(getOption("browser"), old_options$browser) From b43bf112c2f7b8870b3593d21424c2ca85641a43 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 10:27:59 +0000 Subject: [PATCH 4/9] fix(sess): restore help methods on runtime stop --- sess/R/runtime.R | 74 +++++++++++++++++++++-------------- sess/inst/tinytest/test-ipc.R | 36 +++++++++++------ 2 files changed, 70 insertions(+), 40 deletions(-) diff --git a/sess/R/runtime.R b/sess/R/runtime.R index 16cc2a29..23b31975 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -121,19 +121,33 @@ invisible(NULL) } +.runtime_s3_dispatch_env <- function(generic, envir) { + generic_function <- try(get(generic, envir = envir), silent = TRUE) + if (inherits(generic_function, "try-error") || !is.function(generic_function)) { + return(envir) + } + generic_envir <- environment(generic_function) + if (is.null(generic_envir)) envir else generic_envir +} + .runtime_register_s3 <- function(generic, class, method, envir) { state <- .runtime_state() - original <- utils::getS3method(generic, class, envir = envir, optional = TRUE) + registration_envir <- new.env(parent = envir) + dispatch_envir <- .runtime_s3_dispatch_env(generic, envir) + original <- utils::getS3method(generic, class, envir = dispatch_envir, + optional = TRUE) original_namespace_methods <- if (isNamespace(envir)) { getNamespaceInfo(envir, "S3methods") } else { NULL } - registerS3method(generic, class, method, envir = envir) + registerS3method(generic, class, method, envir = registration_envir) state$s3_methods[[length(state$s3_methods) + 1L]] <- list( generic = generic, class = class, - envir = envir, + envir = registration_envir, + namespace_envir = envir, + dispatch_envir = dispatch_envir, original = original, installed = method, original_namespace_methods = original_namespace_methods, @@ -176,36 +190,38 @@ state$task_callbacks <- list() for (entry in rev(state$s3_methods)) { - current <- utils::getS3method(entry$generic, entry$class, - envir = entry$envir, optional = TRUE) - if (identical(current, entry$installed) && !is.null(entry$original)) { - try(registerS3method(entry$generic, entry$class, entry$original, - envir = entry$envir), silent = TRUE) - } else if (identical(current, entry$installed) && is.null(entry$original)) { - generic <- try(get(entry$generic, envir = entry$envir), silent = TRUE) - dispatch_env <- if (!inherits(generic, "try-error") && - is.function(generic) && - !is.null(environment(generic))) { - environment(generic) + current <- utils::getS3method( + entry$generic, + entry$class, + envir = entry$dispatch_envir, + optional = TRUE + ) + restore_namespace_methods <- isNamespace(entry$namespace_envir) && + identical(getNamespaceInfo(entry$namespace_envir, "S3methods"), + entry$installed_namespace_methods) + + if (identical(current, entry$installed)) { + if (!is.null(entry$original)) { + try(registerS3method(entry$generic, entry$class, entry$original, + envir = entry$envir), silent = TRUE) } else { - asNamespace("base") - } - table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) - method_name <- paste(entry$generic, entry$class, sep = ".") - if (is.environment(table) && + dispatch_env <- entry$dispatch_envir + table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) + method_name <- paste(entry$generic, entry$class, sep = ".") + if (is.environment(table) && exists(method_name, envir = table, inherits = FALSE) && identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { - rm(list = method_name, envir = table) - } - if (isNamespace(entry$envir) && - identical(getNamespaceInfo(entry$envir, "S3methods"), - entry$installed_namespace_methods)) { - try( - setNamespaceInfo(entry$envir, "S3methods", entry$original_namespace_methods), - silent = TRUE - ) + rm(list = method_name, envir = table) + } } } + if (restore_namespace_methods) { + try( + setNamespaceInfo(entry$namespace_envir, "S3methods", + entry$original_namespace_methods), + silent = TRUE + ) + } } state$s3_methods <- list() @@ -276,7 +292,7 @@ runtime_stop <- function() { state <- .runtime_state() if (!isTRUE(state$active) && !length(state$options) && !length(state$bindings) && !length(state$hooks) && - !length(state$s3_methods) && !length(state$task_callbacks) && + !length(state$s3_methods) && !length(state$task_callbacks) && !length(state$devices) && !length(state$fields)) { .runtime_clear_viewer_state() return(invisible(NULL)) diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index 67c06afb..905abbf4 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -170,7 +170,8 @@ local({ old_plot_hook <- getHook("plot.new") old_grid_hook <- getHook("grid.newpage") old_help_method <- utils::getS3method("print", "help_files_with_topic", - envir = utils_ns) + envir = baseenv()) + old_s3_registry <- getNamespaceInfo(utils_ns, "S3methods") stale_registry <- new.env(parent = emptyenv()) assign("stale view", "stale_view_id", envir = stale_registry) @@ -186,8 +187,11 @@ local({ expect_true(isTRUE(sess:::.runtime_state()$active)) expect_false(identical(get("View", utils_ns, inherits = FALSE), old_view)) expect_true(is.function(getOption("viewer"))) - expect_false(identical(utils::getS3method("print", "help_files_with_topic", - envir = utils_ns), old_help_method)) + expect_false(identical( + utils::getS3method("print", "help_files_with_topic", envir = baseenv()), + old_help_method + )) + expect_identical(getNamespaceInfo(utils_ns, "S3methods"), old_s3_registry) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 1L) expect_false(identical(getHook("plot.new"), old_plot_hook)) expect_false(identical(getHook("grid.newpage"), old_grid_hook)) @@ -234,8 +238,11 @@ local({ expect_identical(getHook("plot.new"), old_plot_hook) expect_identical(getHook("grid.newpage"), old_grid_hook) expect_false(runtime_device %in% grDevices::dev.list()) - expect_identical(utils::getS3method("print", "help_files_with_topic", - envir = utils_ns), old_help_method) + expect_identical( + utils::getS3method("print", "help_files_with_topic", envir = baseenv()), + old_help_method + ) + expect_identical(getNamespaceInfo(utils_ns, "S3methods"), old_s3_registry) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) expect_equal(length(grep("^sess.plot$", getTaskCallbackNames())), 0L) }) @@ -317,9 +324,13 @@ local({ expect_false(sess:::.transport_empty_read_is_eof(cons[[2L]])) expect_false(sess:::.transport_empty_read_is_eof(NULL)) close(cons[[1L]]) - chunk <- processx::conn_read_chars(cons[[2L]]) + # Windows can report a closed pipe as a read error (system error 5) instead + # of returning an empty string; the polling cleanup path is tested below. + chunk <- tryCatch(processx::conn_read_chars(cons[[2L]]), error = function(e) NULL) expect_true(is.null(chunk) || !length(chunk) || !any(nzchar(chunk))) - expect_true(sess:::.transport_empty_read_is_eof(cons[[2L]])) + if (!is.null(chunk) && length(chunk)) { + expect_true(sess:::.transport_empty_read_is_eof(cons[[2L]])) + } }) # EOF observed by the polling loop stops the runtime and releases the transport @@ -437,8 +448,9 @@ local({ option_names <- c("browser", "viewer", "page_viewer", "help_type", "device") original_options <- lapply(option_names, getOption) names(original_options) <- option_names - original_help_method <- utils::getS3method("print", "help_files_with_topic", - envir = utils_ns) + original_help_method <- utils::getS3method( + "print", "help_files_with_topic", envir = baseenv() + ) original_plot_hook <- getHook("plot.new") original_grid_hook <- getHook("grid.newpage") second <- NULL @@ -472,8 +484,10 @@ local({ expect_identical(getOption("page_viewer"), original_options$page_viewer) expect_identical(getOption("help_type"), original_options$help_type) expect_identical(getOption("device"), original_options$device) - expect_identical(utils::getS3method("print", "help_files_with_topic", - envir = utils_ns), original_help_method) + expect_identical( + utils::getS3method("print", "help_files_with_topic", envir = baseenv()), + original_help_method + ) expect_identical(getHook("plot.new"), original_plot_hook) expect_identical(getHook("grid.newpage"), original_grid_hook) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) From 7b8f3268c5939509510b49c12c73800fbc8e786d Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 10:49:12 +0000 Subject: [PATCH 5/9] fix(sess): keep polling after transient null results --- sess/R/hooks.R | 17 ++++++++++++++--- sess/R/runtime.R | 4 ++-- sess/R/server.R | 13 ++++++++++++- sess/inst/tinytest/test-ipc.R | 10 ++++++++++ src/session.ts | 7 +++++++ src/test/suite/session.test.ts | 35 ++++++++++++++++++++++++++++++++-- 6 files changed, 78 insertions(+), 8 deletions(-) diff --git a/sess/R/hooks.R b/sess/R/hooks.R index 9eb5c0b5..c0f0db5e 100644 --- a/sess/R/hooks.R +++ b/sess/R/hooks.R @@ -236,7 +236,9 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA plot_updated <<- FALSE last_plot_record_length <<- curr_length .runtime_set_field("latest_plot_record", record) - notify_client("plot_updated") + sent <- notify_client("plot_updated") + # Temporary CI diagnostic; remove after poll/callback cause is known. + message("[sess diagnostic] plot_updated notify sent=", isTRUE(sent)) } } } @@ -252,7 +254,12 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA .runtime_set_hook("grid.newpage", new_plot, "replace") update_plot() - .runtime_add_task_callback(update_plot, name = "sess.plot") + .runtime_add_task_callback(function(...) { + # Temporary CI diagnostic; remove after poll/callback cause is known. + message("[sess diagnostic] entered sess.plot task callback; active=", + isTRUE(.runtime_state()$active)) + update_plot(...) + }, name = "sess.plot") } # 5. rstudioapi hooks @@ -272,8 +279,12 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA # This notifies the client whenever a top-level command is completed, # suggesting that the Global Environment might have changed. .runtime_add_task_callback(function(...) { + # Temporary CI diagnostic; remove after poll/callback cause is known. + message("[sess diagnostic] entered sess.workspace task callback; active=", + isTRUE(.runtime_state()$active)) if (!isTRUE(.runtime_state()$active)) return(FALSE) - notify_client("workspace_updated") + sent <- notify_client("workspace_updated") + message("[sess diagnostic] workspace_updated notify sent=", isTRUE(sent)) TRUE }, name = "sess.workspace") diff --git a/sess/R/runtime.R b/sess/R/runtime.R index 23b31975..f6e788f6 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -209,7 +209,7 @@ table <- get0(".__S3MethodsTable__.", envir = dispatch_env, inherits = FALSE) method_name <- paste(entry$generic, entry$class, sep = ".") if (is.environment(table) && - exists(method_name, envir = table, inherits = FALSE) && + exists(method_name, envir = table, inherits = FALSE) && identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { rm(list = method_name, envir = table) } @@ -293,7 +293,7 @@ runtime_stop <- function() { if (!isTRUE(state$active) && !length(state$options) && !length(state$bindings) && !length(state$hooks) && !length(state$s3_methods) && !length(state$task_callbacks) && - !length(state$devices) && !length(state$fields)) { + !length(state$devices) && !length(state$fields)) { .runtime_clear_viewer_state() return(invisible(NULL)) } diff --git a/sess/R/server.R b/sess/R/server.R index 89402921..1556dd67 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -154,11 +154,22 @@ poll_connection <- function(generation = .sess_env$transport_generation) { } ) - if (is.null(ready) || is.null(.sess_env$con) || + if (is.null(.sess_env$con) || !identical(generation, .sess_env$transport_generation)) { return() } + # processx can return NULL transiently for a just-closed connection before + # the following poll reports readable EOF. Keep the loop alive so the read + # path can confirm EOF and run transport/runtime cleanup. + if (is.null(ready)) { + if (!identical(.sess_env$poll_null_diagnostic_generation, generation)) { + .sess_env$poll_null_diagnostic_generation <- generation + # Temporary CI diagnostic; remove after poll/callback cause is known. + message("[sess diagnostic] poll returned NULL; generation=", generation, + "; connected=", !is.null(.sess_env$con)) + } + } if (!is.null(ready) && length(ready) > 0 && identical(ready[[1]], "ready")) { chunk <- tryCatch( processx::conn_read_chars(con), diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index 905abbf4..3308d899 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -360,8 +360,18 @@ local({ sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) expect_true(isTRUE(sess:::.runtime_state()$active)) + # An idle poll must retain the connection and schedule another poll. + sess:::poll_connection(.sess_env$transport_generation) + expect_true(!is.null(.sess_env$con)) close(cons[[1L]]) sess:::poll_connection(.sess_env$transport_generation) + # A processx poll may return NULL once immediately after peer closure before + # reporting readable EOF. Let the recurring poll callback observe that EOF. + for (i in seq_len(50L)) { + if (is.null(.sess_env$con)) break + Sys.sleep(0.02) + later::run_now() + } expect_null(.sess_env$con) expect_false(isTRUE(sess:::.runtime_state()$active)) expect_equal(length(grep("^sess.workspace$", getTaskCallbackNames())), 0L) diff --git a/src/session.ts b/src/session.ts index 9fd9125f..7d6c3c86 100644 --- a/src/session.ts +++ b/src/session.ts @@ -1707,6 +1707,11 @@ async function handleNotification(message: Record, socket: IpcS } case 'workspace_updated': { + // Temporary CI diagnostic; remove after poll/callback cause is known. + console.info( + '[sess diagnostic] received workspace_updated; active=', + socket === activeSession?.socket + ); if (socket === activeSession?.socket) { scheduleWorkspaceRefresh(); } @@ -1772,6 +1777,8 @@ async function handleNotification(message: Record, socket: IpcS break; } case 'plot_updated': { + // Temporary CI diagnostic; remove after poll/callback cause is known. + console.info('[sess diagnostic] received plot_updated'); void updatePlot(); break; } diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index 16554d48..e6df7d26 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -28,6 +28,7 @@ async function waitFor(condition: () => T | Promise, timeout = 10000, inte suite('Session Communication', () => { let sandbox: sinon.SinonSandbox; + let commandMarkerPath: string | undefined; setup(() => { sandbox = sinon.createSandbox(); @@ -52,6 +53,10 @@ suite('Session Communication', () => { await session.cleanupSession(pid.toString()); } } + if (commandMarkerPath) { + await fs.remove(commandMarkerPath); + commandMarkerPath = undefined; + } sandbox.restore(); }); @@ -85,8 +90,34 @@ suite('Session Communication', () => { const term = rTerminal.rTerm; await new Promise(resolve => setTimeout(resolve, 2000)); - - term.sendText('my_list <- list(hello_vscode = 12345)\n'); + + const markerPath = path.join( + os.tmpdir(), + `vscode-r-command-marker-${process.pid}-${Date.now()}` + ); + commandMarkerPath = markerPath; + await fs.remove(markerPath); + term.sendText( + `my_list <- list(hello_vscode = 12345); ` + + `writeLines("evaluated", ${JSON.stringify(markerPath)})\n` + ); + + // This filesystem marker is independent of the IPC path and confirms + // that R received and evaluated the command before probing the RPC. + await waitFor(() => fs.pathExists(markerPath), 10000, 200); + + // Verify the read/request path independently from the pushed workspace + // refresh notification so a failure distinguishes poll-loop issues + // from task-callback notification issues. + let rpcWorkspace: { globalenv?: Record } | undefined; + await waitFor(async () => { + rpcWorkspace = await session.sessionRequest({ + method: 'workspace', + params: {} + }) as { globalenv?: Record } | undefined; + return rpcWorkspace?.globalenv?.['my_list']; + }, 10000, 200); + assert.ok(rpcWorkspace?.globalenv?.['my_list'], 'workspace RPC should include my_list'); await waitFor(() => { const ge = session.workspaceData?.globalenv; From 7aa21ea4037ebf20d235056deee6e9ea5827c1ff Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 11:04:51 +0000 Subject: [PATCH 6/9] test(sess): isolate callback delivery in session e2e --- sess/R/hooks.R | 11 +++++++ sess/R/runtime.R | 20 +++++++++++- sess/R/server.R | 13 ++++++++ sess/inst/tinytest/test-ipc.R | 3 -- src/test/suite/session.test.ts | 57 +++++++++++++++++++++++++++++++++- 5 files changed, 99 insertions(+), 5 deletions(-) diff --git a/sess/R/hooks.R b/sess/R/hooks.R index c0f0db5e..b3435d38 100644 --- a/sess/R/hooks.R +++ b/sess/R/hooks.R @@ -16,6 +16,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA if (isTRUE(state$active)) runtime_stop() state <- .runtime_state() state$active <- TRUE + state$diagnostics <- .runtime_empty_diagnostics() completed <- FALSE on.exit(if (!completed) try(runtime_stop(), silent = TRUE), add = TRUE) @@ -236,7 +237,11 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA plot_updated <<- FALSE last_plot_record_length <<- curr_length .runtime_set_field("latest_plot_record", record) + .runtime_diagnostic_increment("plot_notify_attempts") sent <- notify_client("plot_updated") + if (isTRUE(sent)) { + .runtime_diagnostic_increment("plot_notify_sent") + } # Temporary CI diagnostic; remove after poll/callback cause is known. message("[sess diagnostic] plot_updated notify sent=", isTRUE(sent)) } @@ -255,6 +260,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA update_plot() .runtime_add_task_callback(function(...) { + .runtime_diagnostic_increment("plot_callback_entries") # Temporary CI diagnostic; remove after poll/callback cause is known. message("[sess diagnostic] entered sess.plot task callback; active=", isTRUE(.runtime_state()$active)) @@ -279,11 +285,16 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA # This notifies the client whenever a top-level command is completed, # suggesting that the Global Environment might have changed. .runtime_add_task_callback(function(...) { + .runtime_diagnostic_increment("workspace_callback_entries") # Temporary CI diagnostic; remove after poll/callback cause is known. message("[sess diagnostic] entered sess.workspace task callback; active=", isTRUE(.runtime_state()$active)) if (!isTRUE(.runtime_state()$active)) return(FALSE) + .runtime_diagnostic_increment("workspace_notify_attempts") sent <- notify_client("workspace_updated") + if (isTRUE(sent)) { + .runtime_diagnostic_increment("workspace_notify_sent") + } message("[sess diagnostic] workspace_updated notify sent=", isTRUE(sent)) TRUE }, name = "sess.workspace") diff --git a/sess/R/runtime.R b/sess/R/runtime.R index f6e788f6..2de0c980 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -1,4 +1,15 @@ # Runtime state is deliberately independent from the IPC connection state. +.runtime_empty_diagnostics <- function() { + list( + workspace_callback_entries = 0L, + workspace_notify_attempts = 0L, + workspace_notify_sent = 0L, + plot_callback_entries = 0L, + plot_notify_attempts = 0L, + plot_notify_sent = 0L + ) +} + .runtime_state <- function() { if (is.null(.sess_env$runtime)) { state <- new.env(parent = emptyenv()) @@ -10,11 +21,18 @@ state$task_callbacks <- list() state$devices <- list() state$fields <- list() + state$diagnostics <- .runtime_empty_diagnostics() .sess_env$runtime <- state } .sess_env$runtime } +.runtime_diagnostic_increment <- function(name) { + state <- .runtime_state() + state$diagnostics[[name]] <- state$diagnostics[[name]] + 1L + invisible(state$diagnostics[[name]]) +} + .runtime_clear_viewer_state <- function() { # Open dataviews and title-to-id mappings belong to one runtime connection. # Keep an empty registry available, but never restore data from a prior run. @@ -210,7 +228,7 @@ method_name <- paste(entry$generic, entry$class, sep = ".") if (is.environment(table) && exists(method_name, envir = table, inherits = FALSE) && - identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { + identical(get(method_name, envir = table, inherits = FALSE), entry$installed)) { rm(list = method_name, envir = table) } } diff --git a/sess/R/server.R b/sess/R/server.R index 1556dd67..7945bd10 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -137,6 +137,18 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, identical(incomplete, FALSE) } +# Temporary CI-only introspection for separating callback failures from client +# notification handling failures. Remove after the lifecycle regression is found. +.runtime_debug_snapshot <- function() { + state <- .runtime_state() + list( + runtime_active = isTRUE(state$active), + task_callback_names = getTaskCallbackNames(), + transport_generation = .sess_env$transport_generation, + callback_counts = state$diagnostics + ) +} + #' Poll the IPC connection for incoming messages (internal) #' #' Runs as a recurring later callback; dispatches NDJSON messages from vscode. @@ -242,6 +254,7 @@ dispatch_message <- function(line) { # Request from vscode → R must reply handlers <- list( "workspace" = function(p) get_workspace_data(), + "debug_runtime_state" = function(p) .runtime_debug_snapshot(), "workspace_children" = function(p) get_workspace_children(p$name, p$path, p$start), "hover" = function(p) handle_hover(p$expr), "completion" = function(p) handle_complete(p$expr, p$trigger), diff --git a/sess/inst/tinytest/test-ipc.R b/sess/inst/tinytest/test-ipc.R index 3308d899..1d5251a4 100644 --- a/sess/inst/tinytest/test-ipc.R +++ b/sess/inst/tinytest/test-ipc.R @@ -360,9 +360,6 @@ local({ sess:::runtime_start(use_rstudioapi = FALSE, use_httpgd = FALSE, use_jgd = FALSE) expect_true(isTRUE(sess:::.runtime_state()$active)) - # An idle poll must retain the connection and schedule another poll. - sess:::poll_connection(.sess_env$transport_generation) - expect_true(!is.null(.sess_env$con)) close(cons[[1L]]) sess:::poll_connection(.sess_env$transport_generation) # A processx poll may return NULL once immediately after peer closure before diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index e6df7d26..5e6798ce 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -14,6 +14,20 @@ import * as plotViewer from '../../plotViewer'; const extension_root: string = path.join(__dirname, '..', '..', '..'); +interface RuntimeDebugSnapshot { + runtime_active?: boolean; + task_callback_names?: string[]; + transport_generation?: number; + callback_counts?: Record; +} + +async function readRuntimeDebugSnapshot(): Promise { + return await session.sessionRequest({ + method: 'debug_runtime_state', + params: {} + }) as RuntimeDebugSnapshot | undefined; +} + async function waitFor(condition: () => T | Promise, timeout = 10000, interval = 100): Promise { const start = Date.now(); while (Date.now() - start < timeout) { @@ -91,6 +105,8 @@ suite('Session Communication', () => { await new Promise(resolve => setTimeout(resolve, 2000)); + const workspaceBaseline = await readRuntimeDebugSnapshot(); + assert.ok(workspaceBaseline?.runtime_active, 'sess runtime should be active before assignment'); const markerPath = path.join( os.tmpdir(), `vscode-r-command-marker-${process.pid}-${Date.now()}` @@ -118,6 +134,22 @@ suite('Session Communication', () => { return rpcWorkspace?.globalenv?.['my_list']; }, 10000, 200); assert.ok(rpcWorkspace?.globalenv?.['my_list'], 'workspace RPC should include my_list'); + + const workspaceDebug = await readRuntimeDebugSnapshot(); + console.info('[session test diagnostic] assignment baseline/after:', + JSON.stringify({ before: workspaceBaseline, after: workspaceDebug })); + assert.ok(workspaceDebug?.runtime_active, 'sess runtime should remain active'); + assert.ok(workspaceDebug?.task_callback_names?.includes('sess.workspace')); + assert.ok( + (workspaceDebug?.callback_counts?.workspace_callback_entries ?? 0) > + (workspaceBaseline?.callback_counts?.workspace_callback_entries ?? 0), + `workspace callback did not run: ${JSON.stringify(workspaceDebug)}` + ); + assert.ok( + (workspaceDebug?.callback_counts?.workspace_notify_sent ?? 0) > + (workspaceBaseline?.callback_counts?.workspace_notify_sent ?? 0), + `workspace_updated notification was not sent: ${JSON.stringify(workspaceDebug)}` + ); await waitFor(() => { const ge = session.workspaceData?.globalenv; @@ -206,7 +238,30 @@ suite('Session Communication', () => { const createWebviewPanelSpy = sandbox.spy(vscode.window, 'createWebviewPanel'); // 1. Test svglite - term.sendText('plot(0, main="svglite")\n'); + const plotBaseline = await readRuntimeDebugSnapshot(); + assert.ok(plotBaseline?.runtime_active, 'sess runtime should be active before plotting'); + const plotMarkerPath = path.join( + os.tmpdir(), + `vscode-r-plot-marker-${process.pid}-${Date.now()}` + ); + commandMarkerPath = plotMarkerPath; + await fs.remove(plotMarkerPath); + term.sendText( + `plot(0, main="svglite"); ` + + `writeLines("evaluated", ${JSON.stringify(plotMarkerPath)})\n` + ); + await waitFor(() => fs.pathExists(plotMarkerPath), 10000, 200); + + const plotDebug = await readRuntimeDebugSnapshot(); + console.info('[session test diagnostic] plot baseline/after:', + JSON.stringify({ before: plotBaseline, after: plotDebug })); + assert.ok(plotDebug?.runtime_active, 'sess runtime should remain active'); + assert.ok(plotDebug?.task_callback_names?.includes('sess.plot')); + assert.ok( + (plotDebug?.callback_counts?.plot_callback_entries ?? 0) > + (plotBaseline?.callback_counts?.plot_callback_entries ?? 0), + `plot task callback did not run: ${JSON.stringify(plotDebug)}` + ); await waitFor(() => createWebviewPanelSpy.calledWith('r.standardPlot'), 10000, 200); assert.ok(createWebviewPanelSpy.calledWith('r.standardPlot'), 'r.standardPlot should be triggered for svglite'); From 03d62da91ed41dcfc8c86fd93143aed856abf99c Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 11:36:14 +0000 Subject: [PATCH 7/9] test(sess): include runtime snapshots in failures --- src/test/suite/session.test.ts | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index 5e6798ce..116cc419 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -106,7 +106,6 @@ suite('Session Communication', () => { await new Promise(resolve => setTimeout(resolve, 2000)); const workspaceBaseline = await readRuntimeDebugSnapshot(); - assert.ok(workspaceBaseline?.runtime_active, 'sess runtime should be active before assignment'); const markerPath = path.join( os.tmpdir(), `vscode-r-command-marker-${process.pid}-${Date.now()}` @@ -136,19 +135,22 @@ suite('Session Communication', () => { assert.ok(rpcWorkspace?.globalenv?.['my_list'], 'workspace RPC should include my_list'); const workspaceDebug = await readRuntimeDebugSnapshot(); + const workspaceSnapshots = JSON.stringify({ before: workspaceBaseline, after: workspaceDebug }); console.info('[session test diagnostic] assignment baseline/after:', - JSON.stringify({ before: workspaceBaseline, after: workspaceDebug })); - assert.ok(workspaceDebug?.runtime_active, 'sess runtime should remain active'); - assert.ok(workspaceDebug?.task_callback_names?.includes('sess.workspace')); + workspaceSnapshots); + assert.ok(workspaceDebug?.runtime_active, + `sess runtime should be active after assignment: ${workspaceSnapshots}`); + assert.ok(workspaceDebug?.task_callback_names?.includes('sess.workspace'), + `sess.workspace callback should be registered: ${workspaceSnapshots}`); assert.ok( (workspaceDebug?.callback_counts?.workspace_callback_entries ?? 0) > (workspaceBaseline?.callback_counts?.workspace_callback_entries ?? 0), - `workspace callback did not run: ${JSON.stringify(workspaceDebug)}` + `workspace callback did not run: ${workspaceSnapshots}` ); assert.ok( (workspaceDebug?.callback_counts?.workspace_notify_sent ?? 0) > (workspaceBaseline?.callback_counts?.workspace_notify_sent ?? 0), - `workspace_updated notification was not sent: ${JSON.stringify(workspaceDebug)}` + `workspace_updated notification was not sent: ${workspaceSnapshots}` ); await waitFor(() => { @@ -239,7 +241,6 @@ suite('Session Communication', () => { // 1. Test svglite const plotBaseline = await readRuntimeDebugSnapshot(); - assert.ok(plotBaseline?.runtime_active, 'sess runtime should be active before plotting'); const plotMarkerPath = path.join( os.tmpdir(), `vscode-r-plot-marker-${process.pid}-${Date.now()}` @@ -253,14 +254,17 @@ suite('Session Communication', () => { await waitFor(() => fs.pathExists(plotMarkerPath), 10000, 200); const plotDebug = await readRuntimeDebugSnapshot(); + const plotSnapshots = JSON.stringify({ before: plotBaseline, after: plotDebug }); console.info('[session test diagnostic] plot baseline/after:', - JSON.stringify({ before: plotBaseline, after: plotDebug })); - assert.ok(plotDebug?.runtime_active, 'sess runtime should remain active'); - assert.ok(plotDebug?.task_callback_names?.includes('sess.plot')); + plotSnapshots); + assert.ok(plotDebug?.runtime_active, + `sess runtime should be active after plotting: ${plotSnapshots}`); + assert.ok(plotDebug?.task_callback_names?.includes('sess.plot'), + `sess.plot callback should be registered: ${plotSnapshots}`); assert.ok( (plotDebug?.callback_counts?.plot_callback_entries ?? 0) > (plotBaseline?.callback_counts?.plot_callback_entries ?? 0), - `plot task callback did not run: ${JSON.stringify(plotDebug)}` + `plot task callback did not run: ${plotSnapshots}` ); await waitFor(() => createWebviewPanelSpy.calledWith('r.standardPlot'), 10000, 200); assert.ok(createWebviewPanelSpy.calledWith('r.standardPlot'), 'r.standardPlot should be triggered for svglite'); From 8cb4a4032ccbb5007467323b94723097c6baef89 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 11:55:21 +0000 Subject: [PATCH 8/9] test(sess): capture runtime startup failure details --- sess/R/hooks.R | 13 ++++++++++++- sess/R/server.R | 28 +++++++++++++++++++++++++--- src/test/suite/session.test.ts | 8 ++++++++ 3 files changed, 45 insertions(+), 4 deletions(-) diff --git a/sess/R/hooks.R b/sess/R/hooks.R index b3435d38..1b034776 100644 --- a/sess/R/hooks.R +++ b/sess/R/hooks.R @@ -12,8 +12,12 @@ register_hooks <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = F #' #' @keywords internal runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FALSE) { + .sess_env$runtime_start_phase <- "initialize" state <- .runtime_state() - if (isTRUE(state$active)) runtime_stop() + if (isTRUE(state$active)) { + .sess_env$runtime_start_phase <- "previous-runtime-stop" + runtime_stop() + } state <- .runtime_state() state$active <- TRUE state$diagnostics <- .runtime_empty_diagnostics() @@ -21,6 +25,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA on.exit(if (!completed) try(runtime_stop(), silent = TRUE), add = TRUE) # 1. Override View() to serve table data via paged RPC. + .sess_env$runtime_start_phase <- "view" if (is.null(.sess_env$dataview_registry)) { .runtime_set_field("dataview_registry", new.env(parent = emptyenv())) } @@ -76,6 +81,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA .runtime_rebind("View", show_dataview, ns = "utils") # 2. Browser & Webview Options + .sess_env$runtime_start_phase <- "viewer-options" make_viewer <- function(method) { function(url, ...) { if (!is.character(url)) { @@ -107,6 +113,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA .runtime_set_option("help_type", "html") # 3. Help System Interception + .sess_env$runtime_start_phase <- "help-s3" sess_print.help_files_with_topic <- function(x, ...) { if (length(x) >= 1 && is.character(x)) { file <- x[1] @@ -139,6 +146,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA invisible(x) } # 4. Plot device: JGD > httpgd > Standard + .sess_env$runtime_start_phase <- "plot" if (use_jgd && nzchar(Sys.getenv("JGD_SOCKET")) && requireNamespace("jgd", quietly = TRUE)) { .runtime_set_option("device", function(...) { jgd::jgd() @@ -269,6 +277,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA } # 5. rstudioapi hooks + .sess_env$runtime_start_phase <- "rstudioapi" if (use_rstudioapi) { rstudioapi_hook <- function(...) { patch_rstudioapi() @@ -282,6 +291,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA } # 6. Workspace Update Callback + .sess_env$runtime_start_phase <- "workspace-callback" # This notifies the client whenever a top-level command is completed, # suggesting that the Global Environment might have changed. .runtime_add_task_callback(function(...) { @@ -299,6 +309,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA TRUE }, name = "sess.workspace") + .sess_env$runtime_start_phase <- "complete" completed <- TRUE invisible(NULL) } diff --git a/sess/R/server.R b/sess/R/server.R index 7945bd10..ea08d2a2 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -100,10 +100,28 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, if (is.na(use_rstudioapi)) use_rstudioapi <- TRUE if (is.na(use_httpgd)) use_httpgd <- TRUE if (is.na(use_jgd)) use_jgd <- FALSE + .sess_env$runtime_start_attempted <- FALSE + .sess_env$runtime_start_error <- NULL + .sess_env$runtime_start_phase <- NULL if (isTRUE(connected) && !is.null(.sess_env$con)) { - runtime_start(use_rstudioapi = use_rstudioapi, - use_httpgd = use_httpgd, - use_jgd = use_jgd) + .sess_env$runtime_start_attempted <- TRUE + tryCatch( + runtime_start(use_rstudioapi = use_rstudioapi, + use_httpgd = use_httpgd, + use_jgd = use_jgd), + error = function(e) { + # Temporary CI diagnostic; rethrow unchanged so startup errors stay visible. + error_call <- conditionCall(e) + .sess_env$runtime_start_error <- list( + step = .sess_env$runtime_start_phase, + message = conditionMessage(e), + call = if (is.null(error_call)) NULL else { + paste(deparse(error_call), collapse = " ") + } + ) + stop(e) + } + ) } invisible(NULL) @@ -145,6 +163,10 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, runtime_active = isTRUE(state$active), task_callback_names = getTaskCallbackNames(), transport_generation = .sess_env$transport_generation, + runtime_start_attempted = isTRUE(.sess_env$runtime_start_attempted), + runtime_start_phase = .sess_env$runtime_start_phase, + runtime_start_error = .sess_env$runtime_start_error, + last_error = geterrmessage(), callback_counts = state$diagnostics ) } diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index 116cc419..a9901510 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -18,6 +18,14 @@ interface RuntimeDebugSnapshot { runtime_active?: boolean; task_callback_names?: string[]; transport_generation?: number; + runtime_start_attempted?: boolean; + runtime_start_phase?: string | null; + runtime_start_error?: { + step?: string | null; + message?: string; + call?: string | null; + } | null; + last_error?: string; callback_counts?: Record; } From fbde5f7273f22f8ae1d5478208653cb85702d354 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 20 Sep 2026 12:10:56 +0000 Subject: [PATCH 9/9] fix(sess): support runtime startup before default packages --- sess/R/hooks.R | 33 +++++------------- sess/R/runtime.R | 28 +++++----------- sess/R/server.R | 57 ++++++++++--------------------- src/session.ts | 7 ---- src/test/suite/session.test.ts | 61 ++-------------------------------- 5 files changed, 37 insertions(+), 149 deletions(-) diff --git a/sess/R/hooks.R b/sess/R/hooks.R index 1b034776..c757c673 100644 --- a/sess/R/hooks.R +++ b/sess/R/hooks.R @@ -20,9 +20,14 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA } state <- .runtime_state() state$active <- TRUE - state$diagnostics <- .runtime_empty_diagnostics() completed <- FALSE - on.exit(if (!completed) try(runtime_stop(), silent = TRUE), add = TRUE) + on.exit({ + if (!completed) { + try(runtime_stop(), silent = TRUE) + } else { + .sess_env$runtime_start_phase <- NULL + } + }, add = TRUE) # 1. Override View() to serve table data via paged RPC. .sess_env$runtime_start_phase <- "view" @@ -245,13 +250,7 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA plot_updated <<- FALSE last_plot_record_length <<- curr_length .runtime_set_field("latest_plot_record", record) - .runtime_diagnostic_increment("plot_notify_attempts") - sent <- notify_client("plot_updated") - if (isTRUE(sent)) { - .runtime_diagnostic_increment("plot_notify_sent") - } - # Temporary CI diagnostic; remove after poll/callback cause is known. - message("[sess diagnostic] plot_updated notify sent=", isTRUE(sent)) + notify_client("plot_updated") } } } @@ -268,10 +267,6 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA update_plot() .runtime_add_task_callback(function(...) { - .runtime_diagnostic_increment("plot_callback_entries") - # Temporary CI diagnostic; remove after poll/callback cause is known. - message("[sess diagnostic] entered sess.plot task callback; active=", - isTRUE(.runtime_state()$active)) update_plot(...) }, name = "sess.plot") } @@ -295,21 +290,11 @@ runtime_start <- function(use_rstudioapi = TRUE, use_httpgd = TRUE, use_jgd = FA # This notifies the client whenever a top-level command is completed, # suggesting that the Global Environment might have changed. .runtime_add_task_callback(function(...) { - .runtime_diagnostic_increment("workspace_callback_entries") - # Temporary CI diagnostic; remove after poll/callback cause is known. - message("[sess diagnostic] entered sess.workspace task callback; active=", - isTRUE(.runtime_state()$active)) if (!isTRUE(.runtime_state()$active)) return(FALSE) - .runtime_diagnostic_increment("workspace_notify_attempts") - sent <- notify_client("workspace_updated") - if (isTRUE(sent)) { - .runtime_diagnostic_increment("workspace_notify_sent") - } - message("[sess diagnostic] workspace_updated notify sent=", isTRUE(sent)) + notify_client("workspace_updated") TRUE }, name = "sess.workspace") - .sess_env$runtime_start_phase <- "complete" completed <- TRUE invisible(NULL) } diff --git a/sess/R/runtime.R b/sess/R/runtime.R index 2de0c980..c62eb22f 100644 --- a/sess/R/runtime.R +++ b/sess/R/runtime.R @@ -1,15 +1,4 @@ # Runtime state is deliberately independent from the IPC connection state. -.runtime_empty_diagnostics <- function() { - list( - workspace_callback_entries = 0L, - workspace_notify_attempts = 0L, - workspace_notify_sent = 0L, - plot_callback_entries = 0L, - plot_notify_attempts = 0L, - plot_notify_sent = 0L - ) -} - .runtime_state <- function() { if (is.null(.sess_env$runtime)) { state <- new.env(parent = emptyenv()) @@ -21,18 +10,11 @@ state$task_callbacks <- list() state$devices <- list() state$fields <- list() - state$diagnostics <- .runtime_empty_diagnostics() .sess_env$runtime <- state } .sess_env$runtime } -.runtime_diagnostic_increment <- function(name) { - state <- .runtime_state() - state$diagnostics[[name]] <- state$diagnostics[[name]] + 1L - invisible(state$diagnostics[[name]]) -} - .runtime_clear_viewer_state <- function() { # Open dataviews and title-to-id mappings belong to one runtime connection. # Keep an empty registry available, but never restore data from a prior run. @@ -41,6 +23,12 @@ invisible(NULL) } +.runtime_named_value <- function(name, value) { + values <- list(value) + names(values) <- name + values +} + .runtime_set_field <- function(name, value) { state <- .runtime_state() index <- which(vapply(state$fields, function(entry) identical(entry$name, name), logical(1))) @@ -69,7 +57,7 @@ if (is.null(state$options[[name]])) { state$options[[name]] <- list(original = getOption(name)) } - do.call(options, setNames(list(value), name)) + do.call(options, .runtime_named_value(name, value)) state$options[[name]]$installed <- getOption(name) invisible(value) } @@ -293,7 +281,7 @@ for (name in names(state$options)) { entry <- state$options[[name]] if (identical(getOption(name), entry$installed)) { - try(do.call(options, setNames(list(entry$original), name)), silent = TRUE) + try(do.call(options, .runtime_named_value(name, entry$original)), silent = TRUE) } } state$options <- list() diff --git a/sess/R/server.R b/sess/R/server.R index ea08d2a2..da93b0d4 100644 --- a/sess/R/server.R +++ b/sess/R/server.R @@ -100,26 +100,32 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, if (is.na(use_rstudioapi)) use_rstudioapi <- TRUE if (is.na(use_httpgd)) use_httpgd <- TRUE if (is.na(use_jgd)) use_jgd <- FALSE - .sess_env$runtime_start_attempted <- FALSE - .sess_env$runtime_start_error <- NULL - .sess_env$runtime_start_phase <- NULL if (isTRUE(connected) && !is.null(.sess_env$con)) { - .sess_env$runtime_start_attempted <- TRUE tryCatch( runtime_start(use_rstudioapi = use_rstudioapi, use_httpgd = use_httpgd, use_jgd = use_jgd), error = function(e) { - # Temporary CI diagnostic; rethrow unchanged so startup errors stay visible. + phase <- .sess_env$runtime_start_phase error_call <- conditionCall(e) - .sess_env$runtime_start_error <- list( - step = .sess_env$runtime_start_phase, - message = conditionMessage(e), - call = if (is.null(error_call)) NULL else { - paste(deparse(error_call), collapse = " ") - } + call_text <- if (is.null(error_call)) { + "" + } else { + paste(deparse(error_call), collapse = " ") + } + call_suffix <- if (nzchar(call_text)) { + paste0(" (", call_text, ")") + } else { + "" + } + message( + "[sess] Runtime startup failed during ", phase, ": ", + conditionMessage(e), call_suffix ) stop(e) + }, + finally = { + .sess_env$runtime_start_phase <- NULL } ) } @@ -155,22 +161,6 @@ connect <- function(pipe_path = NULL, use_rstudioapi = TRUE, use_httpgd = TRUE, identical(incomplete, FALSE) } -# Temporary CI-only introspection for separating callback failures from client -# notification handling failures. Remove after the lifecycle regression is found. -.runtime_debug_snapshot <- function() { - state <- .runtime_state() - list( - runtime_active = isTRUE(state$active), - task_callback_names = getTaskCallbackNames(), - transport_generation = .sess_env$transport_generation, - runtime_start_attempted = isTRUE(.sess_env$runtime_start_attempted), - runtime_start_phase = .sess_env$runtime_start_phase, - runtime_start_error = .sess_env$runtime_start_error, - last_error = geterrmessage(), - callback_counts = state$diagnostics - ) -} - #' Poll the IPC connection for incoming messages (internal) #' #' Runs as a recurring later callback; dispatches NDJSON messages from vscode. @@ -193,17 +183,7 @@ poll_connection <- function(generation = .sess_env$transport_generation) { return() } - # processx can return NULL transiently for a just-closed connection before - # the following poll reports readable EOF. Keep the loop alive so the read - # path can confirm EOF and run transport/runtime cleanup. - if (is.null(ready)) { - if (!identical(.sess_env$poll_null_diagnostic_generation, generation)) { - .sess_env$poll_null_diagnostic_generation <- generation - # Temporary CI diagnostic; remove after poll/callback cause is known. - message("[sess diagnostic] poll returned NULL; generation=", generation, - "; connected=", !is.null(.sess_env$con)) - } - } + # A NULL poll result is transient; keep the loop alive and reschedule below. if (!is.null(ready) && length(ready) > 0 && identical(ready[[1]], "ready")) { chunk <- tryCatch( processx::conn_read_chars(con), @@ -276,7 +256,6 @@ dispatch_message <- function(line) { # Request from vscode → R must reply handlers <- list( "workspace" = function(p) get_workspace_data(), - "debug_runtime_state" = function(p) .runtime_debug_snapshot(), "workspace_children" = function(p) get_workspace_children(p$name, p$path, p$start), "hover" = function(p) handle_hover(p$expr), "completion" = function(p) handle_complete(p$expr, p$trigger), diff --git a/src/session.ts b/src/session.ts index 7d6c3c86..9fd9125f 100644 --- a/src/session.ts +++ b/src/session.ts @@ -1707,11 +1707,6 @@ async function handleNotification(message: Record, socket: IpcS } case 'workspace_updated': { - // Temporary CI diagnostic; remove after poll/callback cause is known. - console.info( - '[sess diagnostic] received workspace_updated; active=', - socket === activeSession?.socket - ); if (socket === activeSession?.socket) { scheduleWorkspaceRefresh(); } @@ -1777,8 +1772,6 @@ async function handleNotification(message: Record, socket: IpcS break; } case 'plot_updated': { - // Temporary CI diagnostic; remove after poll/callback cause is known. - console.info('[sess diagnostic] received plot_updated'); void updatePlot(); break; } diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index a9901510..6d2d38fe 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -14,28 +14,6 @@ import * as plotViewer from '../../plotViewer'; const extension_root: string = path.join(__dirname, '..', '..', '..'); -interface RuntimeDebugSnapshot { - runtime_active?: boolean; - task_callback_names?: string[]; - transport_generation?: number; - runtime_start_attempted?: boolean; - runtime_start_phase?: string | null; - runtime_start_error?: { - step?: string | null; - message?: string; - call?: string | null; - } | null; - last_error?: string; - callback_counts?: Record; -} - -async function readRuntimeDebugSnapshot(): Promise { - return await session.sessionRequest({ - method: 'debug_runtime_state', - params: {} - }) as RuntimeDebugSnapshot | undefined; -} - async function waitFor(condition: () => T | Promise, timeout = 10000, interval = 100): Promise { const start = Date.now(); while (Date.now() - start < timeout) { @@ -113,7 +91,6 @@ suite('Session Communication', () => { await new Promise(resolve => setTimeout(resolve, 2000)); - const workspaceBaseline = await readRuntimeDebugSnapshot(); const markerPath = path.join( os.tmpdir(), `vscode-r-command-marker-${process.pid}-${Date.now()}` @@ -129,9 +106,8 @@ suite('Session Communication', () => { // that R received and evaluated the command before probing the RPC. await waitFor(() => fs.pathExists(markerPath), 10000, 200); - // Verify the read/request path independently from the pushed workspace - // refresh notification so a failure distinguishes poll-loop issues - // from task-callback notification issues. + // Verify the workspace request path after command execution, independently + // from the pushed workspace refresh notification. let rpcWorkspace: { globalenv?: Record } | undefined; await waitFor(async () => { rpcWorkspace = await session.sessionRequest({ @@ -141,25 +117,6 @@ suite('Session Communication', () => { return rpcWorkspace?.globalenv?.['my_list']; }, 10000, 200); assert.ok(rpcWorkspace?.globalenv?.['my_list'], 'workspace RPC should include my_list'); - - const workspaceDebug = await readRuntimeDebugSnapshot(); - const workspaceSnapshots = JSON.stringify({ before: workspaceBaseline, after: workspaceDebug }); - console.info('[session test diagnostic] assignment baseline/after:', - workspaceSnapshots); - assert.ok(workspaceDebug?.runtime_active, - `sess runtime should be active after assignment: ${workspaceSnapshots}`); - assert.ok(workspaceDebug?.task_callback_names?.includes('sess.workspace'), - `sess.workspace callback should be registered: ${workspaceSnapshots}`); - assert.ok( - (workspaceDebug?.callback_counts?.workspace_callback_entries ?? 0) > - (workspaceBaseline?.callback_counts?.workspace_callback_entries ?? 0), - `workspace callback did not run: ${workspaceSnapshots}` - ); - assert.ok( - (workspaceDebug?.callback_counts?.workspace_notify_sent ?? 0) > - (workspaceBaseline?.callback_counts?.workspace_notify_sent ?? 0), - `workspace_updated notification was not sent: ${workspaceSnapshots}` - ); await waitFor(() => { const ge = session.workspaceData?.globalenv; @@ -248,7 +205,6 @@ suite('Session Communication', () => { const createWebviewPanelSpy = sandbox.spy(vscode.window, 'createWebviewPanel'); // 1. Test svglite - const plotBaseline = await readRuntimeDebugSnapshot(); const plotMarkerPath = path.join( os.tmpdir(), `vscode-r-plot-marker-${process.pid}-${Date.now()}` @@ -261,19 +217,6 @@ suite('Session Communication', () => { ); await waitFor(() => fs.pathExists(plotMarkerPath), 10000, 200); - const plotDebug = await readRuntimeDebugSnapshot(); - const plotSnapshots = JSON.stringify({ before: plotBaseline, after: plotDebug }); - console.info('[session test diagnostic] plot baseline/after:', - plotSnapshots); - assert.ok(plotDebug?.runtime_active, - `sess runtime should be active after plotting: ${plotSnapshots}`); - assert.ok(plotDebug?.task_callback_names?.includes('sess.plot'), - `sess.plot callback should be registered: ${plotSnapshots}`); - assert.ok( - (plotDebug?.callback_counts?.plot_callback_entries ?? 0) > - (plotBaseline?.callback_counts?.plot_callback_entries ?? 0), - `plot task callback did not run: ${plotSnapshots}` - ); await waitFor(() => createWebviewPanelSpy.calledWith('r.standardPlot'), 10000, 200); assert.ok(createWebviewPanelSpy.calledWith('r.standardPlot'), 'r.standardPlot should be triggered for svglite');