feat(dsh): stream background output, sandbox facts, spill, and the runtime reach marker (#796)
* refactor(runtime): replace confined with a three-value reach marker A runtime now declares reach: vfs | process | remote, stating whether the workspace dispatch gate is its only door or the code can act around it (host process doors, another machine). The default is process, the no-promise claim, so a custom runtime must narrow its reach explicitly. The dsh sandbox claim reads the aggregate: every runtime at vfs means workspace-write, anything wider means no claim. Mirrored in Python (RuntimeReach in runtime/types.py) with reach declared on monty, quickjs, wasi, vfs, local, and RemoteSandbox. * fix(pyodide): seal the js module so the guest has no host door Pyodide's default exposes the host globalThis as the `js` module, which under Node handed guest code js.process (host env, confirmed reading HOME) and js.fetch (network) — doors around the workspace bridge that made the runtime's reach='vfs' claim false and contradicted its own 'no network' docstring. Pass a null-prototype jsglobals to loadPyodide: `import js` still resolves but the host globals are unreachable through it, while pyodide's internals (which capture their globals at load time) and the FS bridge are unaffected. Pinned by jsglobals.test.ts. * feat(dsh): stamp ShellSandboxInfo on run results and process handles When the world is fully workspace-bound (vfsOnly), run() and start() now fill dsh's optional sandbox field: mode workspace-write, enforcement 'full' (the VFS gate is unbypassable, so unlike an OS sandbox on an old kernel there is no promised effect it fails to govern), denied false (mirage has no out-of-band denial channel; a refused write fails in-band as an ordinary command error), and runnerFailed false (the executor is the runner). The process handle stamps it on settle. Omitted when any runtime reaches beyond the workspace. * feat(dsh): stream background command output through a JobConsole Adds a public ExecuteOptions.sink: pass a JobConsole and the line's output streams into it as each statement finishes, instead of being returned whole (the result then carries only the exit code). This reuses the executor's existing internal sink mechanism, so a compound line flushes per statement and stdout/stderr keep their channels. MirageShellProcess is rewritten over that seam: start() runs the command with a console as its sink and a follow loop drains it into the read buffer, so readOutput() delivers output incrementally and stdout/stderr interleave in order (stderr opened by a marker) rather than stderr being concatenated at the end. The unread backlog is bounded to stdoutMaxBytes (tail kept, lossy flagged) so a reader that never drains cannot grow it without limit. * feat(dsh): spill the full stream to a workspace file on overrun Adds an opt-in spillDir config. When a background command's streamed output overruns its delta budget, the full stdout and stderr are written to files under that workspace directory and readOutput() points at them (stdoutSpillPath/stderrSpillPath), so a reader can recover what the delta dropped by reading the spill through the same VFS. Memory stays bounded: each channel buffers only until the first overrun, then flushes to its file and appends from there. A write failure (no writable mount at the path) disables the sink and leaves the paths undefined, the honest 'no safe path' answer. Default unset, so nothing spills unless a deployment asks for it. * docs(dsh): custom backends, background streaming, and the reach model Corrects the sandbox-claim wording to the reach model (workspace-write when every runtime reaches only the vfs, dropped when one reaches the host), and adds a Custom backends section (registerResourceFactory, host-side before the workspace builds) and a Background commands section (per-statement streaming, bounded backlog, spillDir). * test(dsh,core): satisfy lint on the streaming tests Narrow spill paths with an explicit guard instead of a non-null assertion (forbidden in the dsh package), and drop the now-unnecessary ExecuteResult casts the sink overload already implies. * style: prettier formatting on the streaming changes * fix(core): drain a buffered line into the sink A sink only saw output the command-tree walk emitted, so a whole-line runtime, the syntax gate, a policy denial and a failed line all answered with bytes in hand that a streaming caller never read. executeLine now moves any buffered result into the console on every path, in one place rather than five, and the result stays empty as it already did when the line streamed. * fix(dsh): bound the console store, make the spill dir idempotent Capping the delta did not bound memory: reading a chunk advances a cursor but frees nothing, so an uncapped store held every chunk of a noisy background command for the life of the process. The store now carries a retention budget, and the drain reports a trimmed chunk as lossy and stops the spill, since a file missing the middle of a stream is worse than no file. The spill directory is created through ensureDirPath, which walks the ancestors and accepts a refusal for a directory that now exists, so two commands overrunning at once do not cost the loser its spill.
This commit is contained in:
Zecheng Zhang
@@ -91,7 +91,30 @@ The bundle's default world is one RAM scratch mount at `/tmp`. Mount real resour
|
||||
- { name: monty, captures: [python, python3] }
|
||||
```
|
||||
|
||||
The same blocks work in code, beside live instances, in `MirageService`'s `mounts`. The shell executor reports `workspace-write` confinement, since a command cannot reach anything but the mounts, which is what lets dsh's permission presets compose over it.
|
||||
The same blocks work in code, beside live instances, in `MirageService`'s `mounts`. The shell executor reports a `workspace-write` sandbox to dsh whenever every runtime in the world stays inside the VFS (each runtime's `reach` is `vfs`), which is what lets dsh's permission presets compose over it: a command cannot then reach anything but the mounts, under their modes. Adding a host-reaching runtime such as `local` python drops the claim, since a script could act outside the mounts, and dsh is told there is no sandbox rather than a false one.
|
||||
|
||||
## Custom backends
|
||||
|
||||
A mount is not limited to the builtin resources (`ram`, `s3`, `slack`, `redis`, ...). Register your own resource factory host-side and its name becomes usable in a `mounts` block exactly like a builtin:
|
||||
|
||||
```ts
|
||||
import { registerResourceFactory } from '@struktoai/mirage-node'
|
||||
|
||||
registerResourceFactory('acme', (config) => new AcmeResource(config))
|
||||
```
|
||||
|
||||
```yaml
|
||||
- id: mirage
|
||||
config:
|
||||
mounts:
|
||||
/acme: { resource: acme, mode: read, config: { token: !!js process.env.ACME_TOKEN } }
|
||||
```
|
||||
|
||||
The registration must run before the workspace builds, since a `mounts` block only names a resource, it does not construct one. In a dsh bundle that means a small plugin the profile loads alongside `@struktoai/mirage-dsh` (or a plain import in code, before `MirageService` starts), not the declarative patch. A builtin name cannot be shadowed, so a custom backend needs its own name. Nothing else changes: the shell, the commands, and the sandbox claim treat a custom mount like any other.
|
||||
|
||||
## Background commands
|
||||
|
||||
A backgrounded command streams its output through a workspace console rather than arriving whole: each statement of a compound line lands as it finishes, and stdout and stderr keep their own channels. The unread backlog is bounded to the stdout budget, so a reader that never drains cannot grow it without limit; output past the budget is flagged lossy, with the freshest kept. Set `spillDir` on `MirageShellExecutor` to a workspace path (for example `/tmp` on a ram mount) and an overrunning command's full stdout and stderr are written there, so the agent can read the complete output back through the same VFS; unset, nothing spills.
|
||||
|
||||
A concrete scenario: serving agents from a TypeScript server (an Express endpoint, say), where every request is assigned its own context and workspace, created on entry and gone with the response. What one 2 vCPU / 8 GB box holds then depends on the harness:
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ from collections.abc import Sequence
|
||||
from typing import Any, Callable, ClassVar
|
||||
|
||||
from mirage.runtime.config import RuntimeConfig
|
||||
from mirage.runtime.types import ScriptSource
|
||||
from mirage.runtime.types import RuntimeReach, ScriptSource
|
||||
|
||||
|
||||
class Runtime(ABC):
|
||||
@@ -40,6 +40,17 @@ class Runtime(ABC):
|
||||
|
||||
name: str
|
||||
captures: tuple[str, ...] = ()
|
||||
# Which doors this runtime's code has to the outside world (see
|
||||
# RuntimeReach): "vfs" when the workspace dispatch is its only
|
||||
# one, as the bridged engines (monty, quickjs, wasi) and the vfs
|
||||
# routing marker declare, "process" or "remote" when the code can
|
||||
# act around that gate. The default is "process", the no-promise
|
||||
# claim, so a custom runtime must declare a narrower reach
|
||||
# explicitly rather than inherit it. Embedders read the aggregate:
|
||||
# only a world in which every runtime reaches "vfs" makes "agent
|
||||
# code cannot bypass mount modes and policy" a true statement; one
|
||||
# wider runtime voids it.
|
||||
reach: RuntimeReach = "process"
|
||||
# Per-line admission script for the routing ladder, answering "do
|
||||
# I want this line": a callable taking a PolicyContext, or a
|
||||
# config-borne ScriptSource. None = always willing. Policy, not
|
||||
|
||||
@@ -25,7 +25,7 @@ from mirage.runtime.js.base import JsRuntime
|
||||
from mirage.runtime.mixin import EvaluatorMixin
|
||||
from mirage.runtime.resolver import MountResolver
|
||||
from mirage.runtime.types import (DispatchFn, EvalResult, EvalValue, RunArgs,
|
||||
RunResult, ScriptSource)
|
||||
RunResult, RuntimeReach, ScriptSource)
|
||||
from mirage.runtime.vfs import RuntimeVFS
|
||||
from mirage.runtime.wasm import WasmRuntime, WasmVFS
|
||||
|
||||
@@ -100,6 +100,9 @@ class QuickJsRuntime(JsRuntime, EvaluatorMixin):
|
||||
"""
|
||||
|
||||
name = "quickjs"
|
||||
# The engine is a WASI guest whose `std.open`/`os.readdir` suspend
|
||||
# into the workspace bridge: guest I/O has no door around the gate.
|
||||
reach: RuntimeReach = "vfs"
|
||||
|
||||
config_cls: ClassVar[type[RuntimeConfig]] = HomeConfig
|
||||
config: HomeConfig
|
||||
|
||||
@@ -23,7 +23,7 @@ from mirage.runtime.config import HomeConfig, RuntimeConfig
|
||||
from mirage.runtime.python.base import PythonRuntime
|
||||
from mirage.runtime.python.bootstrap import bootstrap
|
||||
from mirage.runtime.python.flags import init_argv
|
||||
from mirage.runtime.types import RunArgs, RunResult, ScriptSource
|
||||
from mirage.runtime.types import RunArgs, RunResult, RuntimeReach, ScriptSource
|
||||
|
||||
LOCAL_HOME_ENV = "MIRAGE_LOCAL_HOME"
|
||||
|
||||
@@ -42,6 +42,11 @@ class LocalRuntime(PythonRuntime):
|
||||
"""
|
||||
|
||||
name = "local"
|
||||
# Spawns the host interpreter: a real process with the user's own
|
||||
# filesystem and network, doors the workspace gate never sees.
|
||||
# This is the base default; declared here so the claim is explicit
|
||||
# at the one builtin runtime that voids a world's sandbox claim.
|
||||
reach: RuntimeReach = "process"
|
||||
|
||||
config_cls: ClassVar[type[RuntimeConfig]] = HomeConfig
|
||||
config: HomeConfig
|
||||
|
||||
@@ -33,7 +33,7 @@ from mirage.runtime.python.monty.constants import (DEFAULT_PROG,
|
||||
from mirage.runtime.python.monty.osaccess import MirageOSAccess
|
||||
from mirage.runtime.resolver import MountResolver
|
||||
from mirage.runtime.types import (DispatchFn, EvalResult, EvalValue, RunArgs,
|
||||
RunResult, ScriptSource)
|
||||
RunResult, RuntimeReach, ScriptSource)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -55,6 +55,11 @@ class MontyRuntime(PythonRuntime, EvaluatorMixin):
|
||||
"""
|
||||
|
||||
name = "monty"
|
||||
# The pooled worker subprocess exists for crash isolation, not
|
||||
# host access: the interpreter inside it has no host filesystem,
|
||||
# environment, or network door, and its file I/O is serviced only
|
||||
# through the workspace dispatch, so nothing goes around the gate.
|
||||
reach: RuntimeReach = "vfs"
|
||||
# No import system to resolve a module with, so `-m` has nothing to
|
||||
# run; the refusal names this runtime rather than inventing a
|
||||
# "No module named" that would imply a search happened.
|
||||
|
||||
@@ -23,7 +23,8 @@ from mirage.runtime.python.base import PythonRuntime
|
||||
from mirage.runtime.python.bootstrap import bootstrap
|
||||
from mirage.runtime.python.flags import init_argv
|
||||
from mirage.runtime.resolver import MountResolver
|
||||
from mirage.runtime.types import DispatchFn, RunArgs, RunResult, ScriptSource
|
||||
from mirage.runtime.types import (DispatchFn, RunArgs, RunResult, RuntimeReach,
|
||||
ScriptSource)
|
||||
from mirage.runtime.vfs import RuntimeVFS
|
||||
from mirage.runtime.wasm import WasmFsConfig, WasmRuntime, WasmVFS
|
||||
|
||||
@@ -71,6 +72,11 @@ class WasiRuntime(PythonRuntime):
|
||||
"""
|
||||
|
||||
name = "wasi"
|
||||
# Guest file I/O can only travel the workspace bridge; the one
|
||||
# host surface is the interpreter's own build directory, served
|
||||
# read-only (mutations raise PermissionError in WasmVFS), so
|
||||
# nothing goes around the gate.
|
||||
reach: RuntimeReach = "vfs"
|
||||
|
||||
config_cls: ClassVar[type[RuntimeConfig]] = HomeConfig
|
||||
config: HomeConfig
|
||||
|
||||
@@ -20,7 +20,7 @@ from mirage.runtime.base import Runtime
|
||||
from mirage.runtime.mixin import LineExecutorMixin
|
||||
from mirage.runtime.policy.types import PolicyScript
|
||||
from mirage.runtime.sandbox.config import SandboxConfig
|
||||
from mirage.runtime.types import RunResult
|
||||
from mirage.runtime.types import RunResult, RuntimeReach
|
||||
|
||||
|
||||
class RemoteSandbox(Runtime, LineExecutorMixin):
|
||||
@@ -42,6 +42,11 @@ class RemoteSandbox(Runtime, LineExecutorMixin):
|
||||
provider's own config class.
|
||||
"""
|
||||
|
||||
# Lines execute on the provider's machine and act on that
|
||||
# machine's world (its filesystem, its network); the workspace
|
||||
# gate never sees those effects, however well isolated the
|
||||
# sandbox itself is.
|
||||
reach: RuntimeReach = "remote"
|
||||
captures: tuple[str, ...] = ("*", )
|
||||
config_cls: ClassVar[type[SandboxConfig]] = SandboxConfig
|
||||
config: SandboxConfig
|
||||
|
||||
@@ -23,7 +23,7 @@ from mirage.runtime.mixin import LineExecutorMixin
|
||||
from mirage.runtime.python.local import LocalRuntime
|
||||
from mirage.runtime.python.monty import MontyRuntime
|
||||
from mirage.runtime.python.wasi import WasiRuntime
|
||||
from mirage.runtime.types import ScriptSource
|
||||
from mirage.runtime.types import RuntimeReach, ScriptSource
|
||||
|
||||
# One source of truth, preference order (sandboxed first, host last).
|
||||
# The command -> runtime mapping is derived from each class's captures,
|
||||
@@ -55,6 +55,9 @@ class VFSRuntime(Runtime):
|
||||
"""
|
||||
|
||||
name = "vfs"
|
||||
# A vfs-routed line runs on the workspace executor itself: it IS
|
||||
# the gate, so there is no door around it.
|
||||
reach: RuntimeReach = "vfs"
|
||||
captures: tuple[str, ...] = ()
|
||||
|
||||
def __init__(
|
||||
|
||||
@@ -34,6 +34,24 @@ EvalStatus: TypeAlias = Literal["complete", "incomplete", "exit"]
|
||||
# selector that silently matches nothing and reports "no runtime".
|
||||
Language: TypeAlias = Literal["python", "js"]
|
||||
|
||||
# Which doors code executed by a runtime has to the outside world. The
|
||||
# workspace dispatch is a gate: it checks mount modes, session grants,
|
||||
# and policy, records the op, and only then touches the real backend
|
||||
# behind the mount (s3, disk, an API). Reach states whether that gate
|
||||
# is avoidable, not where bytes physically end up; a "vfs" write to an
|
||||
# s3 mount still lands in real s3, but only after the gate said yes.
|
||||
# - "vfs": the gate is the code's only door. The engine runs as an
|
||||
# in-process guest with no syscalls, so its I/O can only travel the
|
||||
# VFS bridge (or the workspace executor itself) and a mount-mode or
|
||||
# policy refusal is final.
|
||||
# - "process": the code has host doors around the gate. It is, or
|
||||
# spawns, a real process on this machine with the user's own
|
||||
# filesystem and network, so it can reach the same backends (and
|
||||
# everything else) without the gate seeing it.
|
||||
# - "remote": the code runs on another machine and acts on that
|
||||
# machine's world; the gate never sees those effects.
|
||||
RuntimeReach: TypeAlias = Literal["vfs", "process", "remote"]
|
||||
|
||||
|
||||
class DispatchFn(Protocol):
|
||||
"""The workspace op dispatch: run ``op`` against the mount owning
|
||||
|
||||
@@ -283,3 +283,7 @@ async def test_eval_failures_raise_eval_error():
|
||||
await rt.eval("throw new Error('boom')")
|
||||
with pytest.raises(EvalError, match="one-shot"):
|
||||
await rt.eval("1", session="s1")
|
||||
|
||||
|
||||
def test_reach_is_vfs():
|
||||
assert QuickJsRuntime.reach == "vfs"
|
||||
|
||||
@@ -324,3 +324,7 @@ async def test_monty_cancelled_eval_session_releases_its_checkout(monkeypatch):
|
||||
result = await runtime.eval("1 + 1", session="s1")
|
||||
assert result.value == 2
|
||||
await runtime.close()
|
||||
|
||||
|
||||
def test_reach_is_vfs():
|
||||
assert MontyRuntime.reach == "vfs"
|
||||
|
||||
@@ -77,3 +77,10 @@ async def test_local_cancellation_kills_subprocess():
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
assert time.monotonic() - start < 5 # killed, not waited out
|
||||
|
||||
|
||||
def test_reach_is_process():
|
||||
# The subprocess sees the host filesystem and network: doors the
|
||||
# workspace gate never sees, so a world holding this runtime may
|
||||
# not claim a sandbox.
|
||||
assert LocalRuntime.reach == "process"
|
||||
|
||||
@@ -234,3 +234,7 @@ async def test_wasi_session_narrowing_reaches_the_guest():
|
||||
r = await ws.execute(f'python3 -c "{code}"')
|
||||
assert (await r.stdout_str()) == ""
|
||||
await ws.close()
|
||||
|
||||
|
||||
def test_reach_is_vfs():
|
||||
assert WasiRuntime.reach == "vfs"
|
||||
|
||||
@@ -190,3 +190,7 @@ async def test_remote_line_invalidates_local_read_caches():
|
||||
assert looked.entry is None
|
||||
finally:
|
||||
await ws.close()
|
||||
|
||||
|
||||
def test_reach_is_remote():
|
||||
assert RemoteSandbox.reach == "remote"
|
||||
|
||||
@@ -67,3 +67,10 @@ def test_script_stored():
|
||||
def test_unknown_config_key_fails_loud():
|
||||
with pytest.raises(TypeError):
|
||||
MarkerRuntime(config={"no_such_knob": 1})
|
||||
|
||||
|
||||
def test_reach_defaults_to_process():
|
||||
# A runtime that declares nothing gets the no-promise claim: it
|
||||
# may act around the workspace gate. Only an explicit "vfs"
|
||||
# narrows it.
|
||||
assert MarkerRuntime().reach == "process"
|
||||
|
||||
@@ -143,3 +143,14 @@ def test_vfs_is_a_pure_routing_marker():
|
||||
assert not isinstance(vfs, LineExecutorMixin)
|
||||
assert not hasattr(vfs, "run_line")
|
||||
assert not hasattr(vfs, "run")
|
||||
|
||||
|
||||
def test_vfs_marker_reach_is_vfs():
|
||||
assert VFSRuntime.reach == "vfs"
|
||||
|
||||
|
||||
def test_default_world_reaches_only_the_vfs():
|
||||
# The default world's sandbox story rests on every entry keeping
|
||||
# its effects behind the workspace gate; `local` (process reach)
|
||||
# is deliberately not a default entry.
|
||||
assert all(NAMED[name].reach == "vfs" for name in DEFAULT_ENTRIES)
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
import { coerceRuntimeConfig, type RuntimeConfig } from './config.ts'
|
||||
import { ScriptSource, type PolicyScript } from './policy/types.ts'
|
||||
import type { RuntimeOptions } from './types.ts'
|
||||
import type { RuntimeOptions, RuntimeReach } from './types.ts'
|
||||
|
||||
/**
|
||||
* An engine the workspace can route commands or whole lines to.
|
||||
@@ -42,14 +42,19 @@ export abstract class Runtime {
|
||||
abstract readonly name: string
|
||||
readonly captures: readonly string[]
|
||||
/**
|
||||
* Whether every effect of code this runtime executes lands inside the
|
||||
* workspace. The bridged engines (monty, pyodide, quickjs) and the vfs
|
||||
* routing marker declare true: their I/O rides the VFS bridge or the
|
||||
* workspace executor itself. False by default and for anything that
|
||||
* executes elsewhere — the host `local` python, a RemoteSandbox — so a
|
||||
* custom runtime must declare confinement rather than inherit it.
|
||||
* Which doors this runtime's code has to the outside world (see
|
||||
* RuntimeReach): 'vfs' when the workspace dispatch is its only one,
|
||||
* as the bridged engines (monty, pyodide, quickjs) and the vfs
|
||||
* routing marker declare, 'process' or 'remote' when the code can
|
||||
* act around that gate. The default is 'process', the no-promise
|
||||
* claim, so a custom runtime must declare a narrower reach
|
||||
* explicitly rather than inherit it. Embedders read the aggregate:
|
||||
* only a world in which every runtime reaches 'vfs' makes "agent
|
||||
* code cannot bypass mount modes and policy" a true statement,
|
||||
* which is what the dsh adapter's sandbox claim is built from; one
|
||||
* wider runtime voids it.
|
||||
*/
|
||||
readonly confined: boolean = false
|
||||
readonly reach: RuntimeReach = 'process'
|
||||
/** The runtime's coerced implementation knobs. */
|
||||
config: RuntimeConfig
|
||||
script?: PolicyScript
|
||||
|
||||
@@ -161,7 +161,9 @@ globalThis.std = {
|
||||
// matching the Python runtime's live file I/O.
|
||||
export class QuickJsRuntime extends JsRuntime implements Evaluator {
|
||||
readonly name = 'quickjs'
|
||||
override readonly confined = true
|
||||
// The engine is a WASI guest whose `std.open`/`os.readdir` suspend
|
||||
// into the workspace bridge: guest I/O has no door around the gate.
|
||||
override readonly reach = 'vfs'
|
||||
readonly [EVALUATOR] = true as const
|
||||
private newAsyncModule: NewAsyncModule | null = null
|
||||
private workspaceBridge: BridgeDispatchFn | null = null
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { PyodideRuntime } from './pyodide.ts'
|
||||
|
||||
// The runtime declares reach='vfs', meaning the workspace bridge is the
|
||||
// guest's only door to the outside. Pyodide's default exposes the host
|
||||
// globalThis as the `js` module, which under Node hands the guest
|
||||
// js.process (host env) and js.fetch (network) — doors around the
|
||||
// bridge. loader.ts seals that with a null-prototype jsglobals; this
|
||||
// pins the seal so a future edit that drops it fails loudly.
|
||||
describe('PyodideRuntime js-module door', () => {
|
||||
it('the guest cannot reach host process or network through js', async () => {
|
||||
const rt = new PyodideRuntime()
|
||||
const probe = `
|
||||
import js
|
||||
flags = []
|
||||
flags.append("process=" + str(hasattr(js, "process")))
|
||||
flags.append("fetch=" + str(hasattr(js, "fetch")))
|
||||
flags.append("require=" + str(hasattr(js, "require")))
|
||||
print(",".join(flags))
|
||||
`
|
||||
const result = await rt.run({
|
||||
code: probe,
|
||||
args: [],
|
||||
env: {},
|
||||
stdin: new Uint8Array(),
|
||||
})
|
||||
const decode = (b: Uint8Array | string | null): string =>
|
||||
typeof b === 'string' ? b : new TextDecoder().decode(b ?? new Uint8Array())
|
||||
expect(decode(result.stderr)).toBe('')
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(decode(result.stdout).trim()).toBe('process=False,fetch=False,require=False')
|
||||
await rt.close()
|
||||
}, 60_000)
|
||||
})
|
||||
@@ -103,7 +103,18 @@ export async function loadPyodideRuntime(
|
||||
envHome() ??
|
||||
(await resolveNodeIndexURL()) ??
|
||||
(isNode() ? null : PYODIDE_CDN_URL)
|
||||
const opts: Record<string, unknown> = { stdout: noopIo, stderr: noopIo }
|
||||
// A null-prototype jsglobals seals the guest's `js` module: `import
|
||||
// js` still resolves, but the host globalThis (js.process, js.fetch,
|
||||
// js.process.env) is not reachable through it, so guest code has no
|
||||
// host-environment or network door around the workspace bridge. This
|
||||
// is what makes the runtime's reach='vfs' claim true and matches the
|
||||
// docstring's "no network" promise; pyodide's own internals capture
|
||||
// the globals they need at load time, not through this object.
|
||||
const opts: Record<string, unknown> = {
|
||||
stdout: noopIo,
|
||||
stderr: noopIo,
|
||||
jsglobals: Object.create(null) as object,
|
||||
}
|
||||
if (options.packageBaseUrl !== undefined) opts.packageBaseUrl = options.packageBaseUrl
|
||||
if (options.lockFileURL !== undefined) opts.lockFileURL = options.lockFileURL
|
||||
if (options.packages !== undefined && options.packages.length > 0) {
|
||||
|
||||
@@ -502,7 +502,7 @@ describe('monty unavailable', () => {
|
||||
name: 'monty',
|
||||
captures: ['python3', 'python'],
|
||||
language: 'python' as const,
|
||||
confined: true,
|
||||
reach: 'vfs' as const,
|
||||
config: {},
|
||||
attach: () => undefined,
|
||||
run: () => Promise.reject(new MontyUnavailableError('install @pydantic/monty')),
|
||||
|
||||
@@ -95,7 +95,10 @@ function toEvalValue(value: unknown): EvalValue {
|
||||
*/
|
||||
export class MontyRuntime extends PythonRuntime implements Evaluator {
|
||||
readonly name = 'monty'
|
||||
override readonly confined = true
|
||||
// The interpreter is an in-process guest with no host syscalls: its
|
||||
// file I/O can only travel the VFS bridge, so every effect passes
|
||||
// the workspace gate (mount modes, policy, recording).
|
||||
override readonly reach = 'vfs'
|
||||
// No import system to resolve a module with, so `-m` has nothing to
|
||||
// run; the refusal names this runtime rather than inventing a
|
||||
// "No module named" that would imply a search happened.
|
||||
|
||||
@@ -257,7 +257,12 @@ const EVAL_INTERRUPT_SECONDS = 10
|
||||
|
||||
export class PyodideRuntime extends PythonRuntime implements Evaluator {
|
||||
readonly name = 'pyodide'
|
||||
override readonly confined = true
|
||||
// The WASM guest's filesystem is the workspace-backed Emscripten FS,
|
||||
// so file effects pass the workspace gate, and loader.ts seals the
|
||||
// `js` module (null-prototype jsglobals) so guest code cannot reach
|
||||
// js.process or js.fetch either. Both doors closed is what makes this
|
||||
// 'vfs'; jsglobals.test.ts pins the seal.
|
||||
override readonly reach = 'vfs'
|
||||
readonly [EVALUATOR] = true as const
|
||||
private pyodide: PyodideInterface | null = null
|
||||
private initPromise: Promise<PyodideInterface> | null = null
|
||||
|
||||
@@ -38,6 +38,10 @@ export abstract class RemoteSandbox<C extends SandboxConfig = SandboxConfig>
|
||||
extends Runtime
|
||||
implements LineExecutor
|
||||
{
|
||||
// Lines execute on the provider's machine and act on that machine's
|
||||
// world (its filesystem, its network); the workspace gate never sees
|
||||
// those effects, however well isolated the sandbox itself is.
|
||||
override readonly reach = 'remote'
|
||||
readonly [LINE_EXECUTOR] = true as const
|
||||
declare config: NormalizedSandboxConfig<C>
|
||||
// Connect-once latch: the first captured line connects; later lines
|
||||
|
||||
@@ -42,7 +42,9 @@ import type { RuntimeOptions } from './types.ts'
|
||||
*/
|
||||
export class VFSRuntime extends Runtime {
|
||||
readonly name = 'vfs'
|
||||
override readonly confined = true
|
||||
// A vfs-routed line runs on the workspace executor itself: it IS the
|
||||
// gate, so there is no door around it.
|
||||
override readonly reach = 'vfs'
|
||||
// Declaring captures (even empty) turns the catch-all off; the
|
||||
// dispatcher reads this bit, not the array's length.
|
||||
readonly restricted: boolean
|
||||
|
||||
@@ -24,6 +24,29 @@ import type { PolicyScript } from './policy/types.ts'
|
||||
*/
|
||||
export type RuntimeLanguage = 'python' | 'js'
|
||||
|
||||
/**
|
||||
* Which doors code executed by a runtime has to the outside world.
|
||||
*
|
||||
* The workspace dispatch is a gate: it checks mount modes, session
|
||||
* grants, and policy, records the op, and only then touches the real
|
||||
* backend behind the mount (S3, disk, an API). Reach states whether
|
||||
* that gate is avoidable, not where bytes physically end up; a 'vfs'
|
||||
* write to an S3 mount still lands in real S3, but only after the
|
||||
* gate said yes.
|
||||
*
|
||||
* - 'vfs': the gate is the code's only door. The engine runs as an
|
||||
* in-process guest with no syscalls, so its I/O can only travel the
|
||||
* VFS bridge (or the workspace executor itself) and a mount-mode or
|
||||
* policy refusal is final.
|
||||
* - 'process': the code has host doors around the gate. It is, or
|
||||
* spawns, a real process on this machine with the user's own
|
||||
* filesystem and network, so it can reach the same backends (and
|
||||
* everything else) without the gate seeing it.
|
||||
* - 'remote': the code runs on another machine and acts on that
|
||||
* machine's world; the gate never sees those effects.
|
||||
*/
|
||||
export type RuntimeReach = 'vfs' | 'process' | 'remote'
|
||||
|
||||
/**
|
||||
* The workspace op dispatch: run `op` against the mount owning `path`
|
||||
* and return its result with the accounting IOResult. Defined here, on
|
||||
|
||||
@@ -16,10 +16,13 @@ import { describe, expect, it } from 'vitest'
|
||||
import { OpsRegistry } from '../ops/registry.ts'
|
||||
import { RAMResource } from '../resource/ram/ram.ts'
|
||||
import { MountMode } from '../types.ts'
|
||||
import { Channel, JobConsole } from '../shell/console/index.ts'
|
||||
import { getTestParser, stdoutStr } from './fixtures/workspace_fixture.ts'
|
||||
import type { ExecuteResult } from './workspace.ts'
|
||||
import { Workspace } from './workspace.ts'
|
||||
|
||||
const DEC = new TextDecoder()
|
||||
|
||||
const ENC = new TextEncoder()
|
||||
|
||||
async function makeWs(): Promise<Workspace> {
|
||||
@@ -345,3 +348,60 @@ describe('execute(): agent harness pattern', () => {
|
||||
await ws.close()
|
||||
})
|
||||
})
|
||||
|
||||
describe('execute({ sink }): streaming output to a console', () => {
|
||||
it('streams the output to the console and returns empty stdout', async () => {
|
||||
const ws = await makeWs()
|
||||
const console_ = new JobConsole()
|
||||
const result = await ws.execute('echo hello', { sink: console_ })
|
||||
// The bytes went to the console, so the result carries only the code.
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(stdoutStr(result)).toBe('')
|
||||
const streamed = DEC.decode(await console_.snapshot(Channel.STDOUT))
|
||||
expect(streamed.trim()).toBe('hello')
|
||||
await ws.close()
|
||||
})
|
||||
|
||||
it('emits each statement of a compound line as its own chunk', async () => {
|
||||
const ws = await makeWs()
|
||||
const console_ = new JobConsole()
|
||||
await ws.execute('echo a; echo b; echo c', { sink: console_ })
|
||||
const [chunks] = await console_.readFrom(0)
|
||||
const stdout = chunks.filter((c) => c.channel === Channel.STDOUT)
|
||||
expect(stdout.length).toBe(3)
|
||||
expect(stdout.map((c) => DEC.decode(c.data).trim())).toEqual(['a', 'b', 'c'])
|
||||
await ws.close()
|
||||
})
|
||||
|
||||
it('routes stderr to the console on its own channel', async () => {
|
||||
const ws = await makeWs()
|
||||
const console_ = new JobConsole()
|
||||
const result = await ws.execute('echo oops >&2', { sink: console_ })
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(DEC.decode(await console_.snapshot(Channel.STDERR)).trim()).toBe('oops')
|
||||
await ws.close()
|
||||
})
|
||||
|
||||
it('sends a syntax error to the console, not the result', async () => {
|
||||
const ws = await makeWs()
|
||||
const console_ = new JobConsole()
|
||||
// The syntax gate answers before the walk that emits, so this is
|
||||
// output the console would never see without the drain.
|
||||
const result = await ws.execute('case x', { sink: console_ })
|
||||
expect(result.exitCode).toBe(2)
|
||||
expect(stdoutStr(result)).toBe('')
|
||||
expect(DEC.decode(result.stderr)).toBe('')
|
||||
expect(DEC.decode(await console_.snapshot(Channel.STDERR))).toContain('syntax error')
|
||||
await ws.close()
|
||||
})
|
||||
|
||||
it("sends a failed command's stderr to the console", async () => {
|
||||
const ws = await makeWs()
|
||||
const console_ = new JobConsole()
|
||||
const result = await ws.execute('cat /ram/missing.txt', { sink: console_ })
|
||||
expect(result.exitCode).not.toBe(0)
|
||||
expect(DEC.decode(result.stderr)).toBe('')
|
||||
expect(DEC.decode(await console_.snapshot(Channel.STDERR))).toContain('missing.txt')
|
||||
await ws.close()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -36,6 +36,7 @@ import {
|
||||
ScriptSource,
|
||||
} from '../runtime/policy/index.ts'
|
||||
import { getTestParser } from './fixtures/workspace_fixture.ts'
|
||||
import { Channel, JobConsole } from '../shell/console/index.ts'
|
||||
import { RAMResource } from '../resource/ram/ram.ts'
|
||||
import { MountMode } from '../types.ts'
|
||||
import { Workspace } from './workspace.ts'
|
||||
@@ -748,6 +749,27 @@ describe('whole-line runtimes', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('a sink receives the line the runtime served', async () => {
|
||||
const parser = await getTestParser()
|
||||
const box = new LineBox()
|
||||
box.captures = ['*']
|
||||
const ws = new Workspace(
|
||||
{ '/': new RAMResource() },
|
||||
{ mode: MountMode.EXEC, shellParser: parser, runtimes: [box, 'vfs'] },
|
||||
)
|
||||
const console_ = new JobConsole()
|
||||
try {
|
||||
// The runtime buffers and never touches the sink itself, so
|
||||
// without the drain a streaming caller reads nothing at all.
|
||||
const result = await ws.execute('echo hi', { sink: console_ })
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(DEC.decode(result.stdout)).toBe('')
|
||||
expect(DEC.decode(await console_.snapshot(Channel.STDOUT))).toBe('box:echo hi')
|
||||
} finally {
|
||||
await ws.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('the vfs entry is a pure routing marker', async () => {
|
||||
// A vfs-resolved line runs on the workspace executor inline; the
|
||||
// entry is a marker with no line door to call.
|
||||
|
||||
@@ -17,6 +17,8 @@ import { IOResult, materialize } from '../../io/types.ts'
|
||||
import { runWithRecording } from '../../observe/context.ts'
|
||||
import type { Observer } from '../../observe/observer.ts'
|
||||
import type { OpRecord } from '../../observe/record.ts'
|
||||
import { Channel } from '../../shell/console/types.ts'
|
||||
import type { JobConsole } from '../../shell/console/job_console.ts'
|
||||
import type { Resource } from '../../resource/base.ts'
|
||||
import { getCurrentSessionFor, runWithSession } from '../../context/session_context.ts'
|
||||
import type { JobTable } from '../../shell/job_table/index.ts'
|
||||
@@ -114,18 +116,57 @@ async function deniedResult(
|
||||
return new ExecuteResult(new Uint8Array(), msg, 126)
|
||||
}
|
||||
|
||||
/**
|
||||
* Move a buffered result into the sink, so a caller that gave one reads
|
||||
* the whole line there.
|
||||
*
|
||||
* Most of a line streams as it runs, but several paths answer with bytes
|
||||
* in hand and never reach the walk that emits: a whole-line runtime
|
||||
* returns its own buffer, and the syntax gate, a policy denial and a
|
||||
* failed line all return before or around the tree. Draining here rather
|
||||
* than at each of those keeps the contract one rule instead of five, and
|
||||
* a path added later cannot forget it. Nothing is emitted twice: a line
|
||||
* that did stream returns empty, which is the same fact this reads.
|
||||
*
|
||||
* @param sink console the caller passed as `ExecuteOptions.sink`.
|
||||
* @param result the line's result, buffered or already streamed.
|
||||
*/
|
||||
async function drainToSink(sink: JobConsole, result: ExecuteResult): Promise<ExecuteResult> {
|
||||
if (result.stdout.byteLength === 0 && result.stderr.byteLength === 0) return result
|
||||
if (result.stdout.byteLength > 0) await sink.emit(Channel.STDOUT, result.stdout)
|
||||
if (result.stderr.byteLength > 0) await sink.emit(Channel.STDERR, result.stderr)
|
||||
return new ExecuteResult(new Uint8Array(), new Uint8Array(), result.exitCode)
|
||||
}
|
||||
|
||||
/**
|
||||
* The body of `Workspace.execute`; see its docstring for the argument
|
||||
* contract. Order of gates: hydrate stores, drain any queued drift
|
||||
* check, parse, syntax gate, provision branch, policy, then the
|
||||
* strategies (whole-line runtime or command tree). Failures fold into
|
||||
* the line's result via `failureResult`, except the kinds that are the
|
||||
* caller's problem (abort, drift), which propagate.
|
||||
* contract. Runs the line, then honors the sink contract for every path
|
||||
* `runLine` can answer on.
|
||||
*/
|
||||
export async function executeLine(
|
||||
env: ExecuteEnv,
|
||||
command: string,
|
||||
options: ExecuteOptions,
|
||||
): Promise<ExecuteResult | ProvisionResult> {
|
||||
const result = await runLine(env, command, options)
|
||||
const sink = options.sink
|
||||
// A provision run answers with a plan, not output, so it has nothing
|
||||
// to stream.
|
||||
if (sink === undefined || !(result instanceof ExecuteResult)) return result
|
||||
return drainToSink(sink, result)
|
||||
}
|
||||
|
||||
/**
|
||||
* Order of gates: hydrate stores, drain any queued drift check, parse,
|
||||
* syntax gate, provision branch, policy, then the strategies (whole-line
|
||||
* runtime or command tree). Failures fold into the line's result via
|
||||
* `failureResult`, except the kinds that are the caller's problem (abort,
|
||||
* drift), which propagate.
|
||||
*/
|
||||
async function runLine(
|
||||
env: ExecuteEnv,
|
||||
command: string,
|
||||
options: ExecuteOptions,
|
||||
): Promise<ExecuteResult | ProvisionResult> {
|
||||
if (options.signal?.aborted === true) {
|
||||
throw makeAbortError()
|
||||
@@ -216,6 +257,7 @@ export async function executeLine(
|
||||
runtimeBindings: env.runtimes.bindings,
|
||||
...(routingDecision !== null ? { routingDecision } : {}),
|
||||
...(options.signal !== undefined ? { signal: options.signal } : {}),
|
||||
...(options.sink !== undefined ? { sink: options.sink } : {}),
|
||||
}
|
||||
try {
|
||||
return await runParsedLine(env, command, options, rootNode, deps, targetSession, stdin)
|
||||
|
||||
@@ -16,6 +16,7 @@ import type { CacheConfig } from '../../cache/file/config.ts'
|
||||
import type { IndexConfig } from '../../cache/index/config.ts'
|
||||
import type { CLISpec } from '../../commands/cli/types.ts'
|
||||
import type { ByteSource } from '../../io/types.ts'
|
||||
import type { JobConsole } from '../../shell/console/index.ts'
|
||||
import type { ObserverStore } from '../../observe/store.ts'
|
||||
import type { OpsRegistry } from '../../ops/registry.ts'
|
||||
import type { Resource } from '../../resource/base.ts'
|
||||
@@ -201,6 +202,21 @@ export interface ExecuteOptions {
|
||||
* Throws for a name that is not a workspace entry.
|
||||
*/
|
||||
runtime?: string
|
||||
/**
|
||||
* Stream the line's output into this console as it is produced,
|
||||
* instead of returning it whole. Each statement of a compound line
|
||||
* emits as it finishes (a single command still lands in one chunk,
|
||||
* since it has nothing to show before it completes), so a reader can
|
||||
* watch a long or compound line run. When set, the returned
|
||||
* `ExecuteResult` carries the exit code but empty stdout/stderr,
|
||||
* because the bytes went to the console; the caller owns the console
|
||||
* and decides when to `finish()` it (typically once this call
|
||||
* resolves, with the exit outcome). Every path answers this way, so
|
||||
* the console is the line's whole output: a line a whole-line runtime
|
||||
* (a RemoteSandbox) served, a syntax error, a policy denial and a
|
||||
* failed line all arrive there rather than in the result.
|
||||
*/
|
||||
sink?: JobConsole
|
||||
/**
|
||||
* @internal The typed line's routing decision, forwarded to nested
|
||||
* evals so inner lines never re-route.
|
||||
|
||||
@@ -90,9 +90,9 @@ describe('MirageService', () => {
|
||||
const ws = new Workspace({ '/data': new RAMResource() })
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(MirageService, { workspace: ws }).await()
|
||||
expect(ctx.mirage.confined).toBe(true)
|
||||
expect(ctx.mirage.vfsOnly).toBe(true)
|
||||
ws.addRuntime(new LocalRuntime({ captures: ['python'] }))
|
||||
expect(ctx.mirage.confined).toBe(false)
|
||||
expect(ctx.mirage.vfsOnly).toBe(false)
|
||||
await ws.close()
|
||||
})
|
||||
|
||||
@@ -113,10 +113,10 @@ describe('MirageService', () => {
|
||||
})
|
||||
await fiber.await()
|
||||
expect(() => ctx.mirage.workspace).toThrow('not ready')
|
||||
expect(ctx.mirage.confined).toBe(false)
|
||||
expect(ctx.mirage.vfsOnly).toBe(false)
|
||||
release()
|
||||
await ctx.mirage.ready
|
||||
expect(ctx.mirage.confined).toBe(false)
|
||||
expect(ctx.mirage.vfsOnly).toBe(false)
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
|
||||
@@ -90,7 +90,7 @@ function toRuntimeEntry(entry: string | MirageRuntimeBlock): RuntimeEntry {
|
||||
}
|
||||
|
||||
// A name shorthand builds exactly as the workspace would build it; holding
|
||||
// the instance early is what lets `confined` classify the world before the
|
||||
// the instance early is what lets `vfsOnly` classify the world before the
|
||||
// (asynchronous) mounts finish resolving.
|
||||
function builtEntry(entry: RuntimeEntry): Runtime {
|
||||
return typeof entry === 'string' ? buildRuntime(entry) : entry
|
||||
@@ -165,17 +165,20 @@ export class MirageService extends Service {
|
||||
}
|
||||
|
||||
/**
|
||||
* True while every runtime in the world is confined to the workspace
|
||||
* (`Runtime.confined`): the default world is, and so are the bridged
|
||||
* engines, while the host `local` python or a remote sandbox executes
|
||||
* beyond anything this service can vouch for. Answers before `ready`
|
||||
* from the constructor-resolved entries (mounts resolve
|
||||
* asynchronously, runtimes never do) and from the live workspace
|
||||
* afterwards, so a later `addRuntime` is seen.
|
||||
* True while every runtime in the world reaches only the vfs
|
||||
* (`Runtime.reach`), meaning the workspace dispatch is the single
|
||||
* gate for anything the agent can execute: every effect passes
|
||||
* mount modes, session grants, and policy before touching the real
|
||||
* backend behind a mount. The default world qualifies, and so do
|
||||
* the bridged engines, while the host `local` python or a remote
|
||||
* sandbox can act around the gate. Answers before `ready` from the
|
||||
* constructor-resolved entries (mounts resolve asynchronously,
|
||||
* runtimes never do) and from the live workspace afterwards, so a
|
||||
* later `addRuntime` is seen.
|
||||
*/
|
||||
get confined(): boolean {
|
||||
get vfsOnly(): boolean {
|
||||
const entries = this.built === null ? (this.plannedRuntimes ?? []) : this.built.runtimeEntries
|
||||
return entries.every((entry) => entry.confined)
|
||||
return entries.every((entry) => entry.reach === 'vfs')
|
||||
}
|
||||
|
||||
private async open(
|
||||
|
||||
@@ -286,3 +286,168 @@ describe('start', () => {
|
||||
expect(await ws.fs.exists('/data/out.txt')).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('streaming', () => {
|
||||
it('delivers a compound line incrementally, before it finishes', async () => {
|
||||
const { shell } = await makeShell()
|
||||
const proc = shell.start(shell.resolve({ command: 'echo first; sleep 0.5; echo second' }))
|
||||
let acc = ''
|
||||
const deadline = Date.now() + 3000
|
||||
while (Date.now() < deadline && !acc.includes('first')) {
|
||||
acc += proc.readOutput().delta
|
||||
if (!acc.includes('first')) await new Promise((r) => setTimeout(r, 15))
|
||||
}
|
||||
expect(acc).toContain('first')
|
||||
// The sleep is still in flight, so the second statement has not run.
|
||||
expect(acc).not.toContain('second')
|
||||
expect(proc.status).toBe('running')
|
||||
await proc.done
|
||||
acc += proc.readOutput().delta
|
||||
expect(acc).toContain('second')
|
||||
})
|
||||
|
||||
it('interleaves stdout and stderr in order, stderr marked', async () => {
|
||||
const { shell } = await makeShell()
|
||||
const proc = shell.start(shell.resolve({ command: 'echo out1; echo err1 >&2; echo out2' }))
|
||||
await proc.done
|
||||
const delta = proc.readOutput().delta
|
||||
expect(delta).toContain('--- stderr ---')
|
||||
expect(delta.indexOf('out1')).toBeLessThan(delta.indexOf('err1'))
|
||||
expect(delta.indexOf('err1')).toBeLessThan(delta.indexOf('out2'))
|
||||
})
|
||||
|
||||
it('caps the unread backlog and flags lossy, keeping the tail', async () => {
|
||||
const { shell } = await makeShell({}, { stdoutMaxBytes: 12 })
|
||||
const proc = shell.start(
|
||||
shell.resolve({ command: 'echo aaaa; echo bbbb; echo cccc; echo dddd' }),
|
||||
)
|
||||
await proc.done
|
||||
const out = proc.readOutput()
|
||||
expect(out.lossy).toBe(true)
|
||||
expect(out.delta).toContain('dddd')
|
||||
expect(out.delta).not.toContain('aaaa')
|
||||
expect(new TextEncoder().encode(out.delta).byteLength).toBeLessThanOrEqual(12)
|
||||
})
|
||||
|
||||
it('delivers the output of a line that never reached the command tree', async () => {
|
||||
const { shell } = await makeShell()
|
||||
// The syntax gate answers before the walk that streams, so this
|
||||
// arrives only because the executor drains a buffered result into
|
||||
// the console.
|
||||
const proc = shell.start(shell.resolve({ command: 'case x' }))
|
||||
await proc.done
|
||||
expect(proc.exitCode).toBe(2)
|
||||
expect(proc.readOutput().delta).toContain('syntax error')
|
||||
})
|
||||
})
|
||||
|
||||
describe('spill', () => {
|
||||
it('does not spill when no directory is configured', async () => {
|
||||
const { shell } = await makeShell({}, { stdoutMaxBytes: 12 })
|
||||
const proc = shell.start(shell.resolve({ command: 'echo aaaa; echo bbbb; echo cccc' }))
|
||||
await proc.done
|
||||
const out = proc.readOutput()
|
||||
expect(out.lossy).toBe(true)
|
||||
expect(out.stdoutSpillPath).toBeUndefined()
|
||||
})
|
||||
|
||||
it('spills the full stdout to a readable workspace file when the delta overruns', async () => {
|
||||
const { shell, ws } = await makeShell({}, { stdoutMaxBytes: 12, spillDir: '/data/spill' })
|
||||
const proc = shell.start(
|
||||
shell.resolve({ command: 'echo aaaa; echo bbbb; echo cccc; echo dddd' }),
|
||||
)
|
||||
await proc.done
|
||||
const out = proc.readOutput()
|
||||
expect(out.lossy).toBe(true)
|
||||
const stdoutPath = out.stdoutSpillPath
|
||||
if (stdoutPath === undefined) throw new Error('expected a stdout spill path')
|
||||
// The delta kept only the tail; the spill file has the whole stream.
|
||||
expect(out.delta).not.toContain('aaaa')
|
||||
const full = await ws.fs.readFileText(stdoutPath)
|
||||
expect(full).toContain('aaaa')
|
||||
expect(full).toContain('dddd')
|
||||
})
|
||||
|
||||
it('spills stdout and stderr to separate files', async () => {
|
||||
const { shell, ws } = await makeShell({}, { stdoutMaxBytes: 12, spillDir: '/data/spill' })
|
||||
const proc = shell.start(
|
||||
shell.resolve({ command: 'echo out1; echo err1 >&2; echo out2; echo out3' }),
|
||||
)
|
||||
await proc.done
|
||||
const out = proc.readOutput()
|
||||
const stdoutPath = out.stdoutSpillPath
|
||||
const stderrPath = out.stderrSpillPath
|
||||
if (stdoutPath === undefined) throw new Error('expected a stdout spill path')
|
||||
if (stderrPath === undefined) throw new Error('expected a stderr spill path')
|
||||
expect(await ws.fs.readFileText(stdoutPath)).toContain('out1')
|
||||
expect(await ws.fs.readFileText(stderrPath)).toContain('err1')
|
||||
})
|
||||
|
||||
it('spills both commands when two overrun into a missing directory at once', async () => {
|
||||
const { shell, ws } = await makeShell({}, { stdoutMaxBytes: 12, spillDir: '/data/spill' })
|
||||
const line = 'echo aaaa; echo bbbb; echo cccc; echo dddd'
|
||||
const first = shell.start(shell.resolve({ command: line }))
|
||||
const second = shell.start(shell.resolve({ command: line }))
|
||||
await Promise.all([first.done, second.done])
|
||||
const paths = [first.readOutput().stdoutSpillPath, second.readOutput().stdoutSpillPath]
|
||||
// Whichever loses the mkdir race still spills, and to its own file.
|
||||
expect(paths[0]).toBeDefined()
|
||||
expect(paths[1]).toBeDefined()
|
||||
expect(paths[0]).not.toBe(paths[1])
|
||||
for (const path of paths) {
|
||||
if (path === undefined) throw new Error('expected a stdout spill path')
|
||||
expect(await ws.fs.readFileText(path)).toContain('aaaa')
|
||||
}
|
||||
})
|
||||
|
||||
it('creates a nested spill directory', async () => {
|
||||
const { shell, ws } = await makeShell({}, { stdoutMaxBytes: 12, spillDir: '/data/runs/spill' })
|
||||
const proc = shell.start(shell.resolve({ command: 'echo aaaa; echo bbbb; echo cccc' }))
|
||||
await proc.done
|
||||
const path = proc.readOutput().stdoutSpillPath
|
||||
if (path === undefined) throw new Error('expected a stdout spill path')
|
||||
expect(path.startsWith('/data/runs/spill/')).toBe(true)
|
||||
expect(await ws.fs.readFileText(path)).toContain('aaaa')
|
||||
})
|
||||
})
|
||||
|
||||
describe('sandbox facts', () => {
|
||||
it('stamps a full-enforcement workspace-write sandbox on a run result', async () => {
|
||||
const { shell } = await makeShell({ 'a.txt': 'x' })
|
||||
const result = await shell.run(shell.resolve({ command: 'cat /data/a.txt' }))
|
||||
expect(result.sandbox).toEqual({
|
||||
mode: 'workspace-write',
|
||||
denied: false,
|
||||
enforcement: 'full',
|
||||
runnerFailed: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('reports the sandbox independently of exit status', async () => {
|
||||
const { shell } = await makeShell()
|
||||
const result = await shell.run(shell.resolve({ command: 'cat /data/nope' }))
|
||||
expect(result.exitCode).not.toBe(0)
|
||||
expect(result.sandbox?.mode).toBe('workspace-write')
|
||||
expect(result.sandbox?.denied).toBe(false)
|
||||
})
|
||||
|
||||
it('omits the sandbox once a runtime executes beyond the workspace', async () => {
|
||||
const { shell, ws } = await makeShell()
|
||||
ws.addRuntime(new LocalRuntime({ captures: ['python'] }))
|
||||
const result = await shell.run(shell.resolve({ command: 'true' }))
|
||||
expect(result.sandbox).toBeUndefined()
|
||||
})
|
||||
|
||||
it('stamps the sandbox on a settled background process', async () => {
|
||||
const { shell } = await makeShell({ 'a.txt': 'bg' })
|
||||
const proc = shell.start(shell.resolve({ command: 'cat /data/a.txt' }))
|
||||
expect(proc.sandbox).toBeUndefined()
|
||||
await proc.done
|
||||
expect(proc.sandbox).toEqual({
|
||||
mode: 'workspace-write',
|
||||
denied: false,
|
||||
enforcement: 'full',
|
||||
runnerFailed: false,
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -22,10 +22,19 @@ import type {
|
||||
ShellProcessRead,
|
||||
ShellProcessStatus,
|
||||
ShellRunResult,
|
||||
ShellSandboxInfo,
|
||||
} from '@deepseek-ai/dsh-shell'
|
||||
import type { ExecuteOptions, ExecuteResult } from '@struktoai/mirage-core'
|
||||
import {
|
||||
Channel,
|
||||
JobConsole,
|
||||
RAMConsoleStore,
|
||||
exitOutcome,
|
||||
KILLED_OUTCOME,
|
||||
} from '@struktoai/mirage-core'
|
||||
import type { ConsoleChunk, ExecuteOptions, ExecuteResult } from '@struktoai/mirage-core'
|
||||
import type { Workspace } from '@struktoai/mirage-node'
|
||||
import { tailCap } from './text.ts'
|
||||
import { SpillSink, ensureDirPath, type SpillTarget } from './spill.ts'
|
||||
import type {} from './service.ts'
|
||||
|
||||
const DEFAULT_TIMEOUT_MS = 120_000
|
||||
@@ -33,6 +42,18 @@ const MAX_TIMEOUT_MS = 600_000
|
||||
const DEFAULT_STDOUT_MAX_BYTES = 200_000
|
||||
const DEFAULT_STDERR_MAX_BYTES = 64_000
|
||||
const STDERR_MARKER = '\n--- stderr ---\n'
|
||||
// What the console may hold that the drain loop has not consumed yet.
|
||||
// Capping the delta does not bound this: a reader's cursor advances but
|
||||
// frees nothing, so an uncapped store keeps every chunk of a noisy
|
||||
// command for the life of the process. Five deltas' worth, because the
|
||||
// drain awaits the spill's own writes and has to be free to fall
|
||||
// briefly behind, and bounded, so a command that outruns it forever
|
||||
// cannot grow the heap.
|
||||
const CONSOLE_RETENTION_BYTES = 5 * DEFAULT_STDOUT_MAX_BYTES
|
||||
|
||||
// Monotonic within the process, so concurrent background commands never
|
||||
// collide on a spill filename. Not reset, so it needs no time or randomness.
|
||||
let spillCounter = 0
|
||||
|
||||
/** Configuration for the mirage shell executor. */
|
||||
export interface MirageShellConfig {
|
||||
@@ -62,6 +83,18 @@ export interface MirageShellConfig {
|
||||
* session, per mirage's `ExecuteOptions` semantics.
|
||||
*/
|
||||
sessionId?: string
|
||||
/**
|
||||
* When set, a background command whose streamed output overruns its
|
||||
* delta budget spills its full stdout and stderr to files under this
|
||||
* workspace directory, and `readOutput()` points at them so a reader
|
||||
* can recover what the delta dropped. The directory is a workspace
|
||||
* path (e.g. `/tmp` on a ram mount), so the agent reads the spill
|
||||
* through the same VFS as everything else; the writes go through the
|
||||
* workspace, so they appear in history like any other write. Unset
|
||||
* (the default) means no spill: output that overruns is simply
|
||||
* flagged `lossy`, the honest "no safe path available" answer.
|
||||
*/
|
||||
spillDir?: string
|
||||
}
|
||||
|
||||
function collect(text: string, maxBytes: number): CollectedOutput {
|
||||
@@ -74,6 +107,7 @@ function executeOptions(
|
||||
signal: AbortSignal,
|
||||
sessionId: string | undefined,
|
||||
fallbackWorkdir: string,
|
||||
sink?: JobConsole,
|
||||
): ExecuteOptions & { provision?: false } {
|
||||
const env = {
|
||||
...(spec.env ?? {}),
|
||||
@@ -91,64 +125,136 @@ function executeOptions(
|
||||
...(cwd !== undefined ? { cwd } : {}),
|
||||
...(Object.keys(env).length > 0 ? { env } : {}),
|
||||
...(spec.stdin !== undefined ? { stdin: new TextEncoder().encode(spec.stdin) } : {}),
|
||||
...(sink !== undefined ? { sink } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A background command over the workspace executor. mirage buffers a
|
||||
* command's whole output internally, so reads deliver everything at
|
||||
* completion rather than incrementally; `kill()` aborts cooperatively (the
|
||||
* executor observes the signal between pipeline stages and inside sleep).
|
||||
* A background command over the workspace executor, streamed through a
|
||||
* `JobConsole`. The command runs with the console as its `sink`, so each
|
||||
* statement of a compound line lands as it finishes rather than the whole
|
||||
* line arriving at the end (a single command still shows up in one chunk,
|
||||
* having nothing to emit before it completes). A background follow loop
|
||||
* drains the console into `pending`, which `readOutput()` hands back and
|
||||
* clears — consuming, so consecutive reads never re-deliver. Unread output
|
||||
* is bounded to `budget` bytes: the head is dropped and `lossy` set once
|
||||
* it overruns, keeping the tail, which is where the full stream spills to
|
||||
* a file. Both ends of the conduit are bounded, because draining the
|
||||
* console does not free it: the console holds a retention budget of its
|
||||
* own, and a command that outruns this loop by that much loses chunks,
|
||||
* which arrives here as a gap in the sequence. `kill()` aborts
|
||||
* cooperatively (the executor observes the signal between pipeline stages
|
||||
* and inside sleep).
|
||||
*/
|
||||
class MirageShellProcess implements ShellProcess {
|
||||
status: ShellProcessStatus = 'running'
|
||||
exitCode: number | null = null
|
||||
signal: NodeJS.Signals | null = null
|
||||
sandbox?: ShellSandboxInfo
|
||||
readonly done: Promise<void>
|
||||
|
||||
private readonly controller: AbortController
|
||||
private readonly stdoutMaxBytes: number
|
||||
private readonly stderrMaxBytes: number
|
||||
private readonly console: JobConsole
|
||||
private readonly budget: number
|
||||
private readonly spill: SpillSink | null
|
||||
private readonly sandboxInfo: ShellSandboxInfo | undefined
|
||||
private readonly consumed: Promise<void>
|
||||
private pending = ''
|
||||
private lossy = false
|
||||
private inStderr = false
|
||||
private settled = false
|
||||
private expectSeq = 0
|
||||
|
||||
constructor(
|
||||
run: Promise<ExecuteResult>,
|
||||
controller: AbortController,
|
||||
stdoutMaxBytes: number,
|
||||
stderrMaxBytes: number,
|
||||
console_: JobConsole,
|
||||
budget: number,
|
||||
spill: SpillSink | null,
|
||||
sandboxInfo: ShellSandboxInfo | undefined,
|
||||
) {
|
||||
this.controller = controller
|
||||
this.stdoutMaxBytes = stdoutMaxBytes
|
||||
this.stderrMaxBytes = stderrMaxBytes
|
||||
this.console = console_
|
||||
this.budget = budget
|
||||
this.spill = spill
|
||||
this.sandboxInfo = sandboxInfo
|
||||
this.consumed = this.consume()
|
||||
this.done = run.then(
|
||||
(result) => {
|
||||
this.finish(result)
|
||||
},
|
||||
(err: unknown) => {
|
||||
this.fail(err)
|
||||
},
|
||||
(result) => this.settle(result, null),
|
||||
(err: unknown) => this.settle(null, err),
|
||||
)
|
||||
}
|
||||
|
||||
private finish(result: ExecuteResult): void {
|
||||
this.settled = true
|
||||
this.status = 'completed'
|
||||
this.exitCode = result.exitCode
|
||||
const stdout = collect(result.stdoutText, this.stdoutMaxBytes)
|
||||
const stderr = collect(result.stderrText, this.stderrMaxBytes)
|
||||
this.lossy = stdout.truncated || stderr.truncated
|
||||
this.pending += stdout.text
|
||||
if (stderr.text !== '') this.pending += STDERR_MARKER + stderr.text
|
||||
private async consume(): Promise<void> {
|
||||
// follow() yields every chunk in sequence and ends on the CONTROL
|
||||
// chunk that finish() appends.
|
||||
for await (const chunk of this.console.follow(0)) {
|
||||
if (chunk.channel === Channel.CONTROL) return
|
||||
// A seq that skips means the console trimmed chunks this loop had
|
||||
// not read: the command outran the drain by a whole retention
|
||||
// budget. Those bytes are gone for good, so say so, and stop the
|
||||
// spill rather than let a file with a hole in it be handed back
|
||||
// as the full stream.
|
||||
if (chunk.seq !== this.expectSeq) {
|
||||
this.lossy = true
|
||||
this.spill?.disable()
|
||||
}
|
||||
this.expectSeq = chunk.seq + 1
|
||||
await this.appendChunk(chunk)
|
||||
}
|
||||
}
|
||||
|
||||
private fail(err: unknown): void {
|
||||
private async appendChunk(chunk: ConsoleChunk): Promise<void> {
|
||||
// The full, uncapped stream goes to the spill sink (if enabled)
|
||||
// before the delta is capped, so nothing dropped from the delta is
|
||||
// lost to a reader that follows the spill path.
|
||||
if (this.spill !== null) await this.spill.ingest(chunk.channel, chunk.data)
|
||||
const text = new TextDecoder().decode(chunk.data)
|
||||
// stderr rides the same delta as stdout, opened by a marker so the
|
||||
// reader can tell the two apart; a run of stderr chunks marks once.
|
||||
if (chunk.channel === Channel.STDERR) {
|
||||
if (!this.inStderr) {
|
||||
this.pending += STDERR_MARKER
|
||||
this.inStderr = true
|
||||
}
|
||||
this.pending += text
|
||||
} else {
|
||||
this.inStderr = false
|
||||
this.pending += text
|
||||
}
|
||||
// Bound the unread backlog: a reader that never drains cannot grow
|
||||
// `pending` without limit. The tail is kept (the freshest output),
|
||||
// matching what the buffered path did at completion.
|
||||
const capped = tailCap(this.pending, this.budget)
|
||||
if (capped.truncated) {
|
||||
this.pending = capped.text
|
||||
this.lossy = true
|
||||
// The delta just dropped bytes; move the full stream to files so
|
||||
// the reader can still recover them from the spill path.
|
||||
if (this.spill !== null) await this.spill.begin()
|
||||
}
|
||||
}
|
||||
|
||||
private async settle(result: ExecuteResult | null, err: unknown): Promise<void> {
|
||||
this.settled = true
|
||||
this.status = 'killed'
|
||||
this.signal = 'SIGTERM'
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
this.pending += STDERR_MARKER + message
|
||||
if (this.sandboxInfo !== undefined) this.sandbox = this.sandboxInfo
|
||||
let outcome: string
|
||||
if (result !== null) {
|
||||
this.status = 'completed'
|
||||
this.exitCode = result.exitCode
|
||||
outcome = exitOutcome(result.exitCode)
|
||||
} else {
|
||||
this.status = 'killed'
|
||||
this.signal = 'SIGTERM'
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
await this.console.emit(Channel.STDERR, new TextEncoder().encode(message))
|
||||
outcome = KILLED_OUTCOME
|
||||
}
|
||||
// The CONTROL chunk ends the follow loop; awaiting `consumed`
|
||||
// guarantees every chunk (the last one included) has landed in
|
||||
// `pending` before `done` resolves, so a read after `done` is whole.
|
||||
await this.console.finish(outcome)
|
||||
await this.consumed
|
||||
}
|
||||
|
||||
readOutput(): ShellProcessRead {
|
||||
@@ -156,7 +262,12 @@ class MirageShellProcess implements ShellProcess {
|
||||
this.pending = ''
|
||||
const lossy = this.lossy
|
||||
this.lossy = false
|
||||
return { delta, lossy }
|
||||
return {
|
||||
delta,
|
||||
lossy,
|
||||
...(this.spill?.stdoutPath !== undefined ? { stdoutSpillPath: this.spill.stdoutPath } : {}),
|
||||
...(this.spill?.stderrPath !== undefined ? { stderrSpillPath: this.spill.stderrPath } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
kill(): boolean {
|
||||
@@ -189,6 +300,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
private readonly stdoutMaxBytes: number
|
||||
private readonly stderrMaxBytes: number
|
||||
private readonly sessionId: string | undefined
|
||||
private readonly spillDir: string | undefined
|
||||
private sessionReady: Promise<void> | null = null
|
||||
|
||||
constructor(ctx: Context, config: MirageShellConfig = {}) {
|
||||
@@ -199,6 +311,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
this.stdoutMaxBytes = config.stdoutMaxBytes ?? DEFAULT_STDOUT_MAX_BYTES
|
||||
this.stderrMaxBytes = config.stderrMaxBytes ?? DEFAULT_STDERR_MAX_BYTES
|
||||
this.sessionId = config.sessionId
|
||||
this.spillDir = config.spillDir
|
||||
}
|
||||
|
||||
// The workspace may still be building (declarative mounts resolve
|
||||
@@ -207,19 +320,65 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
return this.ctx.mirage.ready
|
||||
}
|
||||
|
||||
// A spill sink for one background command, or null when no spill
|
||||
// directory is configured. The target reaches the live workspace so
|
||||
// the full stream lands on a mount the agent can read back.
|
||||
private newSpill(): SpillSink | null {
|
||||
const dir = this.spillDir
|
||||
if (dir === undefined) return null
|
||||
const target: SpillTarget = {
|
||||
ensureDir: async (d) => {
|
||||
const ws = await this.workspace()
|
||||
await ensureDirPath({ exists: (p) => ws.fs.exists(p), mkdir: (p) => ws.fs.mkdir(p) }, d)
|
||||
},
|
||||
write: async (p, bytes) => {
|
||||
const ws = await this.workspace()
|
||||
await ws.fs.writeFile(p, bytes)
|
||||
},
|
||||
append: async (p, bytes) => {
|
||||
const ws = await this.workspace()
|
||||
await ws.fs.append(p, bytes)
|
||||
},
|
||||
}
|
||||
spillCounter += 1
|
||||
return new SpillSink(target, dir, `mirage-shell-${spillCounter.toString()}`)
|
||||
}
|
||||
|
||||
/**
|
||||
* With every runtime in the world confined to the workspace, a
|
||||
* command cannot reach the host, so this executor confines like a
|
||||
* With every runtime in the world reaching only the vfs
|
||||
* (`ctx.mirage.vfsOnly`), the workspace dispatch is the single gate
|
||||
* for anything a command can do, so this executor behaves like a
|
||||
* workspace-write sandbox: reads and writes land only where mounts
|
||||
* (and their modes) allow. Declaring it lets confinement-aware
|
||||
* plugins (dsh's permission presets) compose over this executor. A
|
||||
* world holding a runtime that executes beyond the workspace (the
|
||||
* host `local` python, a remote sandbox) voids that claim, so this
|
||||
* answers undefined then — the base contract's "does not confine" —
|
||||
* and those plugins refuse to compose instead of trusting a lie.
|
||||
* (and their modes) allow. Declaring it lets sandbox-aware plugins
|
||||
* (dsh's permission presets) compose over this executor. A world
|
||||
* holding a runtime with doors around the gate (the host `local`
|
||||
* python, a remote sandbox) voids that claim, so this answers
|
||||
* undefined then (the base contract's "does not sandbox") and those
|
||||
* plugins refuse to compose instead of trusting a lie.
|
||||
*/
|
||||
override get sandboxMode(): ShellExecutor['sandboxMode'] {
|
||||
return this.ctx.mirage.confined ? 'workspace-write' : undefined
|
||||
return this.ctx.mirage.vfsOnly ? 'workspace-write' : undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* The sandbox facts to stamp on this run's result and process handle,
|
||||
* or undefined when the world is not fully workspace-bound (no claim).
|
||||
*
|
||||
* `enforcement` is 'full': when every runtime reaches only the vfs, the
|
||||
* workspace gate cannot be bypassed, so unlike an OS sandbox on an older
|
||||
* kernel there is no promised effect it fails to govern. `denied` is
|
||||
* false because mirage has no out-of-band denial channel: a refused write
|
||||
* (a read-only mount) fails in-band as an ordinary command error with a
|
||||
* nonzero exit, the way EROFS would, not as a separate sandbox verdict,
|
||||
* so there is nothing here to distinguish from the command's own failure.
|
||||
* `runnerFailed` is false because the workspace executor is the runner
|
||||
* and a failure to run surfaces as a rejected/aborted execution, not a
|
||||
* runner that never started.
|
||||
*/
|
||||
private sandboxInfo(): ShellSandboxInfo | undefined {
|
||||
const mode = this.sandboxMode
|
||||
if (mode === undefined) return undefined
|
||||
return { mode, denied: false, enforcement: 'full', runnerFailed: false }
|
||||
}
|
||||
|
||||
resolve(request: ShellExecRequest): ShellExecSpec {
|
||||
@@ -259,6 +418,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
}
|
||||
|
||||
async run(spec: ShellExecSpec): Promise<ShellRunResult> {
|
||||
const sandbox = this.sandboxInfo()
|
||||
// An already-aborted signal never fires its listener, so answer before
|
||||
// dispatch: the command must not run at all.
|
||||
if (spec.signal?.aborted === true) {
|
||||
@@ -270,6 +430,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
timeoutMs: spec.timeoutMs,
|
||||
stdout: { text: '', truncated: false },
|
||||
stderr: { text: '', truncated: false },
|
||||
...(sandbox !== undefined ? { sandbox } : {}),
|
||||
}
|
||||
}
|
||||
const controller = new AbortController()
|
||||
@@ -301,6 +462,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
timeoutMs: spec.timeoutMs,
|
||||
stdout: collect(result.stdoutText, spec.stdoutMaxBytes),
|
||||
stderr: collect(result.stderrText, this.stderrMaxBytes),
|
||||
...(sandbox !== undefined ? { sandbox } : {}),
|
||||
}
|
||||
} catch (err) {
|
||||
if (!controller.signal.aborted) throw err
|
||||
@@ -314,6 +476,7 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
timeoutMs: spec.timeoutMs,
|
||||
stdout: { text: '', truncated: false },
|
||||
stderr: { text: '', truncated: false },
|
||||
...(sandbox !== undefined ? { sandbox } : {}),
|
||||
}
|
||||
} finally {
|
||||
clearTimeout(timer)
|
||||
@@ -322,14 +485,23 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
}
|
||||
|
||||
start(spec: ShellExecSpec): ShellProcess {
|
||||
const sandbox = this.sandboxInfo()
|
||||
const controller = new AbortController()
|
||||
// The console is the streaming conduit, holding what the follow loop
|
||||
// has not drained yet (nothing, when the loop keeps up). Its own
|
||||
// retention budget is what bounds that, since reading a chunk does
|
||||
// not release it.
|
||||
const console_ = new JobConsole(new RAMConsoleStore(CONSOLE_RETENTION_BYTES))
|
||||
const spill = this.newSpill()
|
||||
if (spec.signal?.aborted === true) {
|
||||
controller.abort()
|
||||
return new MirageShellProcess(
|
||||
Promise.reject(new Error('command aborted before start')),
|
||||
controller,
|
||||
console_,
|
||||
spec.stdoutMaxBytes,
|
||||
this.stderrMaxBytes,
|
||||
spill,
|
||||
sandbox,
|
||||
)
|
||||
}
|
||||
const onAbort = (): void => {
|
||||
@@ -341,10 +513,10 @@ export class MirageShellExecutor extends ShellExecutor {
|
||||
.then((ws) =>
|
||||
ws.execute(
|
||||
spec.command,
|
||||
executeOptions(spec, controller.signal, this.sessionId, this.workdir),
|
||||
executeOptions(spec, controller.signal, this.sessionId, this.workdir, console_),
|
||||
),
|
||||
)
|
||||
.finally(() => spec.signal?.removeEventListener('abort', onAbort))
|
||||
return new MirageShellProcess(run, controller, spec.stdoutMaxBytes, this.stderrMaxBytes)
|
||||
return new MirageShellProcess(run, controller, console_, spec.stdoutMaxBytes, spill, sandbox)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Channel } from '@struktoai/mirage-core'
|
||||
import { SpillSink, ensureDirPath, type DirMaker, type SpillTarget } from './spill.ts'
|
||||
|
||||
const ENC = new TextEncoder()
|
||||
const DEC = new TextDecoder()
|
||||
|
||||
function fakeTarget(): { target: SpillTarget; files: Map<string, string>; dirs: Set<string> } {
|
||||
const files = new Map<string, string>()
|
||||
const dirs = new Set<string>()
|
||||
const target: SpillTarget = {
|
||||
ensureDir: (d) => {
|
||||
dirs.add(d)
|
||||
return Promise.resolve()
|
||||
},
|
||||
write: (p, bytes) => {
|
||||
files.set(p, DEC.decode(bytes))
|
||||
return Promise.resolve()
|
||||
},
|
||||
append: (p, bytes) => {
|
||||
files.set(p, (files.get(p) ?? '') + DEC.decode(bytes))
|
||||
return Promise.resolve()
|
||||
},
|
||||
}
|
||||
return { target, files, dirs }
|
||||
}
|
||||
|
||||
describe('SpillSink', () => {
|
||||
it('flushes the buffered stream on begin, then appends later chunks', async () => {
|
||||
const { target, files, dirs } = fakeTarget()
|
||||
const sink = new SpillSink(target, '/spill', 'job1')
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('one'))
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('two'))
|
||||
await sink.begin()
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('three'))
|
||||
expect(dirs.has('/spill')).toBe(true)
|
||||
expect(sink.stdoutPath).toBe('/spill/job1.stdout')
|
||||
expect(files.get('/spill/job1.stdout')).toBe('onetwothree')
|
||||
})
|
||||
|
||||
it('opens a stderr file lazily when stderr first arrives after begin', async () => {
|
||||
const { target, files } = fakeTarget()
|
||||
const sink = new SpillSink(target, '/spill', 'job2')
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('out'))
|
||||
await sink.begin()
|
||||
// No stderr was buffered, so begin created no stderr file.
|
||||
expect(sink.stderrPath).toBeUndefined()
|
||||
await sink.ingest(Channel.STDERR, ENC.encode('boom'))
|
||||
expect(sink.stderrPath).toBe('/spill/job2.stderr')
|
||||
expect(files.get('/spill/job2.stderr')).toBe('boom')
|
||||
})
|
||||
|
||||
it('leaves paths undefined and stops writing when a write fails', async () => {
|
||||
const failing: SpillTarget = {
|
||||
ensureDir: () => Promise.reject(new Error('read-only mount')),
|
||||
write: () => Promise.reject(new Error('read-only mount')),
|
||||
append: () => Promise.reject(new Error('read-only mount')),
|
||||
}
|
||||
const sink = new SpillSink(failing, '/spill', 'job3')
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('data'))
|
||||
await sink.begin()
|
||||
expect(sink.stdoutPath).toBeUndefined()
|
||||
// A later chunk must not throw once the sink has given up.
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('more'))
|
||||
expect(sink.stdoutPath).toBeUndefined()
|
||||
})
|
||||
|
||||
it('stops writing once a reader reports it lost bytes', async () => {
|
||||
const { target, files } = fakeTarget()
|
||||
const sink = new SpillSink(target, '/spill', 'job4')
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('one'))
|
||||
await sink.begin()
|
||||
sink.disable()
|
||||
await sink.ingest(Channel.STDOUT, ENC.encode('two'))
|
||||
expect(sink.stdoutPath).toBeUndefined()
|
||||
expect(files.get('/spill/job4.stdout')).toBe('one')
|
||||
})
|
||||
})
|
||||
|
||||
function fakeDirs(existing: string[] = []): { dirs: DirMaker; made: string[] } {
|
||||
const present = new Set(existing)
|
||||
const made: string[] = []
|
||||
const dirs: DirMaker = {
|
||||
exists: (p) => Promise.resolve(present.has(p)),
|
||||
mkdir: (p) => {
|
||||
made.push(p)
|
||||
present.add(p)
|
||||
return Promise.resolve()
|
||||
},
|
||||
}
|
||||
return { dirs, made }
|
||||
}
|
||||
|
||||
describe('ensureDirPath', () => {
|
||||
it('creates every missing ancestor in order', async () => {
|
||||
const { dirs, made } = fakeDirs(['/data'])
|
||||
await ensureDirPath(dirs, '/data/spill/out')
|
||||
expect(made).toEqual(['/data/spill', '/data/spill/out'])
|
||||
})
|
||||
|
||||
it('creates nothing when the directory is already there', async () => {
|
||||
const { dirs, made } = fakeDirs(['/data', '/data/spill'])
|
||||
await ensureDirPath(dirs, '/data/spill')
|
||||
expect(made).toEqual([])
|
||||
})
|
||||
|
||||
it('accepts a refusal for a directory that now exists', async () => {
|
||||
// Two commands overrunning at once: both probe and see nothing,
|
||||
// one wins the mkdir and the loser is refused for the directory it
|
||||
// wanted. Losing that race must not cost it its spill.
|
||||
const present = new Set(['/data'])
|
||||
const dirs: DirMaker = {
|
||||
exists: (p) => Promise.resolve(present.has(p)),
|
||||
mkdir: (p) => {
|
||||
present.add(p)
|
||||
return Promise.reject(new Error(`EEXIST: ${p}`))
|
||||
},
|
||||
}
|
||||
await expect(ensureDirPath(dirs, '/data/spill')).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('rethrows when the directory is still missing after the refusal', async () => {
|
||||
const dirs: DirMaker = {
|
||||
exists: () => Promise.resolve(false),
|
||||
mkdir: () => Promise.reject(new Error('read-only mount')),
|
||||
}
|
||||
await expect(ensureDirPath(dirs, '/data/spill')).rejects.toThrow('read-only mount')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,171 @@
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
||||
|
||||
import { Channel } from '@struktoai/mirage-core'
|
||||
|
||||
/** Where a spill sink writes: a workspace directory it can create and extend. */
|
||||
export interface SpillTarget {
|
||||
ensureDir(dir: string): Promise<void>
|
||||
write(path: string, bytes: Uint8Array): Promise<void>
|
||||
append(path: string, bytes: Uint8Array): Promise<void>
|
||||
}
|
||||
|
||||
/** The two directory facts `ensureDirPath` needs from a workspace. */
|
||||
export interface DirMaker {
|
||||
exists(path: string): Promise<boolean>
|
||||
mkdir(path: string): Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* Create `dir` and any missing ancestor, treating an existing one as
|
||||
* done.
|
||||
*
|
||||
* `mkdir` is one level and is not idempotent, so a probe alone is not
|
||||
* enough: two commands that first overrun at the same moment both see
|
||||
* the spill directory missing, one creates it and the other is refused
|
||||
* for a directory that is now exactly what it wanted. Losing that race
|
||||
* must not cost a command its spill, so the refusal is re-checked
|
||||
* against existence and only a directory that is still missing is a
|
||||
* real failure.
|
||||
*
|
||||
* @param dirs the workspace's `exists`/`mkdir`.
|
||||
* @param dir absolute workspace path to create.
|
||||
*/
|
||||
export async function ensureDirPath(dirs: DirMaker, dir: string): Promise<void> {
|
||||
let path = ''
|
||||
for (const part of dir.split('/').filter((p) => p !== '')) {
|
||||
path += `/${part}`
|
||||
if (await dirs.exists(path)) continue
|
||||
try {
|
||||
await dirs.mkdir(path)
|
||||
} catch (err) {
|
||||
if (!(await dirs.exists(path))) throw err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function totalLength(parts: Uint8Array[]): number {
|
||||
return parts.reduce((sum, p) => sum + p.byteLength, 0)
|
||||
}
|
||||
|
||||
function concat(parts: Uint8Array[]): Uint8Array {
|
||||
const out = new Uint8Array(totalLength(parts))
|
||||
let at = 0
|
||||
for (const p of parts) {
|
||||
out.set(p, at)
|
||||
at += p.byteLength
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps the full, uncapped stdout and stderr of one command in workspace
|
||||
* files so a reader can recover output the delta budget dropped.
|
||||
*
|
||||
* Memory stays bounded: each channel buffers in memory only until the
|
||||
* delivered delta first overruns its budget, at which point `begin()`
|
||||
* flushes both buffers to their files and every later chunk appends
|
||||
* straight to the file. A write failure (no writable mount at the
|
||||
* configured path) disables the sink and leaves the paths undefined, the
|
||||
* honest "no safe path available" answer, so the process still streams,
|
||||
* just without a spill to point at.
|
||||
*/
|
||||
export class SpillSink {
|
||||
stdoutPath: string | undefined
|
||||
stderrPath: string | undefined
|
||||
|
||||
private readonly target: SpillTarget
|
||||
private readonly dir: string
|
||||
private readonly base: string
|
||||
private started = false
|
||||
private failed = false
|
||||
private stdoutParts: Uint8Array[] = []
|
||||
private stderrParts: Uint8Array[] = []
|
||||
|
||||
constructor(target: SpillTarget, dir: string, base: string) {
|
||||
this.target = target
|
||||
this.dir = dir
|
||||
this.base = base
|
||||
}
|
||||
|
||||
/** Buffer a chunk before spill starts, or append it to the file after. */
|
||||
async ingest(channel: Channel, data: Uint8Array): Promise<void> {
|
||||
if (this.failed) return
|
||||
if (!this.started) {
|
||||
if (channel === Channel.STDERR) this.stderrParts.push(data)
|
||||
else this.stdoutParts.push(data)
|
||||
return
|
||||
}
|
||||
await this.appendFile(channel, data)
|
||||
}
|
||||
|
||||
/** Flush the buffered streams to files; called once, on the first overrun. */
|
||||
async begin(): Promise<void> {
|
||||
if (this.started || this.failed) return
|
||||
try {
|
||||
await this.target.ensureDir(this.dir)
|
||||
if (totalLength(this.stdoutParts) > 0) {
|
||||
this.stdoutPath = `${this.dir}/${this.base}.stdout`
|
||||
await this.target.write(this.stdoutPath, concat(this.stdoutParts))
|
||||
}
|
||||
if (totalLength(this.stderrParts) > 0) {
|
||||
this.stderrPath = `${this.dir}/${this.base}.stderr`
|
||||
await this.target.write(this.stderrPath, concat(this.stderrParts))
|
||||
}
|
||||
this.started = true
|
||||
this.stdoutParts = []
|
||||
this.stderrParts = []
|
||||
} catch {
|
||||
this.disable()
|
||||
}
|
||||
}
|
||||
|
||||
private async appendFile(channel: Channel, data: Uint8Array): Promise<void> {
|
||||
try {
|
||||
if (channel === Channel.STDERR) {
|
||||
if (this.stderrPath === undefined) {
|
||||
this.stderrPath = `${this.dir}/${this.base}.stderr`
|
||||
await this.target.write(this.stderrPath, data)
|
||||
} else {
|
||||
await this.target.append(this.stderrPath, data)
|
||||
}
|
||||
} else {
|
||||
if (this.stdoutPath === undefined) {
|
||||
this.stdoutPath = `${this.dir}/${this.base}.stdout`
|
||||
await this.target.write(this.stdoutPath, data)
|
||||
} else {
|
||||
await this.target.append(this.stdoutPath, data)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
this.disable()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Give up on spilling and report no path.
|
||||
*
|
||||
* Called on a write failure, and by a reader that discovers it lost
|
||||
* bytes before they ever reached here: a file missing the middle of
|
||||
* the stream is worse than no file, because the path presents it as
|
||||
* the whole one.
|
||||
*/
|
||||
disable(): void {
|
||||
this.failed = true
|
||||
this.stdoutPath = undefined
|
||||
this.stderrPath = undefined
|
||||
this.stdoutParts = []
|
||||
this.stderrParts = []
|
||||
}
|
||||
}
|
||||
@@ -38,6 +38,11 @@ const LOCAL_HOME_ENV = 'MIRAGE_LOCAL_HOME'
|
||||
*/
|
||||
export class LocalRuntime extends PythonRuntime {
|
||||
readonly name = 'local'
|
||||
// Spawns the host interpreter: a real process with the user's own
|
||||
// filesystem and network, doors the workspace gate never sees. This
|
||||
// is the base default; declared here so the claim is explicit at the
|
||||
// one builtin runtime that voids a world's sandbox claim.
|
||||
override readonly reach = 'process'
|
||||
private readonly python: string
|
||||
private readonly children = new Set<ChildProcess>()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user