starred/oh-my-posh
1
0
Fork 0
mirror of https://github.com/JanDeDobbeleer/oh-my-posh.git synced 2026-08-24 10:14:12 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat(shell): give the nu serve daemon a Windows transport

Browse Source 
mkfifo doesn't exist on Windows, so the nu serve daemon added earlier
this session never started there - every prompt silently fell back to
the per-prompt subprocess spawn. Named pipes were tried first and
ruled out empirically: nu can read a Windows named pipe via 'open
--raw', but every nu write primitive (save --append/--force, o>, o>>)
fails or silently no-ops against one, so requests have no way to reach
the daemon.

The daemon now serves Windows over loopback HTTP instead: it binds an
ephemeral 127.0.0.1 port, publishes '<port> <token>' to a file the
shell polls, accepts one JSON request per prompt via POST /, and
streams NUL-delimited records back over one long-lived, token-gated
GET /stream response nu's reader job holds open for the daemon's
whole lifetime - the moral equivalent of the fifo pair, with the
stream disconnecting standing in for SIGPIPE as the shell-disappeared
signal (plus a 30s no-client guard against orphaning). serve.go's
request/response plumbing is generalized to io.Reader/io.Writer so
both transports share the same protocol code.

Fixed two bugs the cross-platform path surfaced: .PATH needs
nu's platform-aware 'char esep' (';' on Windows, ':' on Unix) instead
of a hardcoded ':', and '.temp-path' was renamed to '.temp-dir'
in current nu releases.

Also overlaps the right prompt's render with the streamed primary:
nu evaluates PROMPT_COMMAND before PROMPT_COMMAND_RIGHT on every
prompt, so with an rprompt configured the right render was a second,
fully sequential process spawn on top of the streamed left prompt.
PROMPT_COMMAND now kicks off the right render as a background job
before blocking on the primary, so the two spawns overlap instead of
stacking - measured ~47ms to ~34.5ms per cycle on the same binary.

Verified end-to-end on real nu 0.114.2 and a real oh-my-posh binary:
HTTP daemon starts once, survives repeated prompt cycles without
respawning, warm cycles ~9-15ms (on par with pwsh's own serve
daemon), and shuts down with zero orphaned processes. Findings
(named-pipe dead end, HTTP recipe, streaming-through-a-def gotcha,
head-to-head nu/pwsh timings) are recorded in
.agents/skills/project-knowledge/references/nu.md.

Entire-Checkpoint: 6dde1d189faa
This commit is contained in:
Jan De Dobbeleer
2026-08-05 00:17:39 +02:00
parent bed44bf8a2
commit 5c77b4c72f
7 changed files with 970 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.agents/skills/project-knowledge/references/nu.md
+100 -14
View File
@@ -86,9 +86,9 @@
| each {...} }` reading the daemon's own stdout redirected to a second FIFO. **This recipe is
now verified working end-to-end** (2026-08-04, WSL/aarch64, real nu 0.107.0 + a cross-compiled
real `oh-my-posh` binary) and implemented in `src/shell/nu.go`'s `Streaming` case
(`_omp_serve_start`/`_omp_serve_stop`/`_omp_serve_render`). It is Unix-only (gated by `which
mkfifo | is-empty`) - Windows named pipes would need `CreateNamedPipe`-based server plumbing on
the Go side, a much larger change, not attempted.
(`_omp_serve_start`/`_omp_serve_stop`/`_omp_serve_render`). The FIFO transport is Unix-only
(gated by `which mkfifo | is-empty`); on Windows the same daemon runs over an HTTP loopback
transport instead (see below).
- **`job spawn { ^exe args... out> $fifo }` (a direct external-command file redirect written
straight inside a job-spawn closure body) silently never forks the child process at all** - no
error, `job spawn` still returns a job id, but `ps`/`pgrep` show nothing. This is NOT a general
@@ -140,6 +140,54 @@
`job kill`. Use the existing graceful protocol instead: write `{"command":"quit"}\n` to the
daemon's request fifo (matches `serveCommandQuit` in `serve.go`), which makes the real daemon
exit cleanly, which in turn lets the nested nu wrapper's script line finish naturally.
- **Windows named pipes are a dead end for nu-driven daemon I/O** (verified 2026-08-04, nu
0.114.2, Windows 11): nu *can* read from `\\.\pipe\<name>` via `open --raw` (the server must
be multi-instance and tolerate nu's extra probe connections - nu opens the path once for
metadata before the real read), but **every nu write primitive fails against a pipe path**:
`save --append` → "permission denied" (Rust's `OpenOptions::append` is invalid on pipes),
`save --force`/`o>` → "truncate not supported", plain `save` → "file exists", `o>>` →
silently writes nothing. With no way to send requests, a two-pipe FIFO-analogue design is
impossible no matter how the Go server is written. Don't re-litigate this with go-winio
server tweaks - the blocker is nu's client-side write path, not the server.
- **The working Windows transport is HTTP over loopback** (implemented 2026-08-04 in
`src/cli/serve_http.go` + the `_omp_serve_start_http` branch of `src/shell/nu.go`): the
daemon binds `127.0.0.1:0`, writes `"<port> <token>\n"` atomically (temp file + rename) to a
`--port-file` the shell polls, and serves two endpoints - `POST /` (one JSON request per
prompt, same wire format as the fifo/stdin transports) and `GET /stream` (one long-lived
chunked response carrying every NUL-delimited record for the daemon's lifetime, held open by
the shell's reader job). Nu's built-in `http get ... | bytes split 0x[00]` consumes chunked
responses **incrementally** (verified: first record ~3ms after POST, later records arrive
live mid-response), and `http post --max-time 3sec` is a fire-and-forget request writer, so
the shell script shape is identical to the fifo recipe with `http` calls swapped in for
`save --append`/`open --raw`. Measured end-to-end on the same machine as the earlier
baselines: warm prompt cycles ~9-12ms, 6-cycle average incl. cold daemon start ~21ms, vs
~57ms for the per-prompt subprocess fallback and ~43-48ms for the plain streamed spawn.
A 16-byte hex token (`X-Omp-Token` header, constant-time compare) gates both endpoints
since any local process can reach loopback. Caveat: `HTTP_PROXY`-style env vars could in
principle divert loopback requests if nu's client honors them without a `NO_PROXY`
exemption - if a machine shows instant permanent fallback (3 failures), check proxy env.
- **Piping a stream through a `def` that consumes `$in` destroys incrementality** (verified:
first record arrives at response *completion* ~1.5s instead of ~3ms when the
`bytes split`/`each` loop is moved into a helper `def`). The reader loop must live inline in
the `job spawn` closure body - the duplication between the fifo and HTTP reader jobs in
`nu.go` is deliberate, do not refactor it into a shared `def`.
- `$nu.temp-path` no longer exists in nu 0.114.x - it was renamed to **`$nu.temp-dir`**. Since
any nu build new enough for the Streaming feature (needs `commandline set-prompt`) is well
past the rename, generated scripts must use `temp-dir`; `temp-path` raises
`nu::shell::name_not_found` at runtime (it passes parsing, so it only explodes when the
serve path actually executes).
- `o+e> NUL` works at the top level of a nu script on Windows (`NUL` resolves to the Win32
null device; no literal `NUL` file is created) - used by the nested-nu daemon wrapper to
silence the daemon's stderr, mirroring the `out> $fifo` role the wrapper plays on Unix.
- The serve daemon `os.Chdir`s into each render request's `pwd` (`startRenderCycle` in
`serve.go`), and Windows locks a process's working directory against deletion - so a
directory you just `cd`'d out of can still be undeletable ("being used by another process")
until the *next* prompt render moves the daemon elsewhere. The pwsh daemon has the same
behavior; it only surfaces in scripted tests that render in a temp dir and immediately
`rm` it.
- `$env.CMD_DURATION_MS` only exists in an interactive REPL session - a script that `source`s
the init and calls `_omp_stream_primary` directly (e.g. an E2E harness) must set it
manually first or the `match $env.CMD_DURATION_MS` line raises `column_not_found`.
- `which mkfifo | is-empty` is a reliable, portable way to gate Unix-FIFO-only functionality
without any Go-side platform detection/build tags - it's simply empty (falsy) on Windows, no
error.
@@ -151,15 +199,53 @@
nu binary (needed to launch the nested-nu daemon wrapper above) - prefer it over `which nu | get
0.path`, which can raise `nu::shell::access_beyond_end` if the `which` result table happens to
be empty in some invocation contexts.
- **Known unresolved limitation:** a job spawned via `job spawn` outlives the nu process that
spawned it if that process exits normally (verified: `nu -c 'job spawn { ^sleep 30 }; print
done'` prints "done" and exits while `sleep 30` keeps running, orphaned). Nu has no general
"on shell exit" hook (only `pre_prompt`/`pre_execution`/`env_change`/`display_output`), unlike
zsh's `zshexit_functions` or fish's `fish_exit`. This means the serve daemon (and its nested-nu
wrapper and reader job) can be left running as orphaned background processes if a nu session
ends abruptly (window closed, terminal killed) without ever calling `_omp_serve_stop` - there is
currently no nu-side mechanism to guarantee cleanup in that case, unlike the fd-ownership-based
natural cleanup zsh/fish get from their coprocess mechanisms. Not yet mitigated; a future
improvement could investigate whether terminal-close SIGHUP reaches the daemon's process group
the same way it reaches the shell's.
- **Nu evaluates `$env.PROMPT_COMMAND` and then `$env.PROMPT_COMMAND_RIGHT` sequentially on the
same thread before painting anything** (verified in nushell's `prompt_update.rs`,
`update_prompt`: `get_prompt_string(PROMPT_COMMAND, ...)` runs to completion before
`get_prompt_string(PROMPT_COMMAND_RIGHT, ...)`). With an `rprompt` block configured, the base
`omp.nu` right closure (`_omp_get_prompt right`) added a second full blocking process spawn to
every prompt cycle, on top of the streamed primary - measured 2026-08-04 on Windows: left
(streamed, per-prompt fallback) ~46ms + right ~34ms = ~80ms per cycle. The Streaming feature in
`nu.go` now overlaps them: `PROMPT_COMMAND` spawns the right render as a `job spawn` before
blocking on the primary's first record, collects it via `job recv` afterward, and
`PROMPT_COMMAND_RIGHT` just reads `$env._omp_right_prompt` (an env write from `PROMPT_COMMAND`
propagates to the right closure; the right closure itself never needs to write env, avoiding any
dependency on whether its own mutations would propagate). Same-binary before/after: ~47ms →
~34.5ms per full cycle, right prompt byte-identical.
- Head-to-head prompt-cycle numbers on the same Windows machine/config/binary (2026-08-04,
measured, not estimated): pwsh serve daemon warm ~12-15ms; pwsh legacy per-prompt stream spawn
(daemon disabled) ~61-63ms; nu streaming per-prompt fallback ~43-48ms (left only). Nu's
fallback is *faster* than pwsh's own no-daemon fallback - the entire "nu feels slower than
pwsh" gap on Windows was (a) pwsh's working serve daemon (nu's was Unix-FIFO-only until the
HTTP loopback transport above closed that gap: nu warm cycles now ~9-12ms, on par with pwsh's
~12-15ms) plus (b) the
extra sequential right-prompt spawn above, now overlapped. Nothing measured suggests inherent
nu-runtime overhead: raw external spawn cost is comparable from both shells (`^cmd /c exit`
~40ms from nu vs `& cmd /c exit` ~35ms from pwsh; Windows process creation dominates),
`job spawn` + mailbox round-trip costs ~0.1ms, `term size`/`job list | length` are ~0.05ms.
- Per-prompt housekeeping costs worth knowing (Windows, nu 0.114.2): `history` with a ~14k-entry
SQLite history takes ~9ms per call (and `| length`/`| last 1` don't make it cheaper - the
full read happens regardless), so the history-growth bookkeeping in `PROMPT_COMMAND` has a
real, history-size-scaling cost; `which mkfifo` costs ~7ms per call (moot on Windows now that
`_omp_serve_start` dispatches on `$nu.os-info.name` before any mkfifo probe).
- A `oh-my-posh.exe` resolved through the WindowsApps app-execution alias
(`...\Microsoft\WindowsApps\oh-my-posh.exe`, MSIX install) costs ~15-20ms *extra per process
spawn* versus invoking the same-version binary at a plain filesystem path (measured: right
render ~34ms via alias vs ~15ms via direct path, `--version` spawn ~87ms avg via alias vs
~29ms direct from pwsh). Benchmarks of per-prompt spawn cost must pin the executable path, or
the alias tax drowns out the effect being measured.
- **Known unresolved limitation (Unix fifo transport only):** a job spawned via `job spawn`
outlives the nu process that spawned it if that process exits normally (verified: `nu -c 'job
spawn { ^sleep 30 }; print done'` prints "done" and exits while `sleep 30` keeps running,
orphaned). Nu has no general "on shell exit" hook (only
`pre_prompt`/`pre_execution`/`env_change`/`display_output`), unlike zsh's `zshexit_functions`
or fish's `fish_exit`. This means the fifo-transport daemon (and its nested-nu wrapper and
reader job) can be left running as orphaned background processes if a nu session ends abruptly
(window closed, terminal killed) without ever calling `_omp_serve_stop` - there is currently
no nu-side mechanism to guarantee cleanup in that case, unlike the fd-ownership-based natural
cleanup zsh/fish get from their coprocess mechanisms. **The HTTP transport does not share this
gap:** when the nu process dies for any reason its jobs die with it, the `/stream` TCP
connection drops, and the Go server treats that disconnect as the shell-disappeared signal
(the SIGPIPE analogue) and exits gracefully - plus a belt-and-braces guard shuts the daemon
down if no stream client connects within 30s of startup.
src/cli/serve.go
+31 -7
View File
@@ -5,6 +5,7 @@ import (
"bytes"
"encoding/json"
"fmt"
"io"
"os"
"github.com/jandedobbeleer/oh-my-posh/src/cache"
@@ -20,6 +21,12 @@ import (
// prompts (fish).
var requestPipe string
// Windows nu: mkfifo doesn't exist and nu's save/redirect primitives cannot
// write to Windows named pipes, so the daemon serves requests over a
// loopback HTTP endpoint instead and publishes "<port> <token>" to this file
// (see serve_http.go).
var portFile string
var serveCmd = createServeCmd()
func init() {
@@ -77,9 +84,24 @@ func createServeCmd() *cmdtree.Command {
shellName = shell.GENERIC
}
in, err := openServeInput(requestPipe)
if err != nil {
os.Exit(1)
// Bring the transport up before cache.Init: a transport failure
// exits with os.Exit, which would skip the deferred cache.Close.
var run func() bool
if portFile != "" {
server, err := startServeHTTP(portFile)
if err != nil {
os.Exit(1)
}
run = server.run
} else {
in, err := openServeInput(requestPipe)
if err != nil {
os.Exit(1)
}
run = func() bool { return runServeLoop(in, os.Stdout) }
}
options := []cache.Option{cache.Persist}
@@ -92,7 +114,7 @@ func createServeCmd() *cmdtree.Command {
// least once (it reads package-level state set there); if the
// daemon quits/hits EOF before ever handling a render request,
// skip it instead of panicking on that unset state.
if renderedAtLeastOnce := runServeLoop(in, os.Stdout); renderedAtLeastOnce {
if renderedAtLeastOnce := run(); renderedAtLeastOnce {
template.SaveCache()
}
},
@@ -100,9 +122,11 @@ func createServeCmd() *cmdtree.Command {
serveCmd.Flags().StringVar(&shellName, "shell", "", "the shell to serve for")
serveCmd.Flags().StringVar(&requestPipe, "request-pipe", "", "named pipe (fifo) to read requests from instead of stdin")
serveCmd.Flags().StringVar(&portFile, "port-file", "", "serve over loopback HTTP and write '<port> <token>' to this file")
// Hide flags that are for internal use only.
_ = serveCmd.Flags().MarkHidden("request-pipe")
_ = serveCmd.Flags().MarkHidden("port-file")
return serveCmd
}
@@ -151,7 +175,7 @@ type serveActiveCycle struct {
// StreamPrimary) so a broken render costs one prompt, not the daemon; the
// shell additionally redirects this process's stderr so anything unrecovered
// can never reach the user's terminal.
func runServeLoop(in, out *os.File) bool {
func runServeLoop(in io.Reader, out io.Writer) bool {
scanner := bufio.NewScanner(in)
// Env payloads (a POSH_* overlay plus PATH) can exceed the default 64 KB
// scanner buffer, so grow it up front.
@@ -259,7 +283,7 @@ func applyEnvOverlay(env map[string]string, keys map[string]struct{}) {
// A panic while setting up the cycle (e.g. in prompt.New) is recovered and
// reported as "no cycle": the daemon stays alive, the shell's waiter times
// out and falls back to the legacy path for that prompt.
func startRenderCycle(req *serveRequest, out *os.File, envKeys map[string]struct{}) (cycle *serveActiveCycle) {
func startRenderCycle(req *serveRequest, out io.Writer, envKeys map[string]struct{}) (cycle *serveActiveCycle) {
defer func() {
if r := recover(); r != nil {
cycle = nil
@@ -374,7 +398,7 @@ func renderComplete(eng *prompt.Engine) <-chan string {
// copyRecords copies prompt records to out prefixed with the cycle id and
// closes the returned channel once the source channel is exhausted.
func copyRecords(id int64, records <-chan string, out *os.File) chan struct{} {
func copyRecords(id int64, records <-chan string, out io.Writer) chan struct{} {
done := make(chan struct{})
go func() {
src/cli/serve_http.go
+280
View File
@@ -0,0 +1,280 @@
package cli
import (
"bytes"
"context"
"crypto/rand"
"crypto/subtle"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net"
"net/http"
"os"
"sync"
"time"
)
// The HTTP transport exists for Windows nu: mkfifo doesn't exist there and
// nu's write primitives (save --append/--force, o> redirects) all fail
// against Windows named pipes, but its built-in http client both fires
// one-shot POSTs and consumes chunked responses incrementally. Requests
// arrive as single POSTs on "/", and every prompt record flows over one
// long-lived "/stream" response the shell's reader job holds open for the
// daemon's whole lifetime - the moral equivalent of the fifo pair, with the
// stream disconnect replacing SIGPIPE as the shell-disappeared signal.
// serveTokenHeader carries the shared secret published in the port file.
// The listener is loopback-only, but any local process can connect to it -
// the token keeps other users' processes (and browsers doing DNS-rebinding
// tricks) from driving the daemon.
const serveTokenHeader = "X-Omp-Token"
// serveStreamTimeout bounds how long the daemon waits for the shell's
// reader job to connect to /stream - at startup (a daemon nobody ever
// connects to must not linger as an orphan) and per render request.
const serveStreamTimeout = 30 * time.Second
type serveHTTPServer struct {
listener net.Listener
token string
portFile string
// streamReady is closed once the stream client connected; stream is
// written exactly once before that and never mutated after.
streamReady chan struct{}
stream io.Writer
done chan struct{}
doneOnce sync.Once
mu sync.Mutex
active *serveActiveCycle
envKeys map[string]struct{}
rendered bool
}
// startServeHTTP binds a loopback listener on an ephemeral port and
// publishes "<port> <token>" to portFile so the shell can find it. The
// write is atomic (temp file + rename): the shell polls the file and must
// never observe a partial write.
func startServeHTTP(portFile string) (*serveHTTPServer, error) {
listener, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
if err != nil {
return nil, err
}
raw := make([]byte, 16)
if _, err = rand.Read(raw); err != nil {
listener.Close()
return nil, err
}
server := &serveHTTPServer{
listener: listener,
token: hex.EncodeToString(raw),
portFile: portFile,
streamReady: make(chan struct{}),
done: make(chan struct{}),
envKeys: map[string]struct{}{},
}
port := listener.Addr().(*net.TCPAddr).Port
temp := portFile + ".tmp"
if err = os.WriteFile(temp, fmt.Appendf(nil, "%d %s\n", port, server.token), 0o600); err != nil {
listener.Close()
return nil, err
}
if err = os.Rename(temp, portFile); err != nil {
_ = os.Remove(temp)
listener.Close()
return nil, err
}
return server, nil
}
// run serves until quit or stream disconnect and reports whether at least
// one render request was handled (same contract as runServeLoop).
func (s *serveHTTPServer) run() bool {
mux := http.NewServeMux()
mux.HandleFunc("POST /", s.handleRequest)
mux.HandleFunc("GET /stream", s.handleStream)
server := &http.Server{
Handler: mux,
ReadHeaderTimeout: 5 * time.Second,
}
go func() {
_ = server.Serve(s.listener)
}()
// A daemon whose stream client never shows up (the shell died between
// spawning it and connecting the reader job) must not linger forever.
go func() {
select {
case <-s.streamReady:
case <-s.done:
case <-time.After(serveStreamTimeout):
s.shutdown()
}
}()
<-s.done
s.mu.Lock()
s.stopActiveLocked()
rendered := s.rendered
s.mu.Unlock()
_ = server.Close()
_ = os.Remove(s.portFile)
return rendered
}
func (s *serveHTTPServer) shutdown() {
s.doneOnce.Do(func() { close(s.done) })
}
func (s *serveHTTPServer) authorized(r *http.Request) bool {
token := r.Header.Get(serveTokenHeader)
return subtle.ConstantTimeCompare([]byte(token), []byte(s.token)) == 1
}
// stopActiveLocked mirrors runServeLoop's stopActiveCycle; see the
// serveActiveCycle doc for why the previous cycle must be fully stopped
// (Abort returned, copier drained) before the next one may start.
func (s *serveHTTPServer) stopActiveLocked() {
if s.active == nil {
return
}
s.active.engine.Abort()
<-s.active.copierDone
s.active = nil
}
// handleStream is the single long-lived record sink: every cycle's records
// are written (and flushed) to this response. The shell's reader job holds
// it open for the daemon's lifetime, so the connection breaking means the
// shell is gone - the daemon's cue to exit, exactly like SIGPIPE on the
// fifo transport.
func (s *serveHTTPServer) handleStream(w http.ResponseWriter, r *http.Request) {
if !s.authorized(r) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
s.mu.Lock()
if s.stream != nil {
s.mu.Unlock()
http.Error(w, "stream already connected", http.StatusConflict)
return
}
s.stream = &flushWriter{writer: w, flusher: flusher}
s.mu.Unlock()
close(s.streamReady)
w.WriteHeader(http.StatusOK)
flusher.Flush()
select {
case <-r.Context().Done():
s.shutdown()
case <-s.done:
}
}
// handleRequest accepts one serveRequest JSON per POST body - the protocol
// of runServeLoop with the newline-delimited stdin stream replaced by
// one-shot requests. Responses carry no payload; render records flow over
// /stream.
func (s *serveHTTPServer) handleRequest(w http.ResponseWriter, r *http.Request) {
if !s.authorized(r) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
body, err := io.ReadAll(io.LimitReader(r.Body, 1024*1024))
if err != nil {
http.Error(w, "unreadable body", http.StatusBadRequest)
return
}
body = bytes.TrimPrefix(body, []byte{0xEF, 0xBB, 0xBF})
var req serveRequest
if err := json.Unmarshal(body, &req); err != nil {
// Malformed request: ignore for forward/backward compatibility,
// like runServeLoop does with malformed lines.
w.WriteHeader(http.StatusOK)
return
}
switch req.Command {
case serveCommandRender:
// Records have nowhere to go until the reader job is on /stream;
// the connect happens concurrently with the first render request,
// so wait for it instead of dropping the cycle.
select {
case <-s.streamReady:
case <-s.done:
http.Error(w, "shutting down", http.StatusServiceUnavailable)
return
case <-time.After(serveStreamTimeout):
http.Error(w, "no stream client", http.StatusServiceUnavailable)
return
}
s.mu.Lock()
s.stopActiveLocked()
if s.active = startRenderCycle(&req, s.stream, s.envKeys); s.active != nil {
s.rendered = true
}
s.mu.Unlock()
w.WriteHeader(http.StatusOK)
case serveCommandAbort:
s.mu.Lock()
s.stopActiveLocked()
s.mu.Unlock()
w.WriteHeader(http.StatusOK)
case serveCommandQuit:
s.mu.Lock()
s.stopActiveLocked()
s.mu.Unlock()
w.WriteHeader(http.StatusOK)
s.shutdown()
default:
// Unknown command: ignore for forward compatibility.
w.WriteHeader(http.StatusOK)
}
}
// flushWriter flushes after every write so each record leaves the daemon
// immediately - nu's http client surfaces chunks as they arrive, which is
// what makes streamed segment updates paint live.
type flushWriter struct {
writer io.Writer
flusher http.Flusher
}
func (fw *flushWriter) Write(p []byte) (int, error) {
n, err := fw.writer.Write(p)
fw.flusher.Flush()
return n, err
}
src/cli/serve_http_test.go
+205
View File
@@ -0,0 +1,205 @@
package cli
import (
"context"
"encoding/json"
"fmt"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// httpHarness wires a serveHTTPServer to a stream-reading client so each
// test only expresses protocol traffic - the HTTP sibling of serveHarness.
type httpHarness struct {
t *testing.T
url string
token string
reader *recordReader
stream *http.Response
done chan struct{}
rendered bool
}
func startHTTPHarness(t *testing.T) *httpHarness {
t.Helper()
t.Setenv("OMP_CACHE_DIR", t.TempDir())
portFilePath := filepath.Join(t.TempDir(), "omp-serve-test.port")
server, err := startServeHTTP(portFilePath)
require.NoError(t, err)
contents, err := os.ReadFile(portFilePath)
require.NoError(t, err, "the port file must exist as soon as startServeHTTP returns")
fields := strings.Fields(string(contents))
require.Len(t, fields, 2, "port file carries '<port> <token>'")
h := &httpHarness{
t: t,
url: "http://127.0.0.1:" + fields[0],
token: fields[1],
done: make(chan struct{}),
}
go func() {
defer close(h.done)
h.rendered = server.run()
}()
req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, h.url+"/stream", nil)
require.NoError(t, err)
req.Header.Set(serveTokenHeader, h.token)
h.stream, err = http.DefaultClient.Do(req) //nolint:bodyclose // closed in t.Cleanup below
require.NoError(t, err)
require.Equal(t, http.StatusOK, h.stream.StatusCode)
h.reader = newRecordReader(h.stream.Body)
t.Cleanup(func() {
_ = h.stream.Body.Close()
select {
case <-h.done:
case <-time.After(2 * time.Second):
t.Error("serve HTTP server did not exit on harness teardown")
}
})
return h
}
// post sends one serveRequest JSON and returns the response status code.
func (h *httpHarness) post(v any) int {
h.t.Helper()
data, err := json.Marshal(v)
require.NoError(h.t, err)
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, h.url+"/", strings.NewReader(string(data)))
require.NoError(h.t, err)
req.Header.Set(serveTokenHeader, h.token)
resp, err := http.DefaultClient.Do(req)
require.NoError(h.t, err)
defer resp.Body.Close()
return resp.StatusCode
}
func TestServeHTTP_RenderStreamsIDPrefixedRecords(t *testing.T) {
h := startHTTPHarness(t)
pwdOne := filepath.Join(t.TempDir(), "first-dir")
pwdTwo := filepath.Join(t.TempDir(), "second-dir")
require.NoError(t, os.Mkdir(pwdOne, 0o755))
require.NoError(t, os.Mkdir(pwdTwo, 0o755))
chdirBackToWD(t)
status := h.post(map[string]any{"command": "render", "id": 1, "shell": "nu", "pwd": pwdOne})
assert.Equal(t, http.StatusOK, status)
records := h.reader.collect(500 * time.Millisecond)
require.NotEmpty(t, records, "render over HTTP must produce records on /stream")
for _, rec := range records {
assert.Equal(t, "1", rec.id)
}
assert.Contains(t, records[0].payload, "first-dir")
// A second render on a fresh POST must reach the same daemon and follow
// the directory change.
status = h.post(map[string]any{"command": "render", "id": 2, "shell": "nu", "pwd": pwdTwo})
assert.Equal(t, http.StatusOK, status)
records = h.reader.collect(500 * time.Millisecond)
require.NotEmpty(t, records, "second render must still reach the daemon")
assert.Contains(t, records[0].payload, "second-dir")
status = h.post(map[string]any{"command": "quit"})
assert.Equal(t, http.StatusOK, status)
select {
case <-h.done:
case <-time.After(2 * time.Second):
t.Fatal("serve HTTP server did not exit after quit")
}
assert.True(t, h.rendered, "at least one render occurred before quit")
}
func TestServeHTTP_RejectsBadToken(t *testing.T) {
h := startHTTPHarness(t)
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, h.url+"/", strings.NewReader(`{"command":"quit"}`))
require.NoError(t, err)
req.Header.Set(serveTokenHeader, "wrong-token")
resp, err := http.DefaultClient.Do(req)
require.NoError(t, err)
defer resp.Body.Close()
assert.Equal(t, http.StatusUnauthorized, resp.StatusCode, "a wrong token must not drive the daemon")
h.post(map[string]any{"command": "quit"})
}
// TestServeHTTP_StreamDisconnectShutsDown validates the lifecycle contract:
// the stream connection breaking (shell gone without a quit) must terminate
// the daemon, exactly like SIGPIPE does on the fifo transport.
func TestServeHTTP_StreamDisconnectShutsDown(t *testing.T) {
h := startHTTPHarness(t)
require.NoError(t, h.stream.Body.Close())
select {
case <-h.done:
case <-time.After(2 * time.Second):
t.Fatal("serve HTTP server did not exit when the stream client disconnected")
}
}
func TestServeHTTP_PortFileRemovedOnExit(t *testing.T) {
t.Setenv("OMP_CACHE_DIR", t.TempDir())
portFilePath := filepath.Join(t.TempDir(), "omp-serve-test.port")
server, err := startServeHTTP(portFilePath)
require.NoError(t, err)
contents, err := os.ReadFile(portFilePath)
require.NoError(t, err)
fields := strings.Fields(string(contents))
require.Len(t, fields, 2)
done := make(chan struct{})
go func() {
defer close(done)
server.run()
}()
req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, fmt.Sprintf("http://127.0.0.1:%s/", fields[0]), strings.NewReader(`{"command":"quit"}`))
require.NoError(t, err)
req.Header.Set(serveTokenHeader, fields[1])
resp, err := http.DefaultClient.Do(req)
require.NoError(t, err)
resp.Body.Close()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("serve HTTP server did not exit after quit")
}
assert.NoFileExists(t, portFilePath, "the daemon must clean up its own port file")
}
src/cli/serve_test.go
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"bufio"
"bytes"
"encoding/json"
"io"
"os"
"path/filepath"
"slices"
@@ -132,7 +133,7 @@ type recordReader struct {
ch chan serveRecord
}
func newRecordReader(r *os.File) *recordReader {
func newRecordReader(r io.Reader) *recordReader {
rr := &recordReader{ch: make(chan serveRecord, 64)}
go func() {
src/shell/nu.go
+176 -13
View File
@@ -57,6 +57,9 @@ func (f Features) Nu() Code {
// init for every Nu user on a release build.
return `$env._omp_serve_in_fifo = null
$env._omp_serve_out_fifo = null
$env._omp_serve_url = null
$env._omp_serve_token = null
$env._omp_serve_port_file = null
$env._omp_serve_reader_jid = null
$env._omp_serve_cycle = 0
$env._omp_serve_failures = 0
@@ -68,6 +71,10 @@ $env._omp_serve_failures = 0
# its direct child (the nested nu wrapper), never the real serve process it
# launched, which is why shutdown goes through the protocol instead.
def --env _omp_serve_stop [] {
if ($env._omp_serve_url? | is-not-empty) {
try { http post --headers [x-omp-token $env._omp_serve_token] --max-time 500ms --content-type application/json $env._omp_serve_url '{"command":"quit"}' | ignore }
}
if ($env._omp_serve_in_fifo? | is-not-empty) {
try { '{"command":"quit"}' + "\n" | save --append --force $env._omp_serve_in_fifo }
}
@@ -80,23 +87,39 @@ def --env _omp_serve_stop [] {
try { rm -f $env._omp_serve_in_fifo $env._omp_serve_out_fifo }
}
if ($env._omp_serve_port_file? | is-not-empty) {
try { rm -f $env._omp_serve_port_file }
}
$env._omp_serve_in_fifo = null
$env._omp_serve_out_fifo = null
$env._omp_serve_url = null
$env._omp_serve_token = null
$env._omp_serve_port_file = null
$env._omp_serve_reader_jid = null
}
# Starts a persistent "oh-my-posh serve" daemon over a pair of named pipes,
# so streamed renders no longer pay a fresh process-spawn cost per prompt.
# Unix only (no mkfifo on Windows) - _omp_serve_render falls back to the
# per-prompt stream on any failure here.
# Starts a persistent "oh-my-posh serve" daemon, so streamed renders no
# longer pay a fresh process-spawn cost per prompt. Two transports: a fifo
# pair on Unix, loopback HTTP on Windows (no mkfifo there, and nu's write
# primitives all fail against Windows named pipes). _omp_serve_render falls
# back to the per-prompt stream on any failure here.
def --env _omp_serve_start [] {
_omp_serve_stop
if ($nu.os-info.name == "windows") {
_omp_serve_start_http
} else {
_omp_serve_start_fifo
}
}
def --env _omp_serve_start_fifo [] {
if (which mkfifo | is-empty) {
return false
}
let base = ($nu.temp-path | path join $"omp-serve-(random uuid)")
let base = ($nu.temp-dir | path join $"omp-serve-(random uuid)")
let in_fifo = $"($base).in"
let out_fifo = $"($base).out"
@@ -169,13 +192,99 @@ def --env _omp_serve_start [] {
true
}
# Windows transport: requests go out as one-shot "http post" calls and
# records come back over a single long-lived "http get /stream" response
# the reader job holds open for the daemon's lifetime - the moral
# equivalent of the fifo pair. The daemon exits when that stream connection
# breaks, which also covers a hard-killed shell.
def --env _omp_serve_start_http [] {
let port_file = ($nu.temp-dir | path join $"omp-serve-(random uuid).port")
let nu_exe = $nu.current-exe
let exe = $_omp_executable
# Same nested-nu wrapper as the fifo transport: redirects only resolve
# correctly at a nu process's own top level, never inside a job-spawn
# closure. NUL is Windows' null device.
job spawn {
^$nu_exe --no-config-file -c $'^($exe) serve --shell=nu --port-file=($port_file) o+e> NUL'
}
# The daemon publishes "<port> <token>" once its listener is up
# (typically well under 100ms). Written atomically (temp + rename), so
# a non-empty read is always complete.
mut endpoint = []
for _ in 1..100 {
if ($port_file | path exists) {
let parts = (try { open --raw $port_file | decode utf8 | str trim | split row " " } catch { [] })
if ($parts | length) >= 2 {
$endpoint = $parts
break
}
}
sleep 20ms
}
if ($endpoint | is-empty) {
return false
}
let url = $"http://127.0.0.1:($endpoint.0)"
let token = $endpoint.1
# Same reader-job contract as the fifo transport (see the reader in
# _omp_serve_start_fifo), with the /stream response as the record
# source. The loop body is duplicated on purpose: routing records
# through a shared def's $in boundary breaks incremental streaming -
# records only surface after the response completes.
let reader_jid = (job spawn {
let self_id = (job id)
mut cycle_started = -1
try {
for rec in (http get --headers [x-omp-token $token] $"($url)/stream" | bytes split 0x[00]) {
if ($rec | is-empty) {
continue
}
let text = ($rec | decode utf8)
let sep = ($text | str index-of (char --integer 0x1f))
if $sep < 0 {
continue
}
let id = ($text | str substring 0..<$sep | into int)
let payload = ($text | str substring ($sep + 1)..)
if ($payload | str starts-with (char --integer 0x1e)) {
continue
}
if $id != $cycle_started {
$cycle_started = $id
{id: $id, payload: $payload} | job send 0 --tag $self_id
} else {
commandline set-prompt $payload
}
}
} catch { }
})
$env._omp_serve_url = $url
$env._omp_serve_token = $token
$env._omp_serve_port_file = $port_file
$env._omp_serve_reader_jid = $reader_jid
true
}
# Renders the primary prompt through the daemon, (re)starting it on demand.
# Returns null on any failure, in which case the caller falls back to the
# per-prompt stream.
def --env _omp_serve_render [
clear: bool
] {
if ($env._omp_serve_in_fifo? | is-empty) and not (_omp_serve_start) {
let started = ($env._omp_serve_in_fifo? | is-not-empty) or ($env._omp_serve_url? | is-not-empty)
if not $started and not (_omp_serve_start) {
return null
}
@@ -198,7 +307,7 @@ def --env _omp_serve_render [
# into; sending it as-is silently drops the whole request (encoding/json
# fails to unmarshal a JSON array into a string value, and serve.go
# ignores malformed lines for forward/backward compatibility).
mut env_rec = {PATH: ($env.PATH | str join (char --integer 0x3a))}
mut env_rec = {PATH: ($env.PATH | str join (char esep))}
for name in ($env | columns | where {|c| $c starts-with "POSH_" }) {
let value = ($env | get $name)
if ($value | describe) == "string" {
@@ -222,10 +331,17 @@ def --env _omp_serve_render [
env: $env_rec
}
let write_ok = (try {
($req | to json --raw) + "\n" | save --append $env._omp_serve_in_fifo
true
} catch { false })
let write_ok = if ($env._omp_serve_url? | is-not-empty) {
(try {
http post --headers [x-omp-token $env._omp_serve_token] --max-time 3sec --content-type application/json $env._omp_serve_url ($req | to json --raw) | ignore
true
} catch { false })
} else {
(try {
($req | to json --raw) + "\n" | save --append $env._omp_serve_in_fifo
true
} catch { false })
}
if not $write_ok {
_omp_serve_stop
@@ -324,6 +440,43 @@ def --env _omp_stream_primary [
}
}
def _omp_spawn_right [] {
let execution_time = match $env.CMD_DURATION_MS {
'0823' => -1
$ms => { $ms | into int }
}
let no_status = if $nu.history-enabled {
not ($env.POSH_EXECUTED? | default false)
} else {
$execution_time < 0
}
let exe = $_omp_executable
let args = [
--shell=nu
$"--shell-version=($env.POSH_SHELL_VERSION)"
$"--status=($env.LAST_EXIT_CODE)"
$"--no-status=($no_status)"
$"--execution-time=($execution_time)"
$"--terminal-width=((term size).columns)"
$"--job-count=(job list | length)"
]
job spawn {
let self_id = (job id)
(^$exe print right --save-cache ...$args) | job send 0 --tag $self_id
}
}
def --env _omp_collect_right [jid: any] {
$env._omp_right_prompt = (try {
job recv --tag $jid --timeout 3sec
} catch {
$env._omp_right_prompt? | default ''
})
}
$env.PROMPT_COMMAND = {||
let hist = if $nu.history-enabled { history } else { [] }
let hist_len = ($hist | length)
@@ -340,8 +493,18 @@ $env.PROMPT_COMMAND = {||
$env.POSH_EXECUTED = ($nu.history-enabled and ($hist_len > ($env.POSH_LAST_HISTORY_LEN? | default 0)))
$env.POSH_LAST_HISTORY_LEN = $hist_len
_omp_stream_primary $clear
}`
# The right prompt renders concurrently with the streamed primary: Nu
# evaluates PROMPT_COMMAND before PROMPT_COMMAND_RIGHT on every prompt
# update, so by the time the primary's first record is back the right
# render (a full process spawn, ~35ms on Windows) has already completed
# in the background instead of adding a second sequential spawn.
let right_jid = (_omp_spawn_right)
let prompt = (_omp_stream_primary $clear)
_omp_collect_right $right_jid
$prompt
}
$env.PROMPT_COMMAND_RIGHT = {|| $env._omp_right_prompt? | default '' }`
case PromptMark, RPrompt, PoshGit, Azure, LineError, Jobs, Tooltips, FTCSMarks, CursorPositioning, Async, KeyHandlers, VIMode:
fallthrough
default:
src/shell/nu_test.go
+176 -13
View File
@@ -16,6 +16,9 @@ $env.TRANSIENT_PROMPT_COMMAND = {|| _omp_get_prompt transient }
^$_omp_executable notice
$env._omp_serve_in_fifo = null
$env._omp_serve_out_fifo = null
$env._omp_serve_url = null
$env._omp_serve_token = null
$env._omp_serve_port_file = null
$env._omp_serve_reader_jid = null
$env._omp_serve_cycle = 0
$env._omp_serve_failures = 0
@@ -27,6 +30,10 @@ $env._omp_serve_failures = 0
# its direct child (the nested nu wrapper), never the real serve process it
# launched, which is why shutdown goes through the protocol instead.
def --env _omp_serve_stop [] {
if ($env._omp_serve_url? | is-not-empty) {
try { http post --headers [x-omp-token $env._omp_serve_token] --max-time 500ms --content-type application/json $env._omp_serve_url '{"command":"quit"}' | ignore }
}
if ($env._omp_serve_in_fifo? | is-not-empty) {
try { '{"command":"quit"}' + "\n" | save --append --force $env._omp_serve_in_fifo }
}
@@ -39,23 +46,39 @@ def --env _omp_serve_stop [] {
try { rm -f $env._omp_serve_in_fifo $env._omp_serve_out_fifo }
}
if ($env._omp_serve_port_file? | is-not-empty) {
try { rm -f $env._omp_serve_port_file }
}
$env._omp_serve_in_fifo = null
$env._omp_serve_out_fifo = null
$env._omp_serve_url = null
$env._omp_serve_token = null
$env._omp_serve_port_file = null
$env._omp_serve_reader_jid = null
}
# Starts a persistent "oh-my-posh serve" daemon over a pair of named pipes,
# so streamed renders no longer pay a fresh process-spawn cost per prompt.
# Unix only (no mkfifo on Windows) - _omp_serve_render falls back to the
# per-prompt stream on any failure here.
# Starts a persistent "oh-my-posh serve" daemon, so streamed renders no
# longer pay a fresh process-spawn cost per prompt. Two transports: a fifo
# pair on Unix, loopback HTTP on Windows (no mkfifo there, and nu's write
# primitives all fail against Windows named pipes). _omp_serve_render falls
# back to the per-prompt stream on any failure here.
def --env _omp_serve_start [] {
_omp_serve_stop
if ($nu.os-info.name == "windows") {
_omp_serve_start_http
} else {
_omp_serve_start_fifo
}
}
def --env _omp_serve_start_fifo [] {
if (which mkfifo | is-empty) {
return false
}
let base = ($nu.temp-path | path join $"omp-serve-(random uuid)")
let base = ($nu.temp-dir | path join $"omp-serve-(random uuid)")
let in_fifo = $"($base).in"
let out_fifo = $"($base).out"
@@ -128,13 +151,99 @@ def --env _omp_serve_start [] {
true
}
# Windows transport: requests go out as one-shot "http post" calls and
# records come back over a single long-lived "http get /stream" response
# the reader job holds open for the daemon's lifetime - the moral
# equivalent of the fifo pair. The daemon exits when that stream connection
# breaks, which also covers a hard-killed shell.
def --env _omp_serve_start_http [] {
let port_file = ($nu.temp-dir | path join $"omp-serve-(random uuid).port")
let nu_exe = $nu.current-exe
let exe = $_omp_executable
# Same nested-nu wrapper as the fifo transport: redirects only resolve
# correctly at a nu process's own top level, never inside a job-spawn
# closure. NUL is Windows' null device.
job spawn {
^$nu_exe --no-config-file -c $'^($exe) serve --shell=nu --port-file=($port_file) o+e> NUL'
}
# The daemon publishes "<port> <token>" once its listener is up
# (typically well under 100ms). Written atomically (temp + rename), so
# a non-empty read is always complete.
mut endpoint = []
for _ in 1..100 {
if ($port_file | path exists) {
let parts = (try { open --raw $port_file | decode utf8 | str trim | split row " " } catch { [] })
if ($parts | length) >= 2 {
$endpoint = $parts
break
}
}
sleep 20ms
}
if ($endpoint | is-empty) {
return false
}
let url = $"http://127.0.0.1:($endpoint.0)"
let token = $endpoint.1
# Same reader-job contract as the fifo transport (see the reader in
# _omp_serve_start_fifo), with the /stream response as the record
# source. The loop body is duplicated on purpose: routing records
# through a shared def's $in boundary breaks incremental streaming -
# records only surface after the response completes.
let reader_jid = (job spawn {
let self_id = (job id)
mut cycle_started = -1
try {
for rec in (http get --headers [x-omp-token $token] $"($url)/stream" | bytes split 0x[00]) {
if ($rec | is-empty) {
continue
}
let text = ($rec | decode utf8)
let sep = ($text | str index-of (char --integer 0x1f))
if $sep < 0 {
continue
}
let id = ($text | str substring 0..<$sep | into int)
let payload = ($text | str substring ($sep + 1)..)
if ($payload | str starts-with (char --integer 0x1e)) {
continue
}
if $id != $cycle_started {
$cycle_started = $id
{id: $id, payload: $payload} | job send 0 --tag $self_id
} else {
commandline set-prompt $payload
}
}
} catch { }
})
$env._omp_serve_url = $url
$env._omp_serve_token = $token
$env._omp_serve_port_file = $port_file
$env._omp_serve_reader_jid = $reader_jid
true
}
# Renders the primary prompt through the daemon, (re)starting it on demand.
# Returns null on any failure, in which case the caller falls back to the
# per-prompt stream.
def --env _omp_serve_render [
clear: bool
] {
if ($env._omp_serve_in_fifo? | is-empty) and not (_omp_serve_start) {
let started = ($env._omp_serve_in_fifo? | is-not-empty) or ($env._omp_serve_url? | is-not-empty)
if not $started and not (_omp_serve_start) {
return null
}
@@ -157,7 +266,7 @@ def --env _omp_serve_render [
# into; sending it as-is silently drops the whole request (encoding/json
# fails to unmarshal a JSON array into a string value, and serve.go
# ignores malformed lines for forward/backward compatibility).
mut env_rec = {PATH: ($env.PATH | str join (char --integer 0x3a))}
mut env_rec = {PATH: ($env.PATH | str join (char esep))}
for name in ($env | columns | where {|c| $c starts-with "POSH_" }) {
let value = ($env | get $name)
if ($value | describe) == "string" {
@@ -181,10 +290,17 @@ def --env _omp_serve_render [
env: $env_rec
}
let write_ok = (try {
($req | to json --raw) + "\n" | save --append $env._omp_serve_in_fifo
true
} catch { false })
let write_ok = if ($env._omp_serve_url? | is-not-empty) {
(try {
http post --headers [x-omp-token $env._omp_serve_token] --max-time 3sec --content-type application/json $env._omp_serve_url ($req | to json --raw) | ignore
true
} catch { false })
} else {
(try {
($req | to json --raw) + "\n" | save --append $env._omp_serve_in_fifo
true
} catch { false })
}
if not $write_ok {
_omp_serve_stop
@@ -283,6 +399,43 @@ def --env _omp_stream_primary [
}
}
def _omp_spawn_right [] {
let execution_time = match $env.CMD_DURATION_MS {
'0823' => -1
$ms => { $ms | into int }
}
let no_status = if $nu.history-enabled {
not ($env.POSH_EXECUTED? | default false)
} else {
$execution_time < 0
}
let exe = $_omp_executable
let args = [
--shell=nu
$"--shell-version=($env.POSH_SHELL_VERSION)"
$"--status=($env.LAST_EXIT_CODE)"
$"--no-status=($no_status)"
$"--execution-time=($execution_time)"
$"--terminal-width=((term size).columns)"
$"--job-count=(job list | length)"
]
job spawn {
let self_id = (job id)
(^$exe print right --save-cache ...$args) | job send 0 --tag $self_id
}
}
def --env _omp_collect_right [jid: any] {
$env._omp_right_prompt = (try {
job recv --tag $jid --timeout 3sec
} catch {
$env._omp_right_prompt? | default ''
})
}
$env.PROMPT_COMMAND = {||
let hist = if $nu.history-enabled { history } else { [] }
let hist_len = ($hist | length)
@@ -299,8 +452,18 @@ $env.PROMPT_COMMAND = {||
$env.POSH_EXECUTED = ($nu.history-enabled and ($hist_len > ($env.POSH_LAST_HISTORY_LEN? | default 0)))
$env.POSH_LAST_HISTORY_LEN = $hist_len
_omp_stream_primary $clear
}`
# The right prompt renders concurrently with the streamed primary: Nu
# evaluates PROMPT_COMMAND before PROMPT_COMMAND_RIGHT on every prompt
# update, so by the time the primary's first record is back the right
# render (a full process spawn, ~35ms on Windows) has already completed
# in the background instead of adding a second sequential spawn.
let right_jid = (_omp_spawn_right)
let prompt = (_omp_stream_primary $clear)
_omp_collect_right $right_jid
$prompt
}
$env.PROMPT_COMMAND_RIGHT = {|| $env._omp_right_prompt? | default '' }`
assert.Equal(t, want, got)
}