fix(tui): stop local shell descendants on exit (#127652)

* fix(tui): own local shell process scopes

Keep local shell root results independent from authoritative process-scope
extinction, and make TUI shutdown cancel and join its ProcessSupervisor scope.
On Windows, create commands atomically inside a retained non-breakaway Job
Object and prove extinction from exact Job accounting.

Fixes #127488

* fix(process): preserve optional extinction waits
This commit is contained in:
Peter Steinberger
2026-08-21 17:01:09 -07:00
committed by GitHub
parent 7c8e94804c
commit 89e38cfef9
26 changed files with 2298 additions and 733 deletions
+1 -1
View File
@@ -162,7 +162,7 @@ const rootEntries = [
"src/agents/model-provider-auth.worker.ts!",
"src/agents/prepared-model-catalog.worker.ts!",
// Spawned through computed sibling URLs by the service-child host and relay.
"src/process/supervisor/{service-child-relay,service-child-group-anchor}.ts!",
"src/process/supervisor/{service-child-relay,service-child-group-anchor,service-child-windows-job-anchor}.ts!",
// Loaded by URL from setup-inference-detection.ts; no static import edge exists.
"src/system-agent/setup-inference-detection.worker.ts!",
// Split runtime loaded through a path assembled in subagent-registry.ts.
+2 -1
View File
@@ -1975,7 +1975,7 @@
"test:watch": "node --import tsx scripts/test-projects.mts --watch",
"test:windows:ci": "pnpm test:windows:ci:1 && pnpm test:windows:ci:2",
"test:windows:ci:1": "node --import tsx scripts/test-projects.mts src/state/openclaw-state-ownership.test.ts src/snapshot/local-repository.windows.test.ts src/state/openclaw-database-paths.windows.test.ts src/node-host/invoke-system-run-allowlist.test.ts src/node-host/node-worker-transfer-client.test.ts packages/terminal-core/src/display-string.test.ts src/config/sessions/session-accessor.sqlite-archive.worker.test.ts src/commands/doctor-gateway-auth-token.windows.test.ts src/infra/state-migrations.audit-logs.windows.test.ts src/commands/backup-verify.test.ts src/commands/agents.commands.list.test.ts src/shared/runtime-import.test.ts test/scripts/direct-run-entrypoints.test.ts test/scripts/format-generated-module.test.ts test/scripts/openclaw-cross-os-installer.windows.test.ts test/scripts/openclaw-cross-os-release-workflow.test.ts test/scripts/pnpm-runner.test.ts test/scripts/run-with-env.test.ts test/scripts/ui.test.ts test/scripts/vitest-process-group.test.ts src/agents/tools/media-tool-file-url.windows.test.ts extensions/memory-core/src/memory-extra-file-path.windows.test.ts src/daemon/schtasks.startup-fallback.test.ts src/plugin-sdk/node-host.test.ts test/e2e/qa-lab/runtime/package-openclaw-for-docker.e2e.test.ts extensions/lobster/src/lobster-runner.test.ts",
"test:windows:ci:2": "node --import tsx scripts/test-projects.mts src/infra/sqlite-snapshot.test.ts src/infra/ports.test.ts src/infra/advertised-lan-host.windows.test.ts src/infra/update-managed-service-handoff-command.test.ts src/infra/update-managed-service-handoff-lifecycle.test.ts src/infra/executable-path.test.ts src/infra/fs-safe-remove.test.ts src/infra/state-migrations.legacy-session-store.test.ts src/infra/windows-install-roots.test.ts src/infra/ssh-client.windows.test.ts src/infra/exec-allowlist-pattern.test.ts src/infra/process-env.test.ts src/agents/sessions/windows-git-bash-path.test.ts test/scripts/npm-runner.test.ts test/scripts/ts-topology.test.ts src/test-utils/openclaw-test-state.test.ts src/utils.test.ts src/media/local-media-path.windows.test.ts src/media/web-media.file-url.windows.test.ts src/process/exec.windows.test.ts src/process/exec.windows.integration.test.ts src/process/windows-command.test.ts src/process/terminal-pty.test.ts src/tui/tui.resolve-codex-bin.test.ts src/media-understanding/attachments.file-url.windows.test.ts src/cli/daemon-cli/status.print.test.ts src/cli/mcp-cli.path-case.windows.test.ts src/auto-reply/usage-bar/template.windows.test.ts src/auto-reply/reply.triggers.trigger-handling.stages-inbound-media-into-sandbox-workspace.test.ts src/agents/provider-local-service.env-case.test.ts src/agents/bash-tools.exec.script-preflight.test.ts src/agents/agent-tools.read.windows.test.ts src/agents/agent-tools.read.host-operations.test.ts src/agents/sandbox/fs-paths.test.ts src/agents/sessions/tools/render-utils.test.ts src/agents/sessions/tools/path-utils.test.ts extensions/msteams/src/media-helpers.test.ts extensions/msteams/src/messenger.test.ts extensions/mxc/test/mxc-backend.test.ts extensions/mxc/test/sandbox-policy-loader.test.ts",
"test:windows:ci:2": "node --import tsx scripts/test-projects.mts src/infra/sqlite-snapshot.test.ts src/infra/ports.test.ts src/infra/advertised-lan-host.windows.test.ts src/infra/update-managed-service-handoff-command.test.ts src/infra/update-managed-service-handoff-lifecycle.test.ts src/infra/executable-path.test.ts src/infra/fs-safe-remove.test.ts src/infra/state-migrations.legacy-session-store.test.ts src/infra/windows-install-roots.test.ts src/infra/ssh-client.windows.test.ts src/infra/exec-allowlist-pattern.test.ts src/infra/process-env.test.ts src/agents/sessions/windows-git-bash-path.test.ts test/scripts/npm-runner.test.ts test/scripts/ts-topology.test.ts src/test-utils/openclaw-test-state.test.ts src/utils.test.ts src/media/local-media-path.windows.test.ts src/media/web-media.file-url.windows.test.ts src/process/exec.windows.test.ts src/process/exec.windows.integration.test.ts src/process/windows-command.test.ts src/process/terminal-pty.test.ts src/process/supervisor/supervisor.anchored-shell.real.test.ts src/tui/tui.resolve-codex-bin.test.ts src/media-understanding/attachments.file-url.windows.test.ts src/cli/daemon-cli/status.print.test.ts src/cli/mcp-cli.path-case.windows.test.ts src/auto-reply/usage-bar/template.windows.test.ts src/auto-reply/reply.triggers.trigger-handling.stages-inbound-media-into-sandbox-workspace.test.ts src/agents/provider-local-service.env-case.test.ts src/agents/bash-tools.exec.script-preflight.test.ts src/agents/agent-tools.read.windows.test.ts src/agents/agent-tools.read.host-operations.test.ts src/agents/sandbox/fs-paths.test.ts src/agents/sessions/tools/render-utils.test.ts src/agents/sessions/tools/path-utils.test.ts extensions/msteams/src/media-helpers.test.ts extensions/msteams/src/messenger.test.ts extensions/mxc/test/mxc-backend.test.ts extensions/mxc/test/sandbox-policy-loader.test.ts",
"test:windows:schtasks:integration": "node --import tsx scripts/run-with-env.mts CI_WINDOWS_SCHTASKS_INTEGRATION=1 OPENCLAW_E2E_VERBOSE=1 OPENCLAW_VITEST_MAX_WORKERS=1 -- node scripts/run-vitest.mjs src/daemon/schtasks.integration.e2e.test.ts",
"tool-display:check": "node --import tsx scripts/tool-display.ts --check",
"tool-display:write": "node --import tsx scripts/tool-display.ts --write",
@@ -2063,6 +2063,7 @@
"jiti": "2.7.0",
"json5": "2.2.3",
"jszip": "3.10.1",
"koffi": "3.1.5",
"kysely": "0.29.4",
"linkedom": "0.18.13",
"minimatch": "10.2.5",
+67 -64
View File
@@ -159,6 +159,9 @@ importers:
jszip:
specifier: 3.10.1
version: 3.10.1
koffi:
specifier: 3.1.5
version: 3.1.5
kysely:
specifier: 0.29.4
version: 0.29.4
@@ -3550,78 +3553,78 @@ packages:
'@konnorr/qr-creator@1.0.1':
resolution: {integrity: sha512-EmRR9rny1ENBtQy7TLOguO/79h1EpzXUmqIdMTGp2BW8NkZtRRPr5hmQcE+n9QXYo4T8NjcaJ14hnZg+s/+K+A==}
'@koromix/koffi-darwin-arm64@3.1.2':
resolution: {integrity: sha512-32pU4pNZABIz+l9DNJl51Y+jur4vv+SF4Ip2CSF4OUg1xUyefoLpX0NttDmzGITIrneUEVSEN+dT22524ESKBw==}
'@koromix/koffi-darwin-arm64@3.1.5':
resolution: {integrity: sha512-IpqITl2fJi3QN9bTtNnygWPdK7ScSjw3xtGu8e6feYGvimCysu+spgI5KyeslY2jTnqxGS9xr8pLAbLhGJ8edA==}
cpu: [arm64]
os: [darwin]
'@koromix/koffi-darwin-x64@3.1.2':
resolution: {integrity: sha512-S+H6LQgUoMj77BqDegwlRaxwLXDfwvSJGuceOqtH0I5V8rzKLmu/hC7NBlxOoAlvKlcV63FtdNiE2E9YSltffg==}
'@koromix/koffi-darwin-x64@3.1.5':
resolution: {integrity: sha512-4Tia4BS5EV/+vN9eIrdToanVe+U/2VqTZCBgOzoUbPKjgky51eqM+3J4qdRUvmYJohcJNPob5/hsxeItUZrl1g==}
cpu: [x64]
os: [darwin]
'@koromix/koffi-freebsd-arm64@3.1.2':
resolution: {integrity: sha512-fD0ow2PBE60nw7K6xcbala6qwXxfcYeU62tduNeIPvx0KoWhU2rMKZiDNe+iI5TQb3rxYYjjP+aF2Sdm9y6EXQ==}
'@koromix/koffi-freebsd-arm64@3.1.5':
resolution: {integrity: sha512-bP94uzseFO79NG3flpU3WxfyvltD+jzC/kN8FDZLi8J0VUZNW1Wmu8yq87KEzjX1qT9KyX4Y+elVbGqHXHTv2Q==}
cpu: [arm64]
os: [freebsd]
'@koromix/koffi-freebsd-ia32@3.1.2':
resolution: {integrity: sha512-t8OmL+hoJGDLZDnuLjgLemSYrXX99M7Md+zJX8bMHOtiNbFtkGXn/mV21Pb1ik9JhBXjwK1r4hvBPNlqTMGrHg==}
'@koromix/koffi-freebsd-ia32@3.1.5':
resolution: {integrity: sha512-raFXXAPHzvCQWhaoMUF+Cc2ZWgg2UBU0RVoowHZhaw9nQYPC1pERcPRH+JA+SNIN6g4d2GFW6uPFc+QbUhsagA==}
cpu: [ia32]
os: [freebsd]
'@koromix/koffi-freebsd-x64@3.1.2':
resolution: {integrity: sha512-axbLgiM4Y2vyDOTqlXCI8vkg9wqjwSRsmoWXSKreA5YFJwnYA6Sc4aHMz+qZgUSfFei52Qrv1RGhDyo4kHvqhA==}
'@koromix/koffi-freebsd-x64@3.1.5':
resolution: {integrity: sha512-h6RyBZmPMBIDWTABkJIhzDdYwSnYAJvTacHpEjbT55Arkmw1H15Rl7CFtXuEuBrqh+uivoCrRgA6vszl9CsJ9g==}
cpu: [x64]
os: [freebsd]
'@koromix/koffi-linux-arm64@3.1.2':
resolution: {integrity: sha512-f0hqAIlFcL9wlRGJ/uCfyfspqnGaASk2gLx1UAP3RBgMQl68D1e+fiHNdXa7g9d76ttmpA8/PGNAqc1X4Byy1Q==}
'@koromix/koffi-linux-arm64@3.1.5':
resolution: {integrity: sha512-u0vCmKPu4yQDhl/ri1J6U3vDnvYtYjoZaIWb+oMbRXhVZeiqdE53MGPb+q2A7Dj2n9IbYloAfEICQbL6l0pmiQ==}
cpu: [arm64]
os: [linux]
'@koromix/koffi-linux-ia32@3.1.2':
resolution: {integrity: sha512-UGLPuqeOV/UArsK6oeB5yI/XjSWkFqFlBTC9rUbezBuHJhSibk1EMv7QC0cvtDMu18bo+ucqXWPzh42oT5yYlw==}
'@koromix/koffi-linux-ia32@3.1.5':
resolution: {integrity: sha512-Xa5JbumWglwPVZgrJcLhqyC1wCWlfm7+C00p3FuOTNGp0qoYuf2/tOoAh0/q7+taVm30cMopu/6lRHwdmF6I/w==}
cpu: [ia32]
os: [linux]
'@koromix/koffi-linux-loong64@3.1.2':
resolution: {integrity: sha512-jI0+gM2oDsJ7reOt3XPyO7lyQtZ1CT6NR2uqGQcQVM43cyXBAVYYCUxEH3LHCbgumFaZ+LueIUgbMSwb9pHBxQ==}
'@koromix/koffi-linux-loong64@3.1.5':
resolution: {integrity: sha512-F3i2CeTcqVBUQiSRUBSEzX1VgtXmLLiZb/ouZtXHkWpTLhJNd9TH7s3CizTofci0VRqlKdexGUYhK5vzPDAAHA==}
cpu: [loong64]
os: [linux]
'@koromix/koffi-linux-riscv64@3.1.2':
resolution: {integrity: sha512-yB99adXBRd5T+xXG+f6nnUkC3jCI0iXvPU6RqD9Kx7aZP4Y4NNUWJ5Q4FaP9jb1XmZLY4pGBUiHt8u03Yl7NyA==}
'@koromix/koffi-linux-riscv64@3.1.5':
resolution: {integrity: sha512-2TgQuzy+4PfDg+rw3kOmN6lywEWdzKT3eaLPbOp0b/9DaN7CLBJ/QIR5GhGsMNpeI30zU0YzFeYFxoVoJ/LqFw==}
cpu: [riscv64]
os: [linux]
'@koromix/koffi-linux-x64@3.1.2':
resolution: {integrity: sha512-Oxvo6F3Edzy/Jm2EtbHWkJ2xRB0mXDAe63k5+USL5uiGE5xZjwEUDOBKIhv2BpCZSOAJrfoojFFogj6+ICKQhw==}
'@koromix/koffi-linux-x64@3.1.5':
resolution: {integrity: sha512-2yaIg/1V0m4CiAUMzG4CIlWmq1WJ+QBMlFfaGr9su+OH5fuIqC7V3BbMiB03IwIl1VofIZO5JA4Db4lID8tpbw==}
cpu: [x64]
os: [linux]
'@koromix/koffi-openbsd-ia32@3.1.2':
resolution: {integrity: sha512-SSWzUhL8Ex84JTsO67+MdWZrdwgOzoOrQ0+ZbB+UsivHoAxmWLHKWZaSafNqyBZtxGY1EgtR8AIPouWE9U+Zfw==}
'@koromix/koffi-openbsd-ia32@3.1.5':
resolution: {integrity: sha512-8/OXd+u9omMooykhvdJPEP7u6FFzzrrFo9gOmSHc9/DPt3XkVYOtSsE97PDk6zYaAzwIYHSKjHvIXsfFwFc7sg==}
cpu: [ia32]
os: [openbsd]
'@koromix/koffi-openbsd-x64@3.1.2':
resolution: {integrity: sha512-0ZuI4St7chq3M0d3VivvKIqacZ7RhgohdR476V3HpJkaNdfIywsJIw+GBvqkQahu+4A2Rpu6yQJpWSrfk/Z+Jw==}
'@koromix/koffi-openbsd-x64@3.1.5':
resolution: {integrity: sha512-SpeqldKkuDk2aTj5PVWumy7eq6Tr2GtBPAOI1NiDHhg8xe433KraxrA9V9UjEd+1+kSGJQGR04nQByN3MPA9PQ==}
cpu: [x64]
os: [openbsd]
'@koromix/koffi-win32-arm64@3.1.2':
resolution: {integrity: sha512-8Wn6phw7y53uI52+aBPAqEfZ5pj/HCjg/YtdthqSWYHy+d0MhyASKlcmuP0B5raxQnnA1Bm9LC8UO3M3RojeBw==}
'@koromix/koffi-win32-arm64@3.1.5':
resolution: {integrity: sha512-uej3YAEKAhlfVPoIo5sOwtxhTLRVJ01LgtWrKGpnnAQU3C+Ilmaxdh+Oc2xc1G3NK30N5eCVJpyO9r3pjKC6Vw==}
cpu: [arm64]
os: [win32]
'@koromix/koffi-win32-ia32@3.1.2':
resolution: {integrity: sha512-FkKaPBMawgHMNnp1FwLldXMNvEa139GXkxPi9JD9xU71Kh/ZmuEYHGSD6JwZDmDr4jekVrBrr+eGZ+j6C2mkXg==}
'@koromix/koffi-win32-ia32@3.1.5':
resolution: {integrity: sha512-d42jv2f4PwtJGNJS19Xfn/BRtGsBNNVkw0O0K5tkIGI+yNq4MnPTSUsaGbDIkCLNsCgk/LFqAaE8BExyCdrujA==}
cpu: [ia32]
os: [win32]
'@koromix/koffi-win32-x64@3.1.2':
resolution: {integrity: sha512-FeFC59UU1XX4J3ZaqKrsrEzczzB5qksMJo7/R45vIg8mGNVSLMVE85JRiZpjcp9i5Lbav5Vw47QvwFzBgIfvlw==}
'@koromix/koffi-win32-x64@3.1.5':
resolution: {integrity: sha512-Pyo1WEHEP6Ek2NEn2pquwJzSPLOdY4vymoPSz0an1DgvFWSkOyBYYkGVoxL1ajfj5I1pPdXysYqixtVTzAtOfQ==}
cpu: [x64]
os: [win32]
@@ -7010,8 +7013,8 @@ packages:
resolution: {integrity: sha512-dhG34DXATL5hSxJbIexCft8FChFXtmskoZYnoPWjXQuebWYCNkVeV3KkGegCK9CP1oswI/vQibS2GY7Em/sJJA==}
engines: {node: '>= 8'}
koffi@3.1.2:
resolution: {integrity: sha512-wVwuE21TBl8/si6E0hPorKR2PJ2q33mEWVETANrtSp3kFM8fi2FcD/J5wmxu0T4TBcqmMQ4xKuF1X1ayFmphzw==}
koffi@3.1.5:
resolution: {integrity: sha512-XVwwrxg0Ca6IEUQF4YtGIU4XN0LSselFYpYvgfhh8wafCunhEEx5hPr7LZhp5QyeFA/LcRsKHTqncCdjWjWAlg==}
kysely@0.29.4:
resolution: {integrity: sha512-y5mVgQNkMbs1eK9Xyc0pmNdabN2wHhRYY/5r4W5HrUT1rYCEPeVNSj1RUJeSDKT3U0p+mXCvLgkrFuIafYI6BA==}
@@ -10219,7 +10222,7 @@ snapshots:
'@github/copilot-sdk@1.0.7':
dependencies:
'@github/copilot': 1.0.75
koffi: 3.1.2
koffi: 3.1.5
vscode-jsonrpc: 8.2.1
zod: 4.4.3
@@ -10448,49 +10451,49 @@ snapshots:
'@konnorr/qr-creator@1.0.1': {}
'@koromix/koffi-darwin-arm64@3.1.2':
'@koromix/koffi-darwin-arm64@3.1.5':
optional: true
'@koromix/koffi-darwin-x64@3.1.2':
'@koromix/koffi-darwin-x64@3.1.5':
optional: true
'@koromix/koffi-freebsd-arm64@3.1.2':
'@koromix/koffi-freebsd-arm64@3.1.5':
optional: true
'@koromix/koffi-freebsd-ia32@3.1.2':
'@koromix/koffi-freebsd-ia32@3.1.5':
optional: true
'@koromix/koffi-freebsd-x64@3.1.2':
'@koromix/koffi-freebsd-x64@3.1.5':
optional: true
'@koromix/koffi-linux-arm64@3.1.2':
'@koromix/koffi-linux-arm64@3.1.5':
optional: true
'@koromix/koffi-linux-ia32@3.1.2':
'@koromix/koffi-linux-ia32@3.1.5':
optional: true
'@koromix/koffi-linux-loong64@3.1.2':
'@koromix/koffi-linux-loong64@3.1.5':
optional: true
'@koromix/koffi-linux-riscv64@3.1.2':
'@koromix/koffi-linux-riscv64@3.1.5':
optional: true
'@koromix/koffi-linux-x64@3.1.2':
'@koromix/koffi-linux-x64@3.1.5':
optional: true
'@koromix/koffi-openbsd-ia32@3.1.2':
'@koromix/koffi-openbsd-ia32@3.1.5':
optional: true
'@koromix/koffi-openbsd-x64@3.1.2':
'@koromix/koffi-openbsd-x64@3.1.5':
optional: true
'@koromix/koffi-win32-arm64@3.1.2':
'@koromix/koffi-win32-arm64@3.1.5':
optional: true
'@koromix/koffi-win32-ia32@3.1.2':
'@koromix/koffi-win32-ia32@3.1.5':
optional: true
'@koromix/koffi-win32-x64@3.1.2':
'@koromix/koffi-win32-x64@3.1.5':
optional: true
'@lancedb/lancedb-darwin-arm64@0.31.0':
@@ -13953,23 +13956,23 @@ snapshots:
klona@2.0.6: {}
koffi@3.1.2:
koffi@3.1.5:
optionalDependencies:
'@koromix/koffi-darwin-arm64': 3.1.2
'@koromix/koffi-darwin-x64': 3.1.2
'@koromix/koffi-freebsd-arm64': 3.1.2
'@koromix/koffi-freebsd-ia32': 3.1.2
'@koromix/koffi-freebsd-x64': 3.1.2
'@koromix/koffi-linux-arm64': 3.1.2
'@koromix/koffi-linux-ia32': 3.1.2
'@koromix/koffi-linux-loong64': 3.1.2
'@koromix/koffi-linux-riscv64': 3.1.2
'@koromix/koffi-linux-x64': 3.1.2
'@koromix/koffi-openbsd-ia32': 3.1.2
'@koromix/koffi-openbsd-x64': 3.1.2
'@koromix/koffi-win32-arm64': 3.1.2
'@koromix/koffi-win32-ia32': 3.1.2
'@koromix/koffi-win32-x64': 3.1.2
'@koromix/koffi-darwin-arm64': 3.1.5
'@koromix/koffi-darwin-x64': 3.1.5
'@koromix/koffi-freebsd-arm64': 3.1.5
'@koromix/koffi-freebsd-ia32': 3.1.5
'@koromix/koffi-freebsd-x64': 3.1.5
'@koromix/koffi-linux-arm64': 3.1.5
'@koromix/koffi-linux-ia32': 3.1.5
'@koromix/koffi-linux-loong64': 3.1.5
'@koromix/koffi-linux-riscv64': 3.1.5
'@koromix/koffi-linux-x64': 3.1.5
'@koromix/koffi-openbsd-ia32': 3.1.5
'@koromix/koffi-openbsd-x64': 3.1.5
'@koromix/koffi-win32-arm64': 3.1.5
'@koromix/koffi-win32-ia32': 3.1.5
'@koromix/koffi-win32-x64': 3.1.5
kysely@0.29.4: {}
+1 -1
View File
@@ -64,7 +64,7 @@ const WINDOWS_FILE_URL_SCOPE_RE =
const WINDOWS_SCOPE_RE =
/^(extensions\/mxc\/|src\/agents\/(?:bash-tools\.exec-script-(?:preflight|target)|bash-tools\.exec\.script-preflight\.test)\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive(?:\.worker(?:\.test)?)?|store\.session-lifecycle-mutation\.test)\.ts$|src\/process\/|src\/infra\/(?:(?:advertised-lan-host|exec-allowlist-pattern|fs-safe-remove)(?:\.windows)?(?:\.test)?|ports(?:-inspect|\.test)|ssh-client(?:\.windows\.test)?|update-managed-service-handoff(?:-(?:command|lifecycle)\.test)?|windows-install-roots)\.ts$|src\/shared\/(?:import-specifier|runtime-import)(?:\.test)?\.ts$|src\/test-utils\/openclaw-test-state(?:\.test)?\.ts$|scripts\/(?:android-(?:app-i18n|pin-version)\.ts|ci-run-timings\.mjs|e2e\/lib\/package-compat\.mjs|generate-bundled-channel-config-metadata\.ts|install\.ps1|openclaw-cross-os-release-checks\.ts|plan-release-workflow-matrix\.mjs|run-additional-boundary-checks\.mts|verify-docker-attestations\.mjs|github\/run-openclaw-cross-os-release-checks\.sh|(?:npm-runner|pnpm-runner|ui|vitest-process-group)\.(?:mjs|mts|js)|lib\/(?:direct-run\.(?:mjs|mts)|format-generated-module\.mts|tsx-cli-shim\.mjs|cross-os-release-checks\/[^/]+\.ts))$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|install-ps1|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$|package\.json$|pnpm-lock\.yaml$|pnpm-workspace\.yaml$|\.github\/workflows\/(?:ci|openclaw-cross-os-release-checks-reusable)\.yml$|\.github\/actions\/setup-node-env\/action\.yml$|\.github\/actions\/setup-pnpm-store-cache\/action\.yml$)/;
const WINDOWS_TEST_SCOPE_RE =
/^(extensions\/mxc\/test\/(?:mxc-backend|sandbox-policy-loader)\.test\.ts$|src\/agents\/bash-tools\.exec\.script-preflight\.test\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive\.worker|store\.session-lifecycle-mutation)\.test\.ts$|src\/process\/(?:exec\.windows|terminal-pty|windows-command)\.test\.ts$|src\/infra\/(?:advertised-lan-host(?:\.windows)?|exec-allowlist-pattern|fs-safe-remove|ports|ssh-client\.windows|update-managed-service-handoff-(?:command|lifecycle)|windows-install-roots)\.test\.ts$|src\/shared\/runtime-import\.test\.ts$|src\/state\/openclaw-database-paths\.windows\.test\.ts$|src\/test-utils\/openclaw-test-state\.test\.ts$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$)/;
/^(extensions\/mxc\/test\/(?:mxc-backend|sandbox-policy-loader)\.test\.ts$|src\/agents\/bash-tools\.exec\.script-preflight\.test\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive\.worker|store\.session-lifecycle-mutation)\.test\.ts$|src\/process\/(?:exec\.windows|supervisor\/supervisor\.anchored-shell\.real|terminal-pty|windows-command)\.test\.ts$|src\/infra\/(?:advertised-lan-host(?:\.windows)?|exec-allowlist-pattern|fs-safe-remove|ports|ssh-client\.windows|update-managed-service-handoff-(?:command|lifecycle)|windows-install-roots)\.test\.ts$|src\/shared\/runtime-import\.test\.ts$|src\/state\/openclaw-database-paths\.windows\.test\.ts$|src\/test-utils\/openclaw-test-state\.test\.ts$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$)/;
const WINDOWS_SECRETREF_SCOPE_RE =
/^(?:src\/commands\/doctor-gateway-auth-token(?:\.windows\.test)?\.ts|src\/flows\/(?:doctor-core-checks|doctor-health-contributions)\.ts|src\/gateway\/(?:auth-token-resolution|resolve-configured-secret-input-string)\.ts|src\/infra\/(?:fs-safe|fs-safe-defaults|permissions)\.ts|src\/secrets\/(?:resolve|resolve-errors)\.ts|src\/security\/audit-fs\.ts)$/;
const WINDOWS_SECRETREF_TEST_SCOPE_RE =
+6
View File
@@ -18,3 +18,9 @@ export function resolveRuntimeWorkerUrl(params: {
const extension = path.extname(currentPath) || ".js";
return new URL(`./${params.sourceWorkerName}${extension}`, params.currentModuleUrl);
}
export function resolveRuntimeWorkerArgv(url: URL): string[] {
return url.pathname.endsWith(".ts")
? ["--import", "tsx", fileURLToPath(url)]
: [fileURLToPath(url)];
}
+19 -3
View File
@@ -84,8 +84,7 @@ function isServiceManagedRuntime(): boolean {
return Boolean(process.env.OPENCLAW_SERVICE_MARKER?.trim());
}
export async function createChildAdapter(params: {
argv: string[];
type ChildAdapterInput = {
/** Own a separately signalable tree whose private IPC channel gates worker startup. */
ownedWorker?: true;
/** Preserve the supplied environment exactly by skipping environment-mutating spawn wrappers. */
@@ -97,7 +96,24 @@ export async function createChildAdapter(params: {
input?: string;
stdinMode?: "inherit" | "pipe-open" | "pipe-closed";
secretInput?: SpawnSecretInput;
}): Promise<WorkerChildAdapter> {
} & (
| { argv: string[]; anchoredShellCommand?: never }
| { argv?: never; anchoredShellCommand: string }
);
export async function createChildAdapter(params: ChildAdapterInput): Promise<WorkerChildAdapter> {
if (params.anchoredShellCommand !== undefined) {
return await createServiceChildRelayAdapter({
command: process.platform === "win32" ? params.anchoredShellCommand : "/bin/sh",
args: process.platform === "win32" ? [] : ["-c", params.anchoredShellCommand],
windowsShellCommand: process.platform === "win32" ? params.anchoredShellCommand : undefined,
cwd: params.cwd,
env: params.env,
stdinMode: "pipe-closed",
oomScoreWrapperSelected: false,
});
}
const baseEnv = params.env ? toStringEnv(params.env) : undefined;
const invocation = resolveChildInvocation({
argv: params.argv,
@@ -1,24 +1,21 @@
import { spawn, type ChildProcess } from "node:child_process";
import { Socket } from "node:net";
import { pipeline, type Readable } from "node:stream";
import { createDeferredCore } from "../../shared/deferred.js";
import { GRACEFUL_CANCEL_TIMEOUT_MS } from "./cancellation-policy.js";
import {
encodeServiceChildMessage,
type ServiceChildAnchorMessage,
type ServiceChildAnchorPayload,
type ServiceChildControlMessage,
type ServiceChildStart,
} from "./service-child-protocol.js";
type WithoutEnvelope<T> = T extends unknown ? Omit<T, "generation" | "sequence"> : never;
type ServiceChildAnchorPayload = WithoutEnvelope<ServiceChildAnchorMessage>;
const LINEAGE_EXIT_OBSERVATION_MS = 100;
type AnchorState = "starting" | "active" | "closing" | "closed";
type StdioEntry = "ignore" | "inherit" | "pipe" | number;
type ServiceChildControlMessage = {
generation: string;
sequence: number;
} & ({ type: "cancel"; signal: "SIGTERM" | "SIGKILL" } | { type: "startup-error-ack" });
function commandStdio(start: ServiceChildStart): {
stdio: StdioEntry[];
lineageFd: number;
@@ -62,26 +59,11 @@ export function runServiceChildGroupAnchor(): void {
let stderrDrained = false;
let lineageClosed = false;
let forceCleanup = false;
let resolveForceCleanup: () => void = () => {};
const forceCleanupRequested = new Promise<void>((resolve) => {
resolveForceCleanup = resolve;
});
let resolveLineage!: () => void;
const lineageDone = new Promise<void>((resolve) => {
resolveLineage = resolve;
});
let resolveRootExited!: () => void;
const rootExited = new Promise<void>((resolve) => {
resolveRootExited = resolve;
});
let resolveRootSettled!: () => void;
const rootSettledDone = new Promise<void>((resolve) => {
resolveRootSettled = resolve;
});
let resolveStartupErrorAck!: () => void;
const startupErrorAcknowledged = new Promise<void>((resolve) => {
resolveStartupErrorAck = resolve;
});
const forceCleanupRequested = createDeferredCore();
const lineageDone = createDeferredCore();
const rootExited = createDeferredCore();
const rootSettledDone = createDeferredCore();
const startupErrorAcknowledged = createDeferredCore();
const send = async (message: ServiceChildAnchorPayload) => {
if (!start || !control || control.destroyed) {
@@ -122,7 +104,7 @@ export function runServiceChildGroupAnchor(): void {
await send({ type: "startup-error", error });
// A write callback only proves kernel acceptance. Keep the exact anchor alive until the
// host records the authoritative spawn failure and acknowledges it on this same channel.
await startupErrorAcknowledged;
await startupErrorAcknowledged.promise;
await closeAuthority("lineage-lost", false);
};
@@ -136,7 +118,7 @@ export function runServiceChildGroupAnchor(): void {
if (state === "closing") {
forceCleanup ||= signal === "SIGKILL";
if (forceCleanup) {
resolveForceCleanup();
forceCleanupRequested.resolve();
}
return;
}
@@ -146,7 +128,7 @@ export function runServiceChildGroupAnchor(): void {
if (!forceCleanup) {
// The anchor catches its own signal while every command-group member receives it.
process.kill(0, "SIGTERM");
await Promise.race([lineageDone, termGraceDone, forceCleanupRequested]);
await Promise.race([lineageDone.promise, termGraceDone, forceCleanupRequested.promise]);
}
if (state !== "closing" || !start) {
return;
@@ -154,7 +136,7 @@ export function runServiceChildGroupAnchor(): void {
if (lineageClosed && !rootExit && !forceCleanup) {
// Cleanup already owns the group. A normal root exit may race lineage EOF,
// but the short observation window must not replace the configured TERM grace.
await Promise.race([rootExited, termGraceDone, forceCleanupRequested]);
await Promise.race([rootExited.promise, termGraceDone, forceCleanupRequested.promise]);
}
if (state !== "closing" || !start) {
return;
@@ -162,7 +144,7 @@ export function runServiceChildGroupAnchor(): void {
if (lineageClosed && rootExit && !forceCleanup) {
// Output can outlive lineage and the root. It may preserve the authentic root
// result only within the existing TERM grace, and KILL must wake this wait.
await Promise.race([rootSettledDone, termGraceDone, forceCleanupRequested]);
await Promise.race([rootSettledDone.promise, termGraceDone, forceCleanupRequested.promise]);
if (state !== "closing" || !start) {
return;
}
@@ -183,13 +165,17 @@ export function runServiceChildGroupAnchor(): void {
}
lastHostSequence = message.sequence;
if (message.type === "startup-error-ack") {
resolveStartupErrorAck();
startupErrorAcknowledged.resolve();
return;
}
void requestCleanup("cancel", message.signal);
};
const startCommand = async (next: ServiceChildStart) => {
if (next.controlFd === undefined) {
process.exitCode = 1;
return;
}
start = next;
control = new Socket({ fd: start.controlFd, readable: true, writable: true });
control.setEncoding("utf8");
@@ -247,19 +233,19 @@ export function runServiceChildGroupAnchor(): void {
return;
}
lineageClosed = true;
resolveLineage();
lineageDone.resolve();
if (state === "active") {
// Pipe EOF and the child exit notification race independently. Wait
// briefly for the exact child event before treating EOF as lease loss.
void (async () => {
if (!rootExit) {
await Promise.race([rootExited, delay(LINEAGE_EXIT_OBSERVATION_MS)]);
await Promise.race([rootExited.promise, delay(LINEAGE_EXIT_OBSERVATION_MS)]);
}
if (state !== "active") {
return;
}
if (rootExit && rootSettlementStarted) {
await rootSettledDone;
await rootSettledDone.promise;
}
if (state !== "active") {
return;
@@ -279,7 +265,7 @@ export function runServiceChildGroupAnchor(): void {
}
rootSettlementStarted = true;
await rootResultDelivery;
resolveRootSettled();
rootSettledDone.resolve();
if (lineageClosed && state === "active") {
await closeAuthority("lineage-closed", false);
}
@@ -321,7 +307,7 @@ export function runServiceChildGroupAnchor(): void {
// The host gates public settlement on output EOF, so record the authentic root
// result before cleanup can hard-close an output-holding descendant.
rootResultDelivery = send({ type: "root-result", code, signal });
resolveRootExited();
rootExited.resolve();
void settleRoot();
});
};
@@ -7,15 +7,16 @@ export type ServiceChildStart = {
env?: Record<string, string>;
stdinMode: "inherit" | "pipe-open" | "pipe-closed";
secretFd?: number;
controlFd: number;
controlFd?: number;
windowsShellCommand?: string;
};
type ServiceChildControlMessage = {
export type ServiceChildControlMessage = {
generation: string;
sequence: number;
} & ({ type: "cancel"; signal: "SIGTERM" | "SIGKILL" } | { type: "startup-error-ack" });
type ServiceChildAnchorPayload =
export type ServiceChildAnchorPayload =
| {
type: "ready";
commandPid: number;
@@ -26,6 +27,19 @@ type ServiceChildAnchorPayload =
code: number | null;
signal: NodeJS.Signals | null;
}
| {
type: "result-error";
error: string;
}
| {
type: "output";
stream: "stdout" | "stderr";
chunk: string;
}
| {
type: "output-end";
stream: "stdout" | "stderr";
}
| {
type: "closing";
reason: "cancel" | "lineage-closed" | "lineage-lost" | "parent-lost";
+294 -175
View File
@@ -1,8 +1,11 @@
import { spawn, type ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import type { Duplex, Readable } from "node:stream";
import { fileURLToPath } from "node:url";
import { resolveRuntimeWorkerUrl } from "../../infra/runtime-worker-url.js";
import { toErrorObject } from "../../infra/errors.js";
import {
resolveRuntimeWorkerArgv,
resolveRuntimeWorkerUrl,
} from "../../infra/runtime-worker-url.js";
import { createDeferredCore } from "../../shared/deferred.js";
import { onDecodedOutput } from "../decoded-output.js";
import { addSecretInputStdio, writeSecretInputToChild } from "../spawn-secret-input.js";
@@ -11,31 +14,24 @@ import { toStringEnv } from "./adapters/env.js";
import {
encodeServiceChildMessage,
type ServiceChildAnchorMessage,
type ServiceChildControlMessage,
type ServiceChildRelayMessage,
type ServiceChildStart,
} from "./service-child-protocol.js";
import type { SpawnProcessAdapter, SpawnSecretInput } from "./types.js";
type ServiceAdapter = SpawnProcessAdapter<NodeJS.Signals | null>;
type ServiceChildRelayAdapter = SpawnProcessAdapter<NodeJS.Signals | null> & {
waitForExtinction: () => Promise<void>;
};
type AuthorityState = "starting" | "active" | "closing" | "closed" | "identity-lost";
type StdioEntry = "ignore" | "inherit" | "ipc" | "pipe" | number;
const retainedRelays = new Map<string, ChildProcess>();
const retainedChildren = new Map<string, ChildProcess>();
const PUSHED_OUTPUT_BUFFER_LIMIT_BYTES = 256 * 1024;
function runtimeArgv(url: URL): string[] {
return url.pathname.endsWith(".ts")
? ["--import", "tsx", fileURLToPath(url)]
: [fileURLToPath(url)];
}
function readAnchorMessage(line: string): ServiceChildAnchorMessage {
// SAFETY: the exact private anchor channel writes only encoded anchor protocol messages.
return JSON.parse(line) as ServiceChildAnchorMessage;
}
function readRelayMessage(raw: unknown): ServiceChildRelayMessage {
// SAFETY: the spawned relay is the sole sender on this exact private IPC channel.
return raw as ServiceChildRelayMessage;
function readChildMessage(raw: unknown): ServiceChildRelayMessage | ServiceChildAnchorMessage {
// SAFETY: the spawned relay or Job anchor is the sole writer on each private protocol channel.
return raw as ServiceChildRelayMessage | ServiceChildAnchorMessage;
}
function reserveStdioEntry(stdio: StdioEntry[], value: StdioEntry): number {
@@ -50,12 +46,13 @@ function reserveStdioEntry(stdio: StdioEntry[], value: StdioEntry): number {
return fd;
}
function createOutputRelay(stream: Readable) {
function createOutputRelay(stream?: Readable) {
const listeners = new Set<(chunk: string) => void>();
const rawListeners = new Set<(chunk: Buffer) => void>();
const pending: Array<string | Buffer> = [];
let pendingBytes = 0;
let active = false;
let ended = false;
const deliver = (chunk: string | Buffer) => {
if (typeof chunk === "string") {
listeners.forEach((listener) => listener(chunk));
@@ -73,23 +70,41 @@ function createOutputRelay(stream: Readable) {
}
pending.length = 0;
pendingBytes = 0;
stream.resume();
stream?.resume();
};
const enqueue = (chunk: string | Buffer) => {
const push = (chunk: string | Buffer) => {
if (active) {
deliver(chunk);
return;
return true;
}
const chunkBytes = Buffer.isBuffer(chunk) ? chunk.length : Buffer.byteLength(chunk);
if (!stream && pendingBytes + chunkBytes > PUSHED_OUTPUT_BUFFER_LIMIT_BYTES) {
return false;
}
pending.push(chunk);
if (Buffer.isBuffer(chunk)) {
pendingBytes += chunk.length;
if (pendingBytes >= stream.readableHighWaterMark) {
stream.pause();
}
if (!stream || Buffer.isBuffer(chunk)) {
pendingBytes += chunkBytes;
}
if (stream && pendingBytes >= stream.readableHighWaterMark) {
// POSIX can retain later output in its native pipe until subscription.
stream.pause();
}
return true;
};
onDecodedOutput(stream, enqueue, enqueue);
const end = () => {
ended = true;
};
if (stream) {
onDecodedOutput(stream, push, push);
stream.once("end", end);
stream.once("close", end);
}
return {
get ended() {
return ended;
},
push,
end,
subscribe: (listener: (chunk: string) => void, onRaw?: (chunk: Buffer) => void) => {
listeners.add(listener);
if (onRaw) {
@@ -116,94 +131,97 @@ export async function createServiceChildRelayAdapter(params: {
input?: string;
secretInput?: SpawnSecretInput;
oomScoreWrapperSelected: boolean;
}): Promise<ServiceAdapter> {
windowsShellCommand?: string;
}): Promise<ServiceChildRelayAdapter> {
const generation = randomUUID();
const relayUrl = resolveRuntimeWorkerUrl({
const useWindowsJobAnchor =
process.platform === "win32" && params.windowsShellCommand !== undefined;
const workerUrl = resolveRuntimeWorkerUrl({
currentModuleUrl: import.meta.url,
sourceWorkerName: "service-child-relay",
distWorkerPath: "process/supervisor/service-child-relay.js",
sourceWorkerName: useWindowsJobAnchor
? "service-child-windows-job-anchor"
: "service-child-relay",
distWorkerPath: useWindowsJobAnchor
? "process/supervisor/service-child-windows-job-anchor.js"
: "process/supervisor/service-child-relay.js",
});
const stdio: StdioEntry[] = [params.stdinMode === "inherit" ? "inherit" : "pipe", "pipe", "pipe"];
// SAFETY: stdio contains only SpawnStdioEntry values until lifecycle descriptors are reserved.
addSecretInputStdio(stdio as Parameters<typeof addSecretInputStdio>[0], params.secretInput);
const controlFd = reserveStdioEntry(stdio, "pipe");
const stdio: StdioEntry[] = useWindowsJobAnchor
? ["ignore", "ignore", "ignore"]
: [params.stdinMode === "inherit" ? "inherit" : "pipe", "pipe", "pipe"];
if (!useWindowsJobAnchor) {
// SAFETY: stdio contains only SpawnStdioEntry values until lifecycle descriptors are reserved.
addSecretInputStdio(stdio as Parameters<typeof addSecretInputStdio>[0], params.secretInput);
}
const controlFd = useWindowsJobAnchor ? undefined : reserveStdioEntry(stdio, "pipe");
reserveStdioEntry(stdio, "ipc");
const relay = spawn(process.execPath, runtimeArgv(relayUrl), {
const child = spawn(process.execPath, resolveRuntimeWorkerArgv(workerUrl), {
stdio,
detached: false,
// Windows must keep its exact Job owner alive long enough to observe host IPC loss.
detached: useWindowsJobAnchor,
windowsHide: true,
env: process.env,
});
retainedRelays.set(generation, relay);
relay.unref();
retainedChildren.set(generation, child);
child.unref();
// SAFETY: controlFd was reserved as a pipe in this exact spawn stdio array.
const control = relay.stdio[controlFd] as Duplex | null;
if (!relay.connected || !control || !relay.stdout || !relay.stderr) {
relay.kill("SIGKILL");
retainedRelays.delete(generation);
throw new Error("service child relay channels were not created");
// SAFETY: a defined controlFd was reserved as a pipe in this exact spawn stdio array.
const control = controlFd === undefined ? null : (child.stdio[controlFd] as Duplex | null);
if (!child.connected || (!useWindowsJobAnchor && (!control || !child.stdout || !child.stderr))) {
child.kill("SIGKILL");
retainedChildren.delete(generation);
throw new Error("service child lifecycle channels were not created");
}
const { stdout, stderr } = relay;
const stdoutRelay = createOutputRelay(stdout);
const stderrRelay = createOutputRelay(stderr);
stdout.on("error", () => {});
stderr.on("error", () => {});
const stdoutRelay = createOutputRelay(child.stdout ?? undefined);
const stderrRelay = createOutputRelay(child.stderr ?? undefined);
child.stdout?.on("error", () => {});
child.stderr?.on("error", () => {});
let state: AuthorityState = "starting";
let commandPid: number | undefined;
let outboundSequence = 0;
let inboundSequence = 0;
let rootResult: { code: number | null; signal: NodeJS.Signals | null } | undefined;
let resultError: Error | undefined;
let closingReceipt = false;
let controlError: Error | undefined;
let childError: Error | undefined;
let childDisconnected = false;
let childExited = false;
let requestedSignal: "SIGTERM" | "SIGKILL" | undefined;
let waitError: Error | undefined;
let resolveStartup!: () => void;
let rejectStartup!: (error: Error) => void;
const startup = new Promise<void>((resolve, reject) => {
resolveStartup = resolve;
rejectStartup = reject;
});
let resolveWait:
| ((result: { code: number | null; signal: NodeJS.Signals | null }) => void)
| undefined;
let rejectWait: ((error: Error) => void) | undefined;
let waitPromise: Promise<{ code: number | null; signal: NodeJS.Signals | null }> | undefined;
let waitSettled = false;
const extinction = createDeferredCore();
void extinction.promise.catch(() => undefined);
const startup = createDeferredCore();
const resultCompletion = createDeferredCore<{
code: number | null;
signal: NodeJS.Signals | null;
}>();
const extinctionCompletion = createDeferredCore();
// Failures can arrive before either public wait is requested.
void resultCompletion.promise.catch(() => {});
void extinctionCompletion.promise.catch(() => {});
let startupErrorAckDelivery: Promise<void> | undefined;
const settleWait = () => {
if (waitSettled) {
const error = waitError ?? resultError;
if (error) {
resultCompletion.reject(error);
return;
}
if (waitError) {
waitSettled = true;
rejectWait?.(waitError);
return;
}
if (
!rootResult ||
!(stdout.readableEnded || stdout.closed) ||
!(stderr.readableEnded || stderr.closed)
) {
if (!rootResult || !stdoutRelay.ended || !stderrRelay.ended) {
return;
}
if (requestedSignal && state !== "closed") {
return;
}
waitSettled = true;
resolveWait?.(rootResult);
resultCompletion.resolve(rootResult);
};
// Root result and output EOF cross different channels. Decoder flush listeners were
// registered first, so settlement observes both final text tails before disposal.
stdout.once("end", settleWait);
stdout.once("close", settleWait);
stderr.once("end", settleWait);
stderr.once("close", settleWait);
child.stdout?.once("end", settleWait);
child.stdout?.once("close", settleWait);
child.stderr?.once("end", settleWait);
child.stderr?.once("close", settleWait);
const loseIdentity = (message: string) => {
if (state === "closed" || state === "identity-lost") {
@@ -212,92 +230,191 @@ export async function createServiceChildRelayAdapter(params: {
state = "identity-lost";
waitError = new Error(`service child cleanup identity lost: ${message}`);
if (!commandPid) {
rejectStartup(waitError);
startup.reject(waitError);
}
settleWait();
extinction.reject(waitError);
extinctionCompletion.reject(waitError);
};
let pending = "";
control.setEncoding("utf8");
control.on("data", (chunk: string) => {
pending += chunk;
for (;;) {
const newline = pending.indexOf("\n");
if (newline < 0) {
break;
}
const line = pending.slice(0, newline);
pending = pending.slice(newline + 1);
let message: ServiceChildAnchorMessage;
try {
message = readAnchorMessage(line);
} catch {
loseIdentity("invalid anchor message");
continue;
}
if (message.generation !== generation || message.sequence <= inboundSequence) {
loseIdentity("stale anchor generation or sequence");
continue;
}
inboundSequence = message.sequence;
if (message.type === "ready" && state === "starting") {
commandPid = message.commandPid;
state = "active";
resolveStartup();
} else if (message.type === "root-result") {
rootResult ??= { code: message.code, signal: message.signal };
settleWait();
} else if (message.type === "closing") {
closingReceipt = true;
state = "closing";
} else if (message.type === "startup-error") {
loseIdentity(message.error);
outboundSequence += 1;
control.write(
encodeServiceChildMessage({
type: "startup-error-ack",
generation,
sequence: outboundSequence,
}),
);
const sendChildMessage = (
message: ServiceChildStart | ServiceChildControlMessage,
): Promise<void> =>
new Promise((resolve, reject) => {
if (!child.connected) {
reject(new Error("service child lifecycle IPC is closed"));
return;
}
child.send(message, (error) => {
if (error) {
reject(error);
} else {
resolve();
}
});
});
const sendControlMessage = (message: ServiceChildControlMessage): Promise<void> => {
if (useWindowsJobAnchor) {
return sendChildMessage(message);
}
});
control.once("close", () => {
// Correlation is not authority: only the receipt on this exact channel followed by
// its closure proves the in-group anchor disabled every future cleanup action.
return new Promise((resolve, reject) => {
if (!control || control.destroyed) {
reject(new Error("service child control pipe is closed"));
return;
}
control.write(encodeServiceChildMessage(message), "utf8", (error) => {
if (error) {
reject(error);
} else {
resolve();
}
});
});
};
const finishAuthorityClose = (missingReceiptError: string) => {
if (!closingReceipt) {
loseIdentity("anchor channel closed without a matching closing receipt");
loseIdentity(missingReceiptError);
return;
}
state = "closed";
rootResult ??= { code: null, signal: requestedSignal ?? null };
if (!rootResult && !resultError && !waitError) {
rootResult = { code: null, signal: requestedSignal ?? null };
}
settleWait();
extinction.resolve();
});
control.on("error", (error) => {
loseIdentity(error.message);
});
extinctionCompletion.resolve();
};
relay.on("message", (raw: unknown) => {
const message = readRelayMessage(raw);
if (!message || typeof message !== "object" || message.generation !== generation) {
const handleAnchorMessage = (message: ServiceChildAnchorMessage) => {
if (message.generation !== generation || message.sequence <= inboundSequence) {
loseIdentity("stale anchor generation or sequence");
return;
}
inboundSequence = message.sequence;
if (message.type === "ready" && state === "starting") {
commandPid = message.commandPid;
state = "active";
startup.resolve();
} else if (message.type === "root-result") {
if (!resultError) {
rootResult ??= { code: message.code, signal: message.signal };
}
settleWait();
} else if (message.type === "result-error") {
resultError ??= new Error(`service child result unavailable: ${message.error}`);
settleWait();
} else if (message.type === "output") {
if (!(message.stream === "stdout" ? stdoutRelay : stderrRelay).push(message.chunk)) {
resultError ??= new Error(
`service child ${message.stream} exceeded its pre-subscription buffer`,
);
settleWait();
}
} else if (message.type === "output-end") {
(message.stream === "stdout" ? stdoutRelay : stderrRelay).end();
settleWait();
} else if (message.type === "closing") {
closingReceipt = true;
state = "closing";
} else if (message.type === "startup-error") {
if (useWindowsJobAnchor) {
startup.reject(new Error(message.error));
} else {
loseIdentity(message.error);
}
outboundSequence += 1;
startupErrorAckDelivery = sendControlMessage({
type: "startup-error-ack",
generation,
sequence: outboundSequence,
});
void startupErrorAckDelivery.catch((error: unknown) =>
loseIdentity(toErrorObject(error, "startup error acknowledgement failed").message),
);
}
};
if (control) {
let pending = "";
control.setEncoding("utf8");
control.on("data", (chunk: string) => {
pending += chunk;
for (;;) {
const newline = pending.indexOf("\n");
if (newline < 0) {
break;
}
const line = pending.slice(0, newline);
pending = pending.slice(newline + 1);
try {
const message = readChildMessage(JSON.parse(line));
if (!("sequence" in message)) {
throw new Error("invalid anchor message");
}
handleAnchorMessage(message);
} catch {
loseIdentity("invalid anchor message");
}
}
});
control.once("close", () => {
finishAuthorityClose(
childError?.message ??
controlError?.message ??
"anchor channel closed without a matching closing receipt",
);
});
control.on("error", (error) => {
controlError ??= error;
});
}
child.on("message", (raw: unknown) => {
const message = readChildMessage(raw);
if (!message || typeof message !== "object") {
if (useWindowsJobAnchor) {
loseIdentity("invalid anchor message");
}
return;
}
if (useWindowsJobAnchor) {
if (!("sequence" in message)) {
loseIdentity("invalid anchor message");
return;
}
handleAnchorMessage(message);
return;
}
if (message.generation !== generation) {
return;
}
if (message.type === "relay-error") {
loseIdentity(message.error);
} else if (message.type === "anchor-exit" && state !== "closed" && !closingReceipt) {
loseIdentity(`anchor exited (${message.code ?? message.signal ?? "unknown"})`);
}
});
relay.once("error", (error) => {
loseIdentity(error.message);
child.once("error", (error) => {
// The direct control pipe may still contain the anchor's authoritative closing receipt.
childError ??= error;
});
relay.once("exit", (code, signal) => {
retainedRelays.delete(generation);
if (state !== "closed" && !closingReceipt) {
loseIdentity(`relay exited (${code ?? signal ?? "unknown"})`);
const finishWindowsAuthority = () => {
if (!useWindowsJobAnchor || !childDisconnected || !childExited) {
return;
}
finishAuthorityClose(
childError?.message ?? "Windows service child anchor exited without a closing receipt",
);
retainedChildren.delete(generation);
};
child.once("disconnect", () => {
childDisconnected = true;
finishWindowsAuthority();
});
child.once("exit", () => {
childExited = true;
if (useWindowsJobAnchor) {
finishWindowsAuthority();
} else {
retainedChildren.delete(generation);
}
});
@@ -311,25 +428,37 @@ export async function createServiceChildRelayAdapter(params: {
stdinMode: params.stdinMode,
secretFd: params.secretInput?.fd,
controlFd,
windowsShellCommand: params.windowsShellCommand,
};
relay.send(start);
try {
await sendChildMessage(start);
} catch (error) {
child.kill("SIGKILL");
retainedChildren.delete(generation);
throw error;
}
const [startupResult, secretDeliveryResult] = await Promise.allSettled([
startup,
writeSecretInputToChild(relay, params.secretInput),
startup.promise,
writeSecretInputToChild(child, params.secretInput),
]);
const startupError = startupResult.status === "rejected" ? startupResult.reason : undefined;
const secretDeliveryError =
secretDeliveryResult.status === "rejected" ? secretDeliveryResult.reason : undefined;
if (startupError !== undefined || secretDeliveryError !== undefined) {
relay.kill("SIGKILL");
retainedRelays.delete(generation);
if (useWindowsJobAnchor && startupError !== undefined) {
await startupErrorAckDelivery;
await extinctionCompletion.promise;
} else {
child.kill("SIGKILL");
retainedChildren.delete(generation);
}
// Startup owns command admission, so its exact failure wins over a concurrent
// backpressured secret pipe closing as a consequence of that failed admission.
throw startupError ?? secretDeliveryError;
}
const stdin = createManagedChildStdin(relay.stdin);
const stdin = createManagedChildStdin(child.stdin);
if (params.input !== undefined) {
stdin?.write(params.input);
stdin?.end();
@@ -345,13 +474,13 @@ export async function createServiceChildRelayAdapter(params: {
requestedSignal = normalized;
outboundSequence += 1;
// The host never converts the diagnostic command PID into group authority.
control.write(
encodeServiceChildMessage({
type: "cancel",
generation,
sequence: outboundSequence,
signal: normalized,
}),
void sendControlMessage({
type: "cancel",
generation,
sequence: outboundSequence,
signal: normalized,
}).catch((error: unknown) =>
loseIdentity(toErrorObject(error, "service child cancellation failed").message),
);
};
@@ -366,19 +495,9 @@ export async function createServiceChildRelayAdapter(params: {
stdoutRelay.drain();
stderrRelay.drain();
settleWait();
if (waitError) {
throw waitError;
}
if (rootResult && waitSettled) {
return rootResult;
}
waitPromise ??= new Promise((resolve, reject) => {
resolveWait = resolve;
rejectWait = reject;
});
return await waitPromise;
return await resultCompletion.promise;
},
waitForExtinction: () => extinction.promise,
waitForExtinction: async () => await extinctionCompletion.promise,
kill,
dispose: () => {
stdoutRelay.clear();
+10 -9
View File
@@ -1,14 +1,10 @@
import { spawn, type ChildProcess } from "node:child_process";
import { fileURLToPath } from "node:url";
import { resolveRuntimeWorkerUrl } from "../../infra/runtime-worker-url.js";
import {
resolveRuntimeWorkerArgv,
resolveRuntimeWorkerUrl,
} from "../../infra/runtime-worker-url.js";
import type { ServiceChildRelayMessage, ServiceChildStart } from "./service-child-protocol.js";
function runtimeArgv(url: URL): string[] {
return url.pathname.endsWith(".ts")
? ["--import", "tsx", fileURLToPath(url)]
: [fileURLToPath(url)];
}
type StdioEntry = "ignore" | "inherit" | "ipc" | number;
function reserveIpcFd(stdio: StdioEntry[]): void {
@@ -58,6 +54,11 @@ export function runServiceChildRelay(): void {
return;
}
generation = start.generation;
if (start.controlFd === undefined) {
report({ type: "relay-error", generation, error: "service child control fd is missing" });
process.exitCode = 1;
return;
}
const anchorUrl = resolveRuntimeWorkerUrl({
currentModuleUrl: import.meta.url,
sourceWorkerName: "service-child-group-anchor",
@@ -76,7 +77,7 @@ export function runServiceChildRelay(): void {
}
reserveIpcFd(stdio);
try {
anchor = spawn(process.execPath, runtimeArgv(anchorUrl), {
anchor = spawn(process.execPath, resolveRuntimeWorkerArgv(anchorUrl), {
stdio,
detached: true,
windowsHide: true,
@@ -0,0 +1,595 @@
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import { resolveEnvironmentValue } from "../../infra/process-env.js";
import { createWindowsOutputDecoder } from "../../infra/windows-encoding.js";
import { getWindowsCmdExePath } from "../../infra/windows-install-roots.js";
import { createDeferredCore } from "../../shared/deferred.js";
import type {
ServiceChildAnchorMessage,
ServiceChildAnchorPayload,
ServiceChildStart,
} from "./service-child-protocol.js";
import { createWindowsJobBindings } from "./service-child-windows-job-native.js";
import {
buildWindowsJobEnvironmentBlock,
isWindowsJobServiceStart,
} from "./service-child-windows-job-start.js";
type AnchorState = "starting" | "active" | "closing" | "closed";
type NativeHandle = bigint;
type WindowsJobBindings = ReturnType<typeof createWindowsJobBindings>;
type CommandStdio = ReturnType<WindowsJobBindings["createCommandStdio"]>;
type ClosingReason = Extract<ServiceChildAnchorMessage, { type: "closing" }>["reason"];
type OutputStream = {
name: "stdout" | "stderr";
handle?: NativeHandle;
decoder?: ReturnType<typeof createWindowsOutputDecoder>;
ended: boolean;
};
const JOB_OBJECT_BASIC_ACCOUNTING_INFORMATION = 1;
const JOB_OBJECT_EXTENDED_LIMIT_INFORMATION = 9;
const STARTF_USESTDHANDLES = 0x0000_0100;
const CREATE_NEW_PROCESS_GROUP = 0x0000_0200;
const CREATE_UNICODE_ENVIRONMENT = 0x0000_0400;
const EXTENDED_STARTUPINFO_PRESENT = 0x0008_0000;
const WAIT_OBJECT_0 = 0;
const WAIT_TIMEOUT = 258;
const WAIT_FAILED = 0xffff_ffff;
const ERROR_BROKEN_PIPE = 109;
const IDLE_OBSERVATION_MS = 10;
const OUTPUT_BUFFER_BYTES = 64 * 1024;
const OUTPUT_ROUNDS_PER_TURN = 2;
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
function sendProcessMessage(message: ServiceChildAnchorMessage): Promise<void> {
return new Promise((resolve, reject) => {
if (!process.connected || !process.send) {
reject(new Error("Windows Job anchor IPC is closed"));
return;
}
process.send(message, (error) => (error ? reject(error) : resolve()));
});
}
export function runServiceChildWindowsJobAnchor(): void {
let start: ServiceChildStart | undefined;
let state: AnchorState = "starting";
let sequence = 0;
let lastHostSequence = 0;
let outboundQueue = Promise.resolve();
let job: NativeHandle | undefined;
let processHandle: NativeHandle | undefined;
let bindings: WindowsJobBindings | undefined;
let pendingCommandStdio: CommandStdio | undefined;
let rootObserved = false;
let extinctionProven = false;
let terminationRequested = false;
let closeReason: ClosingReason | undefined;
let lifecycleTimer: NodeJS.Timeout | undefined;
let lifecycleImmediate: NodeJS.Immediate | undefined;
let lifecycleRunning = false;
let lifecycleRerun = false;
const outputStreams: OutputStream[] = [];
const outputBuffer = Buffer.allocUnsafe(OUTPUT_BUFFER_BYTES);
const startupErrorAcknowledged = createDeferredCore();
const cleanupFinished = createDeferredCore();
void cleanupFinished.promise.catch(() => {});
const send = (payload: ServiceChildAnchorPayload): Promise<void> => {
if (!start) {
return Promise.reject(new Error("Windows Job anchor has not started"));
}
sequence += 1;
const message: ServiceChildAnchorMessage = {
...payload,
generation: start.generation,
sequence,
};
const delivery = outboundQueue.then(() => sendProcessMessage(message));
outboundQueue = delivery.catch(() => {});
return delivery;
};
const deliver = async (payload: ServiceChildAnchorPayload): Promise<void> => {
if (!process.connected) {
if (!closeReason) {
void requestCleanup("parent-lost");
}
return;
}
try {
await send(payload);
} catch (error) {
if (process.connected) {
throw error;
}
if (!closeReason) {
void requestCleanup("parent-lost");
}
}
};
const stopLifecycle = () => {
clearTimeout(lifecycleTimer);
clearImmediate(lifecycleImmediate);
lifecycleTimer = undefined;
lifecycleImmediate = undefined;
};
const finishAnchor = (exitCode: number) => {
stopLifecycle();
process.exitCode = exitCode;
if (process.connected) {
process.disconnect?.();
}
};
const closeOutputHandle = (stream: OutputStream) => {
const handle = stream.handle;
if (handle === undefined) {
return;
}
if (!bindings) {
throw new Error(`${stream.name} output bindings were not initialized`);
}
if (!bindings.CloseHandle(handle)) {
throw bindings.lastError(`CloseHandle(${stream.name} pipe)`);
}
stream.handle = undefined;
};
const closeNativeHandles = () => {
let closeError: Error | undefined;
for (const stream of outputStreams) {
try {
closeOutputHandle(stream);
} catch (error) {
closeError ??= error instanceof Error ? error : new Error(errorMessage(error));
}
}
for (const handle of [processHandle, job]) {
if (handle !== undefined && bindings && !bindings.CloseHandle(handle)) {
const error = bindings.lastError("CloseHandle");
closeError ??= error;
}
}
processHandle = undefined;
job = undefined;
if (closeError) {
throw closeError;
}
};
const closeAuthority = async (reason: ClosingReason) => {
if (state === "closed") {
return;
}
state = "closed";
stopLifecycle();
try {
if (process.connected) {
await send({ type: "closing", reason });
}
closeNativeHandles();
cleanupFinished.resolve();
finishAnchor(0);
} catch (error) {
try {
closeNativeHandles();
} catch {
// Preserve the original delivery/close failure while KILL_ON_JOB_CLOSE still owns cleanup.
}
cleanupFinished.reject(error);
finishAnchor(1);
}
};
const failAuthority = async (error: unknown) => {
if (state === "closed") {
return;
}
state = "closed";
stopLifecycle();
try {
if (process.connected && start) {
await send({ type: "result-error", error: errorMessage(error) }).catch(() => {});
}
closeNativeHandles();
} catch {
// An owner failure never receives a closing receipt or claims unobserved extinction.
} finally {
cleanupFinished.reject(error);
finishAnchor(1);
}
};
const finishOutput = async (stream: OutputStream) => {
if (stream.ended) {
return;
}
closeOutputHandle(stream);
stream.ended = true;
const tail = stream.decoder?.flush();
if (tail) {
await deliver({ type: "output", stream: stream.name, chunk: tail });
}
await deliver({ type: "output-end", stream: stream.name });
};
const observeOutput = async (stream: OutputStream): Promise<boolean> => {
if (stream.ended) {
return false;
}
if (!bindings || stream.handle === undefined || !stream.decoder) {
throw new Error(`${stream.name} output ownership was not initialized`);
}
const available = [0];
if (!bindings.PeekNamedPipe(stream.handle, null, 0, null, available, null)) {
const errorCode = bindings.getLastErrorCode();
if (errorCode !== ERROR_BROKEN_PIPE) {
throw new Error(`PeekNamedPipe(${stream.name}) failed (Win32 error ${errorCode})`);
}
await finishOutput(stream);
return true;
}
const availableBytes = available[0];
if (
typeof availableBytes !== "number" ||
!Number.isSafeInteger(availableBytes) ||
availableBytes < 0
) {
throw new Error(`PeekNamedPipe(${stream.name}) returned an invalid byte count`);
}
if (availableBytes === 0) {
return false;
}
const requestedBytes = Math.min(availableBytes, outputBuffer.length);
const bytesRead = [0];
// This anchor owns the only read handle, so a read capped at the peeked bytes cannot block.
if (!bindings.ReadFile(stream.handle, outputBuffer, requestedBytes, bytesRead, null)) {
const errorCode = bindings.getLastErrorCode();
if (errorCode !== ERROR_BROKEN_PIPE) {
throw new Error(`ReadFile(${stream.name}) failed (Win32 error ${errorCode})`);
}
await finishOutput(stream);
return true;
}
const count = bytesRead[0];
if (
typeof count !== "number" ||
!Number.isSafeInteger(count) ||
count < 0 ||
count > requestedBytes
) {
throw new Error(`ReadFile(${stream.name}) returned an invalid byte count`);
}
if (count === 0) {
throw new Error(`ReadFile(${stream.name}) returned no available bytes`);
}
const text = stream.decoder.decode(outputBuffer.subarray(0, count));
if (text) {
await deliver({ type: "output", stream: stream.name, chunk: text });
}
return true;
};
const observeRoot = async (): Promise<boolean> => {
if (rootObserved) {
return false;
}
if (!bindings || !processHandle) {
throw new Error("Windows root process ownership was not initialized");
}
const waitResult = bindings.WaitForSingleObject(processHandle, 0);
if (waitResult === WAIT_TIMEOUT) {
return false;
}
if (waitResult === WAIT_FAILED) {
throw bindings.lastError("WaitForSingleObject(root)");
}
if (waitResult !== WAIT_OBJECT_0) {
throw new Error(`WaitForSingleObject(root) returned unexpected result ${waitResult}`);
}
const exitCode = [0];
if (!bindings.GetExitCodeProcess(processHandle, exitCode)) {
throw bindings.lastError("GetExitCodeProcess");
}
const code = exitCode[0];
if (typeof code !== "number" || !Number.isSafeInteger(code) || code < 0 || code > 0xffff_ffff) {
throw new Error("GetExitCodeProcess returned an invalid exit code");
}
rootObserved = true;
await deliver({ type: "root-result", code, signal: null });
return true;
};
const observeJob = (): boolean => {
if (extinctionProven) {
return false;
}
if (!bindings || !job) {
throw new Error("Windows Job ownership was not initialized");
}
const accounting: { ActiveProcesses?: unknown } = {};
if (
!bindings.QueryInformationJobObject(
job,
JOB_OBJECT_BASIC_ACCOUNTING_INFORMATION,
accounting,
bindings.basicAccountingSize,
null,
)
) {
throw bindings.lastError("QueryInformationJobObject");
}
const count = accounting.ActiveProcesses;
if (typeof count !== "number" || !Number.isSafeInteger(count) || count < 0) {
throw new Error("QueryInformationJobObject returned an invalid active process count");
}
extinctionProven = count === 0;
return extinctionProven;
};
const runLifecycle = async () => {
if (lifecycleRunning || state === "closed" || !processHandle) {
lifecycleRerun ||= lifecycleRunning;
return;
}
lifecycleRunning = true;
let advanced = false;
try {
advanced = (await observeRoot()) || advanced;
// The admitted root itself proves the Job is nonempty until its exact HANDLE signals.
advanced = (rootObserved && observeJob()) || advanced;
for (let round = 0; round < OUTPUT_ROUNDS_PER_TURN; round += 1) {
let outputAdvanced = false;
for (const stream of outputStreams) {
outputAdvanced = (await observeOutput(stream)) || outputAdvanced;
}
advanced ||= outputAdvanced;
if (!outputAdvanced) {
break;
}
}
if (extinctionProven && rootObserved && outputStreams.every((stream) => stream.ended)) {
await closeAuthority(closeReason ?? "lineage-closed");
return;
}
} catch (error) {
await failAuthority(error);
return;
} finally {
lifecycleRunning = false;
}
const immediate = advanced || lifecycleRerun;
lifecycleRerun = false;
scheduleLifecycle(immediate);
};
const scheduleLifecycle = (immediate: boolean) => {
if (state === "closed" || !processHandle) {
return;
}
if (lifecycleRunning) {
lifecycleRerun ||= immediate;
return;
}
if (immediate && lifecycleTimer) {
clearTimeout(lifecycleTimer);
lifecycleTimer = undefined;
}
if (lifecycleTimer || lifecycleImmediate) {
return;
}
if (immediate) {
lifecycleImmediate = setImmediate(() => {
lifecycleImmediate = undefined;
void runLifecycle();
});
return;
}
lifecycleTimer = setTimeout(() => {
lifecycleTimer = undefined;
void runLifecycle();
}, IDLE_OBSERVATION_MS);
// A disconnected IPC channel cannot retain Node; cleanup must retain its exact Job owner.
if (state === "active" && process.connected) {
lifecycleTimer.unref();
}
};
const requestCleanup = (reason: "cancel" | "parent-lost" | "lineage-lost"): Promise<void> => {
if (state === "closed") {
return cleanupFinished.promise;
}
closeReason ??= reason;
state = "closing";
if (!processHandle) {
void closeAuthority(reason);
return cleanupFinished.promise;
}
if (!terminationRequested && !extinctionProven) {
terminationRequested = true;
if (!bindings || !job) {
void failAuthority(new Error("Windows Job cleanup authority was not initialized"));
return cleanupFinished.promise;
}
if (!bindings.TerminateJobObject(job, 1)) {
const error = bindings.lastError("TerminateJobObject");
void failAuthority(error);
return cleanupFinished.promise;
}
}
scheduleLifecycle(true);
return cleanupFinished.promise;
};
const reportStartupError = async (error: unknown) => {
if (!process.connected) {
return;
}
await send({ type: "startup-error", error: errorMessage(error) });
await startupErrorAcknowledged.promise;
};
const startCommand = async (next: ServiceChildStart) => {
start = next;
if (typeof next.windowsShellCommand !== "string") {
state = "closed";
finishAnchor(1);
return;
}
try {
const koffi = (await import("koffi")).default;
if (state !== "starting") {
return;
}
bindings = createWindowsJobBindings(koffi);
bindings.assertLayouts();
job = bindings.requireHandle(bindings.CreateJobObjectW(null, null), "CreateJobObjectW");
if (
!bindings.SetExtendedLimits(
job,
JOB_OBJECT_EXTENDED_LIMIT_INFORMATION,
bindings.extendedLimits,
bindings.extendedLimitsSize,
)
) {
throw bindings.lastError("SetInformationJobObject(KILL_ON_JOB_CLOSE)");
}
const commandStdio = bindings.createCommandStdio();
pendingCommandStdio = commandStdio;
let processAttributes:
| ReturnType<WindowsJobBindings["createProcessAttributeList"]>
| undefined;
const processInfo: Record<string, unknown> = {};
try {
processAttributes = bindings.createProcessAttributeList(commandStdio.inheritedHandles, job);
const shell =
resolveEnvironmentValue(next.env, "COMSPEC", "win32") || getWindowsCmdExePath(next.env);
const commandLine = Buffer.from(
`"${shell}" /d /s /c "${next.windowsShellCommand}"\0`,
"utf16le",
);
if (
!bindings.CreateProcessW(
shell,
commandLine,
null,
null,
1,
CREATE_NEW_PROCESS_GROUP | CREATE_UNICODE_ENVIRONMENT | EXTENDED_STARTUPINFO_PRESENT,
buildWindowsJobEnvironmentBlock(next.env),
next.cwd ?? null,
{
StartupInfo: {
cb: bindings.startupInfoExSize,
dwFlags: STARTF_USESTDHANDLES,
hStdInput: commandStdio.stdinHandle,
hStdOutput: commandStdio.stdoutWriteHandle,
hStdError: commandStdio.stderrWriteHandle,
},
lpAttributeList: processAttributes.attributeList,
},
processInfo,
)
) {
throw bindings.lastError("CreateProcessW(JOB_LIST)");
}
// JOB_LIST makes containment atomic: a successful root never exists outside its Job.
processHandle = bindings.requireHandle(processInfo.hProcess, "CreateProcessW process");
const threadHandle = bindings.requireHandle(processInfo.hThread, "CreateProcessW thread");
if (!bindings.CloseHandle(threadHandle)) {
throw bindings.lastError("CloseHandle(command thread)");
}
} finally {
processAttributes?.release();
pendingCommandStdio?.closeChildHandles();
}
const commandPid = Number(processInfo.dwProcessId);
const handles = commandStdio.takeOutputReadHandles();
// Admit both transferred HANDLEs before decoder construction can throw.
outputStreams.push(
{ name: "stdout", handle: handles.stdoutReadHandle, ended: false },
{ name: "stderr", handle: handles.stderrReadHandle, ended: false },
);
for (const stream of outputStreams) {
stream.decoder = createWindowsOutputDecoder();
}
commandStdio.close();
pendingCommandStdio = undefined;
state = "active";
const readyDelivery = send({ type: "ready", commandPid, anchorPid: process.pid });
// Queue ready before polling so an instantly exiting command cannot overtake admission.
scheduleLifecycle(true);
await readyDelivery;
} catch (error) {
if (state === "closed") {
return;
}
if (state === "closing") {
await cleanupFinished.promise.catch(() => {});
return;
}
if (processHandle && outputStreams.some((stream) => !stream.decoder)) {
for (const stream of outputStreams) {
closeOutputHandle(stream);
stream.ended = true;
}
}
await reportStartupError(error);
await (processHandle ? requestCleanup("lineage-lost") : closeAuthority("lineage-lost"));
} finally {
pendingCommandStdio?.close();
pendingCommandStdio = undefined;
}
};
process.once("disconnect", () => {
startupErrorAcknowledged.resolve();
if (!start) {
state = "closed";
process.exitCode = 1;
return;
}
void requestCleanup("parent-lost");
});
process.once("SIGTERM", () => void requestCleanup("parent-lost"));
process.once("SIGINT", () => void requestCleanup("parent-lost"));
process.on("message", (raw: unknown) => {
if (isWindowsJobServiceStart(raw) && start === undefined && state === "starting") {
void startCommand(raw);
return;
}
const message = asOptionalRecord(raw);
if (
!start ||
state === "closed" ||
!message ||
(message.type !== "cancel" && message.type !== "startup-error-ack") ||
typeof message.generation !== "string" ||
typeof message.sequence !== "number" ||
message.generation !== start.generation ||
message.sequence <= lastHostSequence
) {
if (start && state !== "closed") {
void requestCleanup("lineage-lost");
}
return;
}
lastHostSequence = message.sequence;
if (message.type === "startup-error-ack") {
startupErrorAcknowledged.resolve();
} else {
void requestCleanup("cancel");
}
});
}
runServiceChildWindowsJobAnchor();
@@ -0,0 +1,379 @@
type NativeHandle = bigint;
const JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x0000_2000;
const PROC_THREAD_ATTRIBUTE_HANDLE_LIST = 0x0002_0002;
const PROC_THREAD_ATTRIBUTE_JOB_LIST = 0x0002_000d;
const HANDLE_FLAG_INHERIT = 0x0000_0001;
const GENERIC_READ = 0x8000_0000;
const FILE_SHARE_READ = 0x0000_0001;
const FILE_SHARE_WRITE = 0x0000_0002;
const OPEN_EXISTING = 3;
const FILE_ATTRIBUTE_NORMAL = 0x0000_0080;
export function createWindowsJobBindings(koffi: typeof import("koffi").default) {
if (process.arch !== "x64" && process.arch !== "arm64") {
throw new Error(`Windows Job command ownership requires x64 or arm64, got ${process.arch}`);
}
const kernel32 = koffi.load("kernel32.dll");
const HANDLE = koffi.pointer("HANDLE", koffi.opaque());
const VOID_POINTER = koffi.pointer("VOID_POINTER", koffi.opaque());
const SECURITY_ATTRIBUTES = koffi.struct("SECURITY_ATTRIBUTES", {
nLength: "uint32_t",
lpSecurityDescriptor: VOID_POINTER,
bInheritHandle: "int32_t",
});
const BASIC_LIMITS = koffi.struct("JOBOBJECT_BASIC_LIMIT_INFORMATION", {
PerProcessUserTimeLimit: "int64_t",
PerJobUserTimeLimit: "int64_t",
LimitFlags: "uint32_t",
MinimumWorkingSetSize: "uintptr_t",
MaximumWorkingSetSize: "uintptr_t",
ActiveProcessLimit: "uint32_t",
Affinity: "uintptr_t",
PriorityClass: "uint32_t",
SchedulingClass: "uint32_t",
});
const IO_COUNTERS = koffi.struct("IO_COUNTERS", {
ReadOperationCount: "uint64_t",
WriteOperationCount: "uint64_t",
OtherOperationCount: "uint64_t",
ReadTransferCount: "uint64_t",
WriteTransferCount: "uint64_t",
OtherTransferCount: "uint64_t",
});
const EXTENDED_LIMITS = koffi.struct("JOBOBJECT_EXTENDED_LIMIT_INFORMATION", {
BasicLimitInformation: BASIC_LIMITS,
IoInfo: IO_COUNTERS,
ProcessMemoryLimit: "uintptr_t",
JobMemoryLimit: "uintptr_t",
PeakProcessMemoryUsed: "uintptr_t",
PeakJobMemoryUsed: "uintptr_t",
});
const BASIC_ACCOUNTING = koffi.struct("JOBOBJECT_BASIC_ACCOUNTING_INFORMATION", {
TotalUserTime: "int64_t",
TotalKernelTime: "int64_t",
ThisPeriodTotalUserTime: "int64_t",
ThisPeriodTotalKernelTime: "int64_t",
TotalPageFaultCount: "uint32_t",
TotalProcesses: "uint32_t",
ActiveProcesses: "uint32_t",
TotalTerminatedProcesses: "uint32_t",
});
const STARTUPINFO = koffi.struct("STARTUPINFOW", {
cb: "uint32_t",
lpReserved: VOID_POINTER,
lpDesktop: VOID_POINTER,
lpTitle: VOID_POINTER,
dwX: "uint32_t",
dwY: "uint32_t",
dwXSize: "uint32_t",
dwYSize: "uint32_t",
dwXCountChars: "uint32_t",
dwYCountChars: "uint32_t",
dwFillAttribute: "uint32_t",
dwFlags: "uint32_t",
wShowWindow: "uint16_t",
cbReserved2: "uint16_t",
lpReserved2: VOID_POINTER,
hStdInput: HANDLE,
hStdOutput: HANDLE,
hStdError: HANDLE,
});
const STARTUPINFOEX = koffi.struct("STARTUPINFOEXW", {
StartupInfo: STARTUPINFO,
lpAttributeList: VOID_POINTER,
});
const PROCESS_INFORMATION = koffi.struct("PROCESS_INFORMATION", {
hProcess: HANDLE,
hThread: HANDLE,
dwProcessId: "uint32_t",
dwThreadId: "uint32_t",
});
const getLastErrorCode = kernel32.func("__stdcall", "GetLastError", "uint32_t", []);
const CloseHandle = kernel32.func("__stdcall", "CloseHandle", "int32_t", [HANDLE]);
const CreateJobObjectW = kernel32.func("__stdcall", "CreateJobObjectW", HANDLE, [
VOID_POINTER,
"str16",
]);
const SetExtendedLimits = kernel32.func("__stdcall", "SetInformationJobObject", "int32_t", [
HANDLE,
"int32_t",
koffi.pointer(EXTENDED_LIMITS),
"uint32_t",
]);
const CreatePipe = kernel32.func("__stdcall", "CreatePipe", "int32_t", [
koffi.out(koffi.pointer(HANDLE)),
koffi.out(koffi.pointer(HANDLE)),
koffi.pointer(SECURITY_ATTRIBUTES),
"uint32_t",
]);
const SetHandleInformation = kernel32.func("__stdcall", "SetHandleInformation", "int32_t", [
HANDLE,
"uint32_t",
"uint32_t",
]);
const CreateFileW = kernel32.func("__stdcall", "CreateFileW", HANDLE, [
"str16",
"uint32_t",
"uint32_t",
koffi.pointer(SECURITY_ATTRIBUTES),
"uint32_t",
"uint32_t",
HANDLE,
]);
const InitializeProcThreadAttributeList = kernel32.func(
"__stdcall",
"InitializeProcThreadAttributeList",
"int32_t",
[VOID_POINTER, "uint32_t", "uint32_t", koffi.inout(koffi.pointer("uintptr_t"))],
);
const UpdateProcThreadAttribute = kernel32.func(
"__stdcall",
"UpdateProcThreadAttribute",
"int32_t",
[VOID_POINTER, "uint32_t", "uintptr_t", VOID_POINTER, "uintptr_t", VOID_POINTER, VOID_POINTER],
);
const DeleteProcThreadAttributeList = kernel32.func(
"__stdcall",
"DeleteProcThreadAttributeList",
"void",
[VOID_POINTER],
);
const CreateProcessW = kernel32.func("__stdcall", "CreateProcessW", "int32_t", [
"str16",
koffi.pointer("uint16_t"),
VOID_POINTER,
VOID_POINTER,
"int32_t",
"uint32_t",
VOID_POINTER,
"str16",
koffi.pointer(STARTUPINFOEX),
koffi.out(koffi.pointer(PROCESS_INFORMATION)),
]);
const WaitForSingleObject = kernel32.func("__stdcall", "WaitForSingleObject", "uint32_t", [
HANDLE,
"uint32_t",
]);
const GetExitCodeProcess = kernel32.func("__stdcall", "GetExitCodeProcess", "int32_t", [
HANDLE,
koffi.out(koffi.pointer("uint32_t")),
]);
const QueryInformationJobObject = kernel32.func(
"__stdcall",
"QueryInformationJobObject",
"int32_t",
[HANDLE, "int32_t", koffi.out(koffi.pointer(BASIC_ACCOUNTING)), "uint32_t", VOID_POINTER],
);
const PeekNamedPipe = kernel32.func("__stdcall", "PeekNamedPipe", "int32_t", [
HANDLE,
VOID_POINTER,
"uint32_t",
VOID_POINTER,
koffi.out(koffi.pointer("uint32_t")),
VOID_POINTER,
]);
const ReadFile = kernel32.func("__stdcall", "ReadFile", "int32_t", [
HANDLE,
koffi.out(koffi.pointer("uint8_t")),
"uint32_t",
koffi.out(koffi.pointer("uint32_t")),
VOID_POINTER,
]);
const TerminateJobObject = kernel32.func("__stdcall", "TerminateJobObject", "int32_t", [
HANDLE,
"uint32_t",
]);
const lastError = (operation: string) =>
new Error(`${operation} failed (Win32 error ${getLastErrorCode()})`);
const requireHandle = (value: unknown, operation: string): NativeHandle => {
if (typeof value !== "bigint" || value === 0n) {
throw lastError(operation);
}
return value;
};
const createCommandStdio = () => {
let stdinHandle: NativeHandle | undefined;
const outputPipes: Record<"stdout" | "stderr", { read?: NativeHandle; write?: NativeHandle }> =
{
stdout: {},
stderr: {},
};
const securityAttributes = {
nLength: koffi.sizeof(SECURITY_ATTRIBUTES),
lpSecurityDescriptor: null,
bInheritHandle: 1,
};
const closeChildHandles = () => {
for (const handle of [stdinHandle, outputPipes.stdout.write, outputPipes.stderr.write]) {
if (handle !== undefined) {
CloseHandle(handle);
}
}
stdinHandle = undefined;
delete outputPipes.stdout.write;
delete outputPipes.stderr.write;
};
const closeRawReadHandles = () => {
for (const pipe of Object.values(outputPipes)) {
const handle = pipe.read;
if (handle !== undefined) {
CloseHandle(handle);
}
delete pipe.read;
}
};
try {
for (const [streamName, pipe] of Object.entries(outputPipes)) {
const read: Array<bigint | null> = [null];
const write: Array<bigint | null> = [null];
if (!CreatePipe(read, write, securityAttributes, 0)) {
throw lastError(`CreatePipe(${streamName})`);
}
// Record a returned write HANDLE before validating its peer so partial failures cannot leak.
if (typeof write[0] === "bigint" && write[0] !== 0n) {
pipe.write = write[0];
}
pipe.read = requireHandle(read[0], `CreatePipe(${streamName} read)`);
pipe.write = requireHandle(write[0], `CreatePipe(${streamName} write)`);
if (!SetHandleInformation(pipe.read, HANDLE_FLAG_INHERIT, 0)) {
throw lastError(`SetHandleInformation(${streamName} read)`);
}
}
const openedStdin = CreateFileW(
"NUL",
GENERIC_READ,
FILE_SHARE_READ | FILE_SHARE_WRITE,
securityAttributes,
OPEN_EXISTING,
FILE_ATTRIBUTE_NORMAL,
null,
);
if (openedStdin === 0xffff_ffff_ffff_ffffn) {
throw lastError("CreateFileW(NUL)");
}
const childStdin = requireHandle(openedStdin, "CreateFileW(NUL)");
stdinHandle = childStdin;
const childStdout = outputPipes.stdout.write;
const childStderr = outputPipes.stderr.write;
if (childStdout === undefined || childStderr === undefined) {
throw new Error("Windows command output handles were not initialized");
}
return {
inheritedHandles: [childStdin, childStdout, childStderr],
stdinHandle: childStdin,
stdoutWriteHandle: childStdout,
stderrWriteHandle: childStderr,
closeChildHandles,
takeOutputReadHandles: () => {
if (outputPipes.stdout.read === undefined || outputPipes.stderr.read === undefined) {
throw new Error("Windows command output handles were already transferred");
}
const output = {
stdoutReadHandle: outputPipes.stdout.read,
stderrReadHandle: outputPipes.stderr.read,
};
delete outputPipes.stdout.read;
delete outputPipes.stderr.read;
return output;
},
close: () => {
closeChildHandles();
closeRawReadHandles();
},
};
} catch (error) {
closeChildHandles();
closeRawReadHandles();
throw error;
}
};
const createProcessAttributeList = (handles: NativeHandle[], job: NativeHandle) => {
const size = [0n];
InitializeProcThreadAttributeList(null, 2, 0, size);
const attributeListSize = size[0] ?? 0n;
if (attributeListSize <= 0n) {
throw lastError("InitializeProcThreadAttributeList(size)");
}
const attributeList = Buffer.alloc(Number(attributeListSize));
if (!InitializeProcThreadAttributeList(attributeList, 2, 0, size)) {
throw lastError("InitializeProcThreadAttributeList");
}
const backingLists: Buffer[] = [];
const release = () => {
DeleteProcThreadAttributeList(attributeList);
backingLists.length = 0;
};
try {
// Both HANDLE arrays must outlive DeleteProcThreadAttributeList and CreateProcessW.
for (const { attribute, values, name } of [
{ attribute: PROC_THREAD_ATTRIBUTE_HANDLE_LIST, values: handles, name: "HANDLE_LIST" },
{ attribute: PROC_THREAD_ATTRIBUTE_JOB_LIST, values: [job], name: "JOB_LIST" },
]) {
const backingList = Buffer.alloc(koffi.sizeof(HANDLE) * values.length);
backingLists.push(backingList);
koffi.encode(backingList, HANDLE, values, values.length);
if (
!UpdateProcThreadAttribute(
attributeList,
0,
attribute,
backingList,
koffi.sizeof(HANDLE) * values.length,
null,
null,
)
) {
throw lastError(`UpdateProcThreadAttribute(${name})`);
}
}
return { attributeList, release };
} catch (error) {
release();
throw error;
}
};
return {
CreateJobObjectW,
SetExtendedLimits,
CreateProcessW,
WaitForSingleObject,
GetExitCodeProcess,
QueryInformationJobObject,
PeekNamedPipe,
ReadFile,
TerminateJobObject,
CloseHandle,
getLastErrorCode,
lastError,
requireHandle,
createCommandStdio,
createProcessAttributeList,
assertLayouts: () => {
const actual = [
koffi.sizeof(STARTUPINFO),
koffi.sizeof(STARTUPINFOEX),
koffi.sizeof(PROCESS_INFORMATION),
koffi.sizeof(BASIC_LIMITS),
koffi.sizeof(EXTENDED_LIMITS),
koffi.sizeof(BASIC_ACCOUNTING),
koffi.offsetof(BASIC_ACCOUNTING, "ActiveProcesses"),
koffi.sizeof(SECURITY_ATTRIBUTES),
];
const expected = [104, 112, 24, 64, 144, 48, 40, 24];
if (actual.some((value, index) => value !== expected[index])) {
throw new Error(`Koffi Win32 layout mismatch: ${actual.join(",")}`);
}
},
extendedLimits: {
BasicLimitInformation: {
LimitFlags: JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
},
},
extendedLimitsSize: koffi.sizeof(EXTENDED_LIMITS),
basicAccountingSize: koffi.sizeof(BASIC_ACCOUNTING),
startupInfoExSize: koffi.sizeof(STARTUPINFOEX),
};
}
@@ -0,0 +1,40 @@
import { asOptionalRecord, isStringRecord } from "@openclaw/normalization-core/record-coerce";
import { mergeProcessEnv } from "../../infra/process-env.js";
import type { ServiceChildStart } from "./service-child-protocol.js";
export function buildWindowsJobEnvironmentBlock(env: Record<string, string> | undefined): Buffer {
const merged = mergeProcessEnv([env], "win32");
for (const [key, value] of Object.entries(merged)) {
if (key.includes("\0") || value.includes("\0")) {
throw new Error("owned command environment contains a NUL byte");
}
}
const entries = Object.keys(merged)
.toSorted((left, right) => {
const leftFolded = left.toUpperCase();
const rightFolded = right.toUpperCase();
return leftFolded < rightFolded ? -1 : leftFolded > rightFolded ? 1 : 0;
})
.map((key) => `${key}=${merged[key]}`);
return Buffer.from(`${entries.join("\0")}\0\0`, "utf16le");
}
export function isWindowsJobServiceStart(value: unknown): value is ServiceChildStart {
const message = asOptionalRecord(value);
return Boolean(
message &&
message.type === "start" &&
typeof message.generation === "string" &&
typeof message.command === "string" &&
Array.isArray(message.args) &&
message.args.every((arg) => typeof arg === "string") &&
(message.cwd === undefined || typeof message.cwd === "string") &&
(message.env === undefined || isStringRecord(message.env)) &&
(message.stdinMode === "inherit" ||
message.stdinMode === "pipe-open" ||
message.stdinMode === "pipe-closed") &&
(message.secretFd === undefined || typeof message.secretFd === "number") &&
(message.controlFd === undefined || typeof message.controlFd === "number") &&
(message.windowsShellCommand === undefined || typeof message.windowsShellCommand === "string"),
);
}
@@ -0,0 +1,193 @@
import { writeFile } from "node:fs/promises";
import { createRequire } from "node:module";
import path from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import { isProcessAlive, waitForDead, waitForPidFile } from "../../../test/helpers/process-wait.js";
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
import { createWindowsOutputDecoder } from "../../infra/windows-encoding.js";
import { getWindowsCmdExePath } from "../../infra/windows-install-roots.js";
import { killPidIfAlive } from "../../test-utils/process-tree.js";
import { createProcessSupervisor } from "./supervisor.js";
const activePids = new Set<number>();
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
afterEach(async () => {
for (const pid of activePids) {
killPidIfAlive(pid);
}
await Promise.all([...activePids].map((pid) => waitForDead(pid, 5_000).catch(() => {})));
activePids.clear();
});
async function createDescendantScope() {
const cwd = tempDirs.make("openclaw-anchored-shell-");
const descendantPath = path.join(cwd, "descendant.cjs");
const descendantPidPath = path.join(cwd, "descendant.pid");
const releasePath = path.join(cwd, "descendant.release");
const rootPath = path.join(cwd, "root.cjs");
await writeFile(
descendantPath,
`
const { existsSync, writeFileSync } = require("node:fs");
const releaseTimer = setInterval(() => {
if (existsSync(process.argv[2])) {
clearInterval(releaseTimer);
}
}, 20);
writeFileSync(process.argv[3], String(process.pid));
`,
"utf8",
);
if (process.platform === "win32") {
const koffiPath = createRequire(import.meta.url).resolve("koffi");
await writeFile(
rootPath,
`
const koffi = require(${JSON.stringify(koffiPath)});
const kernel32 = koffi.load("kernel32.dll");
const handle = koffi.pointer("ANCHORED_SHELL_FIXTURE_HANDLE", koffi.opaque());
const bytes = koffi.pointer("uint8_t");
const createProcess = kernel32.func("__stdcall", "CreateProcessW", "int32_t", [
"str16", koffi.pointer("uint16_t"), "void *", "void *", "int32_t", "uint32_t",
"void *", "str16", bytes, bytes,
]);
const closeHandle = kernel32.func("__stdcall", "CloseHandle", "int32_t", [handle]);
const getLastError = kernel32.func("__stdcall", "GetLastError", "uint32_t", []);
// Production only supports x64/arm64, where these Win32 structures are 104/24 bytes.
const startupInfo = Buffer.alloc(104);
const processInfo = Buffer.alloc(24);
startupInfo.writeUInt32LE(startupInfo.length, 0);
const commandLine = Buffer.from(
[
process.execPath,
${JSON.stringify(descendantPath)},
${JSON.stringify(releasePath)},
${JSON.stringify(descendantPidPath)},
].map((value) => '"' + value + '"').join(" ") + String.fromCharCode(0),
"utf16le",
);
// Native creation avoids libuv's private Job; no inherited handles guarantees pipe EOF.
if (!createProcess(
process.execPath, commandLine, null, null, 0, 0x08000000, null, null,
startupInfo, processInfo,
)) {
throw new Error("fixture CreateProcessW failed (Win32 error " + getLastError() + ")");
}
for (const offset of [8, 0]) {
if (!closeHandle(processInfo.readBigUInt64LE(offset))) {
throw new Error("fixture CloseHandle failed (Win32 error " + getLastError() + ")");
}
}
${fragmentedOutputFixture()}
`,
"utf8",
);
} else {
await writeFile(
rootPath,
`
const { spawn } = require("node:child_process");
const child = spawn(process.execPath, [${JSON.stringify(descendantPath)}, ${JSON.stringify(releasePath)}, ${JSON.stringify(descendantPidPath)}], {
stdio: ["ignore", "ignore", "ignore", 3],
});
child.unref();
${fragmentedOutputFixture()}
`,
"utf8",
);
}
const supervisor = createProcessSupervisor();
const scopeKey = `anchored-shell:${cwd}`;
const run = await supervisor.spawn({
mode: "anchored-shell",
command: "node root.cjs",
sessionId: "anchored-shell-real",
backendId: "anchored-shell-real",
scopeKey,
cwd,
env:
process.platform === "win32"
? {
...process.env,
COMSPEC: getWindowsCmdExePath(process.env),
ComSpec: "Z:\\invalid-later-duplicate\\cmd.exe",
}
: process.env,
});
return {
run,
supervisor,
scopeKey,
readPid: () => waitForPidFile(descendantPidPath, 5_000),
release: () => writeFile(releasePath, "", "utf8"),
};
}
function fragmentedOutputFixture(): string {
return `
process.stdout.write("owned-stdout-one\\nowned-stdout-two\\n");
process.stderr.write("owned-stderr-one\\nowned-stderr-two\\n");
process.stdout.write(Buffer.from([0xf0, 0x9f]));
process.stderr.write(Buffer.from([0xf0, 0x9f]));
setTimeout(() => {
process.stdout.write(Buffer.from([0x98, 0x80, 0xe2, 0x82]));
process.stderr.write(Buffer.from([0x98, 0x80, 0xe2, 0x82]));
}, 50);
`;
}
async function expectPending(promise: Promise<void>) {
const settled = await Promise.race([
promise.then(() => true),
new Promise<false>((resolve) => {
setImmediate(() => resolve(false));
}),
]);
expect(settled).toBe(false);
}
describe("supervisor anchored shell real process ownership", () => {
it.each([
{ name: "cancels retained descendants idempotently", cancel: true },
{ name: "releases ownership after descendants exit naturally", cancel: false },
])("$name after root settlement and fragmented output flush", async ({ cancel }) => {
const { run, supervisor, scopeKey, readPid, release } = await createDescendantScope();
const result = await run.wait();
const decoder = createWindowsOutputDecoder();
const finalTail = decoder.decode(Buffer.from([0xe2, 0x82])) + decoder.flush();
expect(result).toMatchObject({ reason: "exit", exitCode: 0, exitSignal: null });
expect(finalTail).not.toBe("");
expect(result.stdout.replaceAll("\r\n", "\n")).toBe(
`owned-stdout-one\nowned-stdout-two\n😀${finalTail}`,
);
expect(result.stderr.replaceAll("\r\n", "\n")).toBe(
`owned-stderr-one\nowned-stderr-two\n😀${finalTail}`,
);
const descendantPid = await readPid();
activePids.add(descendantPid);
expect(descendantPid).toBeGreaterThan(0);
expect(isProcessAlive(descendantPid)).toBe(true);
await expectPending(run.waitForExtinction!());
if (cancel) {
supervisor.cancelScope(scopeKey);
supervisor.cancelScope(scopeKey);
} else {
await release();
}
await Promise.all([
run.waitForExtinction!(),
supervisor.waitForScope(scopeKey),
supervisor.waitForScope(scopeKey),
]);
expect(supervisor.getRecord(run.runId)).toMatchObject({
state: "exited",
terminationReason: "exit",
exitCode: 0,
});
await waitForDead(descendantPid, 5_000);
});
});
@@ -0,0 +1,175 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../../test/helpers/promise.js";
import {
createSilentIdleArgv,
createStubChildAdapter,
createWriteStdoutArgv,
spawnChild,
type StubChildAdapter,
} from "./supervisor.test-support.js";
const { createChildAdapterMock, createPtyAdapterMock } = vi.hoisted(() => ({
createChildAdapterMock: vi.fn(),
createPtyAdapterMock: vi.fn(),
}));
vi.mock("./adapters/child.js", () => ({
createChildAdapter: createChildAdapterMock,
}));
vi.mock("./adapters/pty.js", () => ({
createPtyAdapter: createPtyAdapterMock,
}));
let createProcessSupervisor: typeof import("./supervisor.js").createProcessSupervisor;
describe("process supervisor scope extinction", () => {
beforeEach(async () => {
vi.resetModules();
({ createProcessSupervisor } = await import("./supervisor.js"));
createChildAdapterMock.mockReset();
createPtyAdapterMock.mockReset();
vi.useRealTimers();
});
afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});
it("keeps extinction waiting optional for ordinary child adapters", async () => {
const adapter = createStubChildAdapter();
createChildAdapterMock.mockResolvedValue(adapter);
const supervisor = createProcessSupervisor();
const run = await spawnChild(supervisor, {
sessionId: "ordinary-child",
scopeKey: "scope:ordinary-child",
argv: createSilentIdleArgv(),
});
expect(run.waitForExtinction).toBeUndefined();
const drain = supervisor.waitForScope("scope:ordinary-child");
const drained = vi.fn();
void drain.then(drained);
await Promise.resolve();
expect(drained).not.toHaveBeenCalled();
adapter.settle(0);
await expect(run.wait()).resolves.toMatchObject({ reason: "exit", exitCode: 0 });
await expect(drain).resolves.toBeUndefined();
expect(adapter.disposeMock).toHaveBeenCalledOnce();
});
it("preserves root output when authoritative extinction settles first", async () => {
const adapter = createStubChildAdapter();
adapter.oomScoreWrapperSelected = true;
const extinction = createDeferred();
adapter.waitForExtinction = async () => await extinction.promise;
createChildAdapterMock.mockResolvedValue(adapter);
const supervisor = createProcessSupervisor();
const run = await spawnChild(supervisor, {
sessionId: "s1",
argv: createWriteStdoutArgv("ok"),
timeoutMs: 1_000,
stdinMode: "pipe-closed",
});
expect(run.waitForExtinction).toBeTypeOf("function");
extinction.resolve();
await Promise.resolve();
expect(adapter.disposeMock).not.toHaveBeenCalled();
adapter.emitStdout("ok");
adapter.settle(0);
const exit = await run.wait();
expect(exit.reason).toBe("exit");
expect(exit.exitCode).toBe(0);
expect(exit.stdout).toBe("ok");
expect(exit.oomScoreWrapperSelected).toBe(true);
expect(adapter.disposeMock).toHaveBeenCalledTimes(1);
});
it.each([
{ outcome: "process-tree extinction", failure: false },
{ outcome: "cleanup identity loss", failure: true },
])("retains root-result cancellation ownership until $outcome", async ({ failure }) => {
const extinction = createDeferred();
const adapter = Object.assign(createStubChildAdapter(), {
waitForExtinction: () => extinction.promise,
});
createChildAdapterMock.mockResolvedValue(adapter);
const supervisor = createProcessSupervisor();
const run = await spawnChild(supervisor, {
sessionId: "root-result-before-extinction",
scopeKey: "scope:root-result-before-extinction",
argv: createSilentIdleArgv(),
});
expect(run.waitForExtinction).toBeTypeOf("function");
adapter.emitStdout("authentic root output");
adapter.settle(23);
const root = await run.wait();
expect(root).toMatchObject({ reason: "exit", exitCode: 23, stdout: "authentic root output" });
expect(adapter.disposeMock).not.toHaveBeenCalled();
supervisor.cancelScope("scope:root-result-before-extinction");
expect(adapter.killMock).toHaveBeenCalledWith("SIGKILL");
expect(supervisor.getRecord(run.runId)).toMatchObject({
state: "exited",
terminationReason: "exit",
exitCode: 23,
});
if (failure) {
extinction.reject(new Error("cleanup identity lost"));
await expect(run.waitForExtinction?.()).rejects.toThrow("cleanup identity lost");
} else {
extinction.resolve();
await expect(run.waitForExtinction?.()).resolves.toBeUndefined();
}
expect(adapter.disposeMock).toHaveBeenCalledOnce();
await expect(run.wait()).resolves.toBe(root);
supervisor.cancel(run.runId);
expect(adapter.killMock).toHaveBeenCalledOnce();
});
it("drains cancelled startups and live siblings before reporting ownership failure", async () => {
const first = createStubChildAdapter();
const sibling = createStubChildAdapter({ pid: 4321 });
const [firstExtinction, siblingExtinction] = [createDeferred(), createDeferred()];
first.waitForExtinction = async () => await firstExtinction.promise;
sibling.waitForExtinction = async () => await siblingExtinction.promise;
const startup = createDeferred<StubChildAdapter>();
createChildAdapterMock.mockReturnValueOnce(startup.promise).mockResolvedValueOnce(sibling);
const supervisor = createProcessSupervisor();
const pending = ["failed-owner", "pending-owner"].map((sessionId) =>
spawnChild(supervisor, {
sessionId,
scopeKey: "scope:failed-drain",
argv: createSilentIdleArgv(),
}),
);
supervisor.cancelScope("scope:failed-drain");
const drain = supervisor.waitForScope("scope:failed-drain");
startup.resolve(first);
const runs = await Promise.all(pending);
expect(first.killMock).toHaveBeenCalledWith("SIGTERM");
expect(sibling.killMock).toHaveBeenCalledWith("SIGTERM");
first.settle(0);
sibling.settle(0);
await Promise.all(runs.map((run) => run.wait()));
const drained = vi.fn();
void drain.then(drained, drained);
firstExtinction.reject(new Error("first owner lost authority"));
await Promise.resolve();
expect(drained).not.toHaveBeenCalled();
expect(sibling.disposeMock).not.toHaveBeenCalled();
siblingExtinction.resolve();
await expect(drain).rejects.toThrow("first owner lost authority");
expect(sibling.disposeMock).toHaveBeenCalledTimes(1);
});
});
@@ -0,0 +1,85 @@
import { vi } from "vitest";
import type { ProcessSupervisor, SpawnInput, SpawnProcessAdapter } from "./types.js";
type ChildSpawnOptions = Omit<Extract<SpawnInput, { mode: "child" }>, "backendId" | "mode">;
export type StubChildAdapter = SpawnProcessAdapter<NodeJS.Signals | null> & {
emitStdout: (chunk: string) => void;
emitStderr: (chunk: string) => void;
settle: (code: number | null, signal?: NodeJS.Signals | null) => void;
killMock: ReturnType<typeof vi.fn>;
disposeMock: ReturnType<typeof vi.fn>;
};
export function createWriteStdoutArgv(output: string): string[] {
if (process.platform === "win32") {
return [process.execPath, "-e", `process.stdout.write(${JSON.stringify(output)})`];
}
return ["/usr/bin/printf", "%s", output];
}
export function createSilentIdleArgv(): string[] {
return [process.execPath, "-e", "setInterval(() => {}, 1_000)"];
}
export function createStubChildAdapter(options?: {
pid?: number;
onKill?: (signal: NodeJS.Signals | undefined, adapter: StubChildAdapter) => void;
}): StubChildAdapter {
const stdoutListeners: Array<(chunk: string) => void> = [];
const stderrListeners: Array<(chunk: string) => void> = [];
let resolveWait:
| ((value: { code: number | null; signal: NodeJS.Signals | null }) => void)
| null = null;
const waitPromise = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>(
(resolve) => {
resolveWait = resolve;
},
);
const killMock = vi.fn();
const disposeMock = vi.fn();
const adapter: StubChildAdapter = {
pid: options?.pid ?? 1234,
stdin: undefined,
onStdout: (listener) => {
stdoutListeners.push(listener);
},
onStderr: (listener) => {
stderrListeners.push(listener);
},
wait: async () => await waitPromise,
kill: (signal) => {
killMock(signal);
options?.onKill?.(signal, adapter);
},
dispose: () => {
disposeMock();
},
emitStdout: (chunk) => {
for (const listener of stdoutListeners) {
listener(chunk);
}
},
emitStderr: (chunk) => {
for (const listener of stderrListeners) {
listener(chunk);
}
},
settle: (code, signal = null) => {
resolveWait?.({ code, signal });
resolveWait = null;
},
killMock,
disposeMock,
};
return adapter;
}
export async function spawnChild(supervisor: ProcessSupervisor, options: ChildSpawnOptions) {
return supervisor.spawn({
...options,
backendId: "test",
mode: "child",
});
}
+9 -152
View File
@@ -3,7 +3,14 @@ import { performance } from "node:perf_hooks";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../../test/helpers/promise.js";
import { mockProcessPlatform } from "../../test-utils/vitest-spies.js";
import type { ManagedRun, SpawnProcessAdapter } from "./types.js";
import {
createSilentIdleArgv,
createStubChildAdapter,
createWriteStdoutArgv,
spawnChild,
type StubChildAdapter,
} from "./supervisor.test-support.js";
import type { ManagedRun } from "./types.js";
const { createChildAdapterMock, createPtyAdapterMock } = vi.hoisted(() => ({
createChildAdapterMock: vi.fn(),
@@ -20,91 +27,6 @@ vi.mock("./adapters/pty.js", () => ({
let createProcessSupervisor: typeof import("./supervisor.js").createProcessSupervisor;
type ProcessSupervisor = ReturnType<typeof createProcessSupervisor>;
type SpawnOptions = Parameters<ProcessSupervisor["spawn"]>[0];
type ChildSpawnOptions = Omit<Extract<SpawnOptions, { mode: "child" }>, "backendId" | "mode">;
type ChildAdapter = SpawnProcessAdapter<NodeJS.Signals | null>;
type StubChildAdapter = ChildAdapter & {
emitStdout: (chunk: string) => void;
emitStderr: (chunk: string) => void;
settle: (code: number | null, signal?: NodeJS.Signals | null) => void;
killMock: ReturnType<typeof vi.fn>;
disposeMock: ReturnType<typeof vi.fn>;
};
function createWriteStdoutArgv(output: string): string[] {
if (process.platform === "win32") {
return [process.execPath, "-e", `process.stdout.write(${JSON.stringify(output)})`];
}
return ["/usr/bin/printf", "%s", output];
}
function createSilentIdleArgv(): string[] {
return [process.execPath, "-e", "setInterval(() => {}, 1_000)"];
}
function createStubChildAdapter(options?: {
pid?: number;
onKill?: (signal: NodeJS.Signals | undefined, adapter: StubChildAdapter) => void;
}): StubChildAdapter {
const stdoutListeners: Array<(chunk: string) => void> = [];
const stderrListeners: Array<(chunk: string) => void> = [];
let resolveWait:
| ((value: { code: number | null; signal: NodeJS.Signals | null }) => void)
| null = null;
const waitPromise = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>(
(resolve) => {
resolveWait = resolve;
},
);
const killMock = vi.fn();
const disposeMock = vi.fn();
const adapter: StubChildAdapter = {
pid: options?.pid ?? 1234,
stdin: undefined,
onStdout: (listener) => {
stdoutListeners.push(listener);
},
onStderr: (listener) => {
stderrListeners.push(listener);
},
wait: async () => await waitPromise,
kill: (signal) => {
killMock(signal);
options?.onKill?.(signal, adapter);
},
dispose: () => {
disposeMock();
},
emitStdout: (chunk) => {
for (const listener of stdoutListeners) {
listener(chunk);
}
},
emitStderr: (chunk) => {
for (const listener of stderrListeners) {
listener(chunk);
}
},
settle: (code, signal = null) => {
resolveWait?.({ code, signal });
resolveWait = null;
},
killMock,
disposeMock,
};
return adapter;
}
async function spawnChild(supervisor: ProcessSupervisor, options: ChildSpawnOptions) {
return supervisor.spawn({
...options,
backendId: "test",
mode: "child",
});
}
describe("process supervisor", () => {
beforeEach(async () => {
vi.resetModules();
@@ -119,72 +41,6 @@ describe("process supervisor", () => {
vi.restoreAllMocks();
});
it("spawns child runs and captures output", async () => {
const adapter = createStubChildAdapter();
adapter.oomScoreWrapperSelected = true;
createChildAdapterMock.mockResolvedValue(adapter);
const supervisor = createProcessSupervisor();
const run = await spawnChild(supervisor, {
sessionId: "s1",
argv: createWriteStdoutArgv("ok"),
timeoutMs: 1_000,
stdinMode: "pipe-closed",
});
adapter.emitStdout("ok");
adapter.settle(0);
const exit = await run.wait();
expect(exit.reason).toBe("exit");
expect(exit.exitCode).toBe(0);
expect(exit.stdout).toBe("ok");
expect(exit.oomScoreWrapperSelected).toBe(true);
expect(adapter.disposeMock).toHaveBeenCalledTimes(1);
});
it.each([
{ outcome: "process-tree extinction", failure: false },
{ outcome: "cleanup identity loss", failure: true },
])("retains root-result cancellation ownership until $outcome", async ({ failure }) => {
const extinction = createDeferred();
const adapter = Object.assign(createStubChildAdapter(), {
waitForExtinction: () => extinction.promise,
});
createChildAdapterMock.mockResolvedValue(adapter);
const supervisor = createProcessSupervisor();
const run = await spawnChild(supervisor, {
sessionId: "root-result-before-extinction",
scopeKey: "scope:root-result-before-extinction",
argv: createSilentIdleArgv(),
});
adapter.emitStdout("authentic root output");
adapter.settle(23);
const root = await run.wait();
expect(root).toMatchObject({ reason: "exit", exitCode: 23, stdout: "authentic root output" });
expect(adapter.disposeMock).not.toHaveBeenCalled();
supervisor.cancelScope("scope:root-result-before-extinction");
expect(adapter.killMock).toHaveBeenCalledWith("SIGTERM");
expect(supervisor.getRecord(run.runId)).toMatchObject({
state: "exited",
terminationReason: "exit",
exitCode: 23,
});
if (failure) {
extinction.reject(new Error("cleanup identity lost"));
await expect(run.waitForExtinction?.()).rejects.toThrow("cleanup identity lost");
} else {
extinction.resolve();
await expect(run.waitForExtinction?.()).resolves.toBeUndefined();
}
expect(adapter.disposeMock).toHaveBeenCalledOnce();
await expect(run.wait()).resolves.toBe(root);
supervisor.cancel(run.runId);
expect(adapter.killMock).toHaveBeenCalledOnce();
});
it("passes private secret input and exact environment to the child adapter", async () => {
const adapter = createStubChildAdapter();
createChildAdapterMock.mockResolvedValue(adapter);
@@ -471,6 +327,7 @@ describe("process supervisor", () => {
expect(createChildAdapterMock).toHaveBeenCalledTimes(2);
expect(first.killMock).toHaveBeenCalledWith("SIGTERM");
expect(replacementRun.pid).toBeUndefined();
expect(replacementRun.waitForExtinction).toBeUndefined();
expect(later.killMock).not.toHaveBeenCalled();
later.settle(0);
+94 -55
View File
@@ -22,6 +22,7 @@ import type {
type ActiveRun = {
run: ManagedRun;
scopeKey?: string;
waitForExtinction: () => Promise<void>;
};
type StartingRun = {
@@ -94,7 +95,9 @@ function resolveElapsedTimeoutReason(params: {
: null;
}
export function createProcessSupervisor(): ProcessSupervisor {
export function createProcessSupervisor(): ProcessSupervisor & {
waitForScope: (scopeKey: string) => Promise<void>;
} {
const registry = createRunRegistry();
const active = new Map<string, ActiveRun>();
const startingRuns = new Map<string, StartingRun>();
@@ -139,6 +142,28 @@ export function createProcessSupervisor(): ProcessSupervisor {
}
};
const waitForScope = async (scopeKey: string): Promise<void> => {
let firstFailure: PromiseRejectedResult | undefined;
while (true) {
const starts = Array.from(startingScopes.get(scopeKey)?.runs ?? []);
const owned = Array.from(active.values())
.filter((current) => current.scopeKey === scopeKey)
.map((current) => current.waitForExtinction());
if (starts.length === 0 && owned.length === 0) {
if (firstFailure) {
throw firstFailure.reason;
}
return;
}
// Startup can become active while the snapshot settles; recheck both maps
// so shutdown cannot outrun an admitted command or retained descendants.
const results = await Promise.allSettled([...owned, ...starts]);
firstFailure ??= results.find(
(result): result is PromiseRejectedResult => result.status === "rejected",
);
}
};
const startRun = async (
input: SpawnInput,
scopeKey: string | undefined,
@@ -194,8 +219,8 @@ export function createProcessSupervisor(): ProcessSupervisor {
}
let forcedReason: TerminationReason | null = startingRun.terminationReason ?? null;
let settled = false;
let extinguished = false;
let resultSettled = false;
let ownershipExtinct = false;
let stdout = "";
let stderr = "";
let stdoutListener = input.onStdout;
@@ -213,7 +238,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
let noOutputTimeoutDeadlineMs: number | null = null;
const setForcedReason = (reason: TerminationReason) => {
if (forcedReason) {
if (forcedReason || resultSettled) {
return;
}
forcedReason = reason;
@@ -223,9 +248,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
let cancelAdapter: ((reason: TerminationReason) => void) | null = null;
const requestCancel = (reason: TerminationReason) => {
if (!settled) {
setForcedReason(reason);
}
setForcedReason(reason);
cancelAdapter?.(reason);
};
startingRun.cancel = requestCancel;
@@ -239,7 +262,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
): NodeJS.Timeout => {
const intervalMs = resolveTimerTimeoutMs(remainingMs, 1);
return setTimeout(() => {
if (settled) {
if (resultSettled) {
return;
}
const nextRemainingMs = Math.min(remainingMs - intervalMs, deadlineMs - performance.now());
@@ -258,7 +281,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
const touchOutput = () => {
registry.touchOutput(runId);
if (!noOutputTimeoutMs || settled) {
if (!noOutputTimeoutMs || resultSettled) {
return;
}
noOutputTimeoutDeadlineMs = performance.now() + noOutputTimeoutMs;
@@ -291,23 +314,29 @@ export function createProcessSupervisor(): ProcessSupervisor {
env: input.env,
});
})()
: await createChildAdapter({
argv: input.argv,
cwd: input.cwd,
env: input.env,
exactEnv: input.exactEnv,
windowsVerbatimArguments: input.windowsVerbatimArguments,
input: input.input,
stdinMode: input.stdinMode,
secretInput: input.secretInput,
});
: input.mode === "anchored-shell"
? await createChildAdapter({
anchoredShellCommand: input.command,
cwd: input.cwd,
env: input.env,
})
: await createChildAdapter({
argv: input.argv,
cwd: input.cwd,
env: input.env,
exactEnv: input.exactEnv,
windowsVerbatimArguments: input.windowsVerbatimArguments,
input: input.input,
stdinMode: input.stdinMode,
secretInput: input.secretInput,
});
registry.updateState(runId, forcedReason ? "exiting" : "running", {
pid: adapter.pid,
...(forcedReason ? { terminationReason: forcedReason } : {}),
});
const clearTimers = (includeForceKill = true) => {
const clearResultTimers = () => {
if (timeoutTimer) {
clearTimeout(timeoutTimer);
timeoutTimer = null;
@@ -316,27 +345,50 @@ export function createProcessSupervisor(): ProcessSupervisor {
clearTimeout(noOutputTimer);
noOutputTimer = null;
}
if (includeForceKill && forceKillTimer) {
};
const releaseOwnership = () => {
if (ownershipExtinct) {
return;
}
ownershipExtinct = true;
if (forceKillTimer) {
clearTimeout(forceKillTimer);
forceKillTimer = null;
}
active.delete(runId);
// Control-channel extinction can precede independently drained output;
// keep decoder subscriptions alive until the root result also settles.
if (resultSettled) {
adapter.dispose();
}
};
const releaseAuthority = () => {
if (extinguished) {
return;
const settleResult = () => {
resultSettled = true;
clearResultTimers();
if (ownershipExtinct) {
adapter.dispose();
} else if (!adapter.waitForExtinction) {
releaseOwnership();
}
extinguished = true;
clearTimers();
adapter.dispose();
active.delete(runId);
};
cancelAdapter = (reason: TerminationReason) => {
if (extinguished || cancelRequested) {
if (ownershipExtinct || (cancelRequested && !(resultSettled && forceKillTimer))) {
return;
}
cancelRequested = true;
if (resultSettled) {
if (forceKillTimer) {
clearTimeout(forceKillTimer);
forceKillTimer = null;
}
// Root completion closes its terminal record, not ownership of
// descendants still retained by the authoritative group or Job.
adapter.kill("SIGKILL");
return;
}
// Windows has no catchable SIGTERM equivalent: the adapter implements it
// with asynchronous taskkill, so waiting the cleanup grace only delays an
// already-expired deadline before the same forced tree termination.
@@ -349,7 +401,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
}
adapter.kill("SIGTERM");
forceKillTimer = setTimeout(() => {
if (!extinguished) {
if (!ownershipExtinct) {
adapter.kill("SIGKILL");
}
}, GRACEFUL_CANCEL_TIMEOUT_MS);
@@ -395,8 +447,6 @@ export function createProcessSupervisor(): ProcessSupervisor {
touchOutput();
}, onRawOutput(rawInput?.onStderrRaw));
const adapterExtinction = adapter.waitForExtinction?.();
const waitPromise = (async (): Promise<RunExit> => {
const result = await adapter.wait();
const deadlineReason = resolveElapsedTimeoutReason({
@@ -405,8 +455,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
noOutputTimeoutDeadlineMs,
});
const terminalReason = forcedReason ?? deadlineReason;
settled = true;
clearTimers(false);
settleResult();
const reason: TerminationReason =
terminalReason ?? (result.signal != null ? ("signal" as const) : ("exit" as const));
@@ -426,19 +475,10 @@ export function createProcessSupervisor(): ProcessSupervisor {
exitCode: exit.exitCode,
exitSignal: exit.exitSignal,
});
if (!adapterExtinction) {
releaseAuthority();
}
return exit;
})().catch((err: unknown) => {
if (!settled) {
settled = true;
clearTimers(false);
if (adapterExtinction) {
adapter.kill("SIGKILL");
} else {
releaseAuthority();
}
if (!resultSettled) {
settleResult();
registry.finalize(runId, {
reason: "spawn-error",
exitCode: null,
@@ -448,15 +488,12 @@ export function createProcessSupervisor(): ProcessSupervisor {
throw err;
});
const extinctionPromise = adapterExtinction
? Promise.allSettled([waitPromise, adapterExtinction]).then(([, extinction]) => {
releaseAuthority();
if (extinction.status === "rejected") {
throw extinction.reason;
}
})
: undefined;
void extinctionPromise?.catch(() => undefined);
const extinctionPromise = adapter.waitForExtinction
? adapter.waitForExtinction().finally(releaseOwnership)
: waitPromise.then(() => undefined);
// Ownership failures remain observable through the explicit join, but a
// caller waiting only for the root result must not create an unhandled rejection.
void extinctionPromise.catch(() => undefined);
const managedRun: ManagedRun = {
runId,
@@ -464,7 +501,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
startedAtMs,
stdin: adapter.stdin,
wait: async () => await waitPromise,
...(extinctionPromise ? { waitForExtinction: async () => await extinctionPromise } : {}),
...(adapter.waitForExtinction && { waitForExtinction: () => extinctionPromise }),
cancel: (reason = "manual-cancel") => {
requestCancel(reason);
},
@@ -477,6 +514,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
active.set(runId, {
run: managedRun,
scopeKey,
waitForExtinction: async () => await extinctionPromise,
});
if (forcedReason) {
managedRun.cancel(forcedReason);
@@ -547,6 +585,7 @@ export function createProcessSupervisor(): ProcessSupervisor {
spawn,
cancel,
cancelScope,
waitForScope,
getRecord: (runId: string) => registry.get(runId),
};
}
+8 -2
View File
@@ -44,7 +44,7 @@ export type ManagedRun = {
startedAtMs: number;
stdin?: ManagedRunStdin;
wait: () => Promise<RunExit>;
/** Present only when tree cleanup can outlive the root process result. */
/** The root result may settle before its independently owned descendants exit. */
waitForExtinction?: () => Promise<void>;
cancel: (reason?: TerminationReason) => void;
/** Stop delivering output callbacks before owner teardown kills the child. */
@@ -119,11 +119,17 @@ type SpawnPtyInput = SpawnBaseInput & {
ptyCommand: string;
};
export type SpawnInput = SpawnChildInput | SpawnPtyInput;
type SpawnAnchoredShellInput = SpawnBaseInput & {
mode: "anchored-shell";
command: string;
};
export type SpawnInput = SpawnChildInput | SpawnPtyInput | SpawnAnchoredShellInput;
export interface ProcessSupervisor {
spawn(input: SpawnInput): Promise<ManagedRun>;
cancel(runId: string, reason?: TerminationReason): void;
cancelScope(scopeKey: string, reason?: TerminationReason): void;
waitForScope?: (scopeKey: string) => Promise<void>;
getRecord(runId: string): RunRecord | undefined;
}
@@ -266,6 +266,7 @@ describe("detectChangedScope Windows routing", () => {
for (const executablePath of [
"src/plugin-sdk/node-host.ts",
"src/plugin-sdk/node-host.test.ts",
"src/process/supervisor/supervisor.anchored-shell.real.test.ts",
"src/process/terminal-pty.test.ts",
"src/tui/tui.ts",
"src/tui/tui.resolve-codex-bin.test.ts",
+128 -131
View File
@@ -1,11 +1,20 @@
// Verifies local shell process handling for TUI local mode.
import { spawn } from "node:child_process";
import { EventEmitter } from "node:events";
import { join } from "node:path";
import type { OverlayHandle } from "@earendil-works/pi-tui";
import { describe, expect, it, vi } from "vitest";
import {
getProcessSupervisor,
type ManagedRun,
type ProcessSupervisor,
} from "../process/supervisor/index.js";
import type { RunExit, SpawnInput } from "../process/supervisor/types.js";
import { createLocalShellRunner } from "./tui-local-shell.js";
vi.mock("../process/supervisor/index.js", () => ({
getProcessSupervisor: vi.fn(),
}));
type ShellSupervisor = ReturnType<typeof getProcessSupervisor>;
const createSelector = () => {
const selector = {
onSelect: undefined as ((item: { value: string; label: string }) => void) | undefined,
@@ -28,7 +37,8 @@ function createOverlayHandle(): OverlayHandle {
}
function createShellHarness(params?: {
spawnCommand?: typeof import("node:child_process").spawn;
spawn?: ProcessSupervisor["spawn"];
supervisor?: ShellSupervisor;
getCwd?: () => string | undefined;
env?: Record<string, string>;
maxOutputChars?: number;
@@ -48,14 +58,22 @@ function createShellHarness(params?: {
lastSelector = createSelector();
return lastSelector;
});
const spawnCommand = params?.spawnCommand ?? vi.fn();
const { runLocalShellLine } = createLocalShellRunner({
const supervisor =
params?.supervisor ??
({
spawn: params?.spawn ?? vi.fn(),
cancel: vi.fn(),
cancelScope: vi.fn(),
waitForScope: vi.fn(async () => {}),
getRecord: vi.fn(),
} satisfies ShellSupervisor);
vi.mocked(getProcessSupervisor).mockReturnValue(supervisor);
const { runLocalShellLine, shutdown } = createLocalShellRunner({
chatLog,
tui,
openOverlay,
closeOverlay,
createSelector: createSelectorSpy,
spawnCommand,
...(params?.getCwd ? { getCwd: params.getCwd } : {}),
...(params?.env ? { env: params.env } : {}),
...(params?.maxOutputChars !== undefined ? { maxOutputChars: params.maxOutputChars } : {}),
@@ -66,20 +84,40 @@ function createShellHarness(params?: {
overlayHandle,
closeOverlay,
createSelectorSpy,
spawnCommand,
supervisor,
runLocalShellLine,
shutdown,
getLastSelector: () => lastSelector,
};
}
function requireSpawnOptions(spawnCommand: ReturnType<typeof vi.fn>): {
env?: Record<string, string>;
} {
const call = spawnCommand.mock.calls[0];
if (!call) {
throw new Error("expected spawn command call");
}
return call[1] as { env?: Record<string, string> };
function createSettlingSpawn(params: { stdout?: string[]; stderr?: string[]; error?: Error }) {
return vi.fn(async (input: SpawnInput) => {
const exit: RunExit = {
reason: "exit",
exitCode: 0,
exitSignal: null,
durationMs: 0,
stdout: "",
stderr: "",
timedOut: false,
noOutputTimedOut: false,
};
return {
runId: "local-shell-run",
startedAtMs: 0,
wait: async () => {
params.stdout?.forEach((chunk) => input.onStdout?.(chunk));
params.stderr?.forEach((chunk) => input.onStderr?.(chunk));
if (params.error) {
throw params.error;
}
return exit;
},
cancel: vi.fn(),
detachOutput: vi.fn(),
} satisfies ManagedRun;
});
}
describe("createLocalShellRunner", () => {
@@ -97,27 +135,15 @@ describe("createLocalShellRunner", () => {
expect(harness.messages).toContain("local shell: not enabled");
expect(harness.messages).toContain("local shell: not enabled for this session");
expect(harness.createSelectorSpy).toHaveBeenCalledTimes(1);
expect(harness.spawnCommand).not.toHaveBeenCalled();
expect(harness.supervisor.spawn).not.toHaveBeenCalled();
expect(harness.closeOverlay).toHaveBeenCalledWith(harness.overlayHandle);
});
it("sets OPENCLAW_SHELL when running local shell commands", async () => {
const spawnCommand = vi.fn((_command: string, _options: unknown) => {
const stdout = new EventEmitter();
const stderr = new EventEmitter();
return {
stdout,
stderr,
on: (event: string, callback: (...args: unknown[]) => void) => {
if (event === "close") {
setImmediate(() => callback(0, null));
}
},
};
});
const spawn = createSettlingSpawn({});
const harness = createShellHarness({
spawnCommand: spawnCommand as unknown as typeof import("node:child_process").spawn,
spawn,
env: { PATH: "/tmp/bin", USER: "dev" },
});
@@ -128,33 +154,22 @@ describe("createLocalShellRunner", () => {
await firstRun;
expect(harness.createSelectorSpy).toHaveBeenCalledTimes(1);
expect(spawnCommand).toHaveBeenCalledTimes(1);
const spawnOptions = requireSpawnOptions(spawnCommand);
expect(spawnOptions.env?.OPENCLAW_SHELL).toBe("tui-local");
expect(spawnOptions.env?.PATH).toBe("/tmp/bin");
expect(spawn).toHaveBeenCalledTimes(1);
const input = spawn.mock.calls[0]?.[0];
expect(input?.mode).toBe("anchored-shell");
expect(input?.env?.OPENCLAW_SHELL).toBe("tui-local");
expect(input?.env?.PATH).toBe("/tmp/bin");
expect(harness.messages).toContain("local shell: enabled for this session");
});
it("keeps stderr visible instead of evicting it when stdout fills the output cap", async () => {
const stdout = new EventEmitter();
const stderr = new EventEmitter();
const spawnCommand = vi.fn(() => ({
stdout,
stderr,
on: (event: string, callback: (...args: unknown[]) => void) => {
if (event === "close") {
setImmediate(() => {
// stdout fills the entire cap; stderr then carries the failure reason.
stdout.emit("data", Buffer.from("0".repeat(20)));
stderr.emit("data", Buffer.from("FATAL"));
callback(0, null);
});
}
},
}));
const spawn = createSettlingSpawn({
stdout: ["0".repeat(20)],
stderr: ["FATAL"],
});
const harness = createShellHarness({
spawnCommand: spawnCommand as unknown as typeof import("node:child_process").spawn,
spawn,
maxOutputChars: 20,
});
@@ -168,23 +183,9 @@ describe("createLocalShellRunner", () => {
});
it("keeps a whole code point when the combined output tail starts inside an emoji", async () => {
const stdout = new EventEmitter();
const stderr = new EventEmitter();
const spawnCommand = vi.fn(() => ({
stdout,
stderr,
on: (event: string, callback: (...args: unknown[]) => void) => {
if (event === "close") {
setImmediate(() => {
stdout.emit("data", Buffer.from("x😀"));
stderr.emit("data", Buffer.from("tail"));
callback(0, null);
});
}
},
}));
const spawn = createSettlingSpawn({ stdout: ["x😀"], stderr: ["tail"] });
const harness = createShellHarness({
spawnCommand: spawnCommand as unknown as typeof import("node:child_process").spawn,
spawn,
maxOutputChars: 6,
});
@@ -196,41 +197,6 @@ describe("createLocalShellRunner", () => {
expect(harness.messages.join("\n")).not.toMatch(/[\uD800-\uDFFF]/u);
});
it("preserves UTF-8 characters split across stdout and stderr chunks", async () => {
const stdout = new EventEmitter();
const stderr = new EventEmitter();
const spawnCommand = vi.fn(() => ({
stdout,
stderr,
on: (event: string, callback: (...args: unknown[]) => void) => {
if (event === "close") {
setImmediate(() => {
const stdoutBytes = Buffer.from("猫", "utf8");
const stderrBytes = Buffer.from("😀", "utf8");
stdout.emit("data", stdoutBytes.subarray(0, 1));
stderr.emit("data", stderrBytes.subarray(0, 2));
setImmediate(() => {
stdout.emit("data", stdoutBytes.subarray(1));
stderr.emit("data", stderrBytes.subarray(2));
callback(0, null);
});
});
}
},
}));
const harness = createShellHarness({
spawnCommand: spawnCommand as unknown as typeof import("node:child_process").spawn,
});
const run = harness.runLocalShellLine("!unicode");
harness.getLastSelector()?.onSelect?.({ value: "yes", label: "Yes" });
await run;
expect(harness.messages).toContain("[local] 猫");
expect(harness.messages).toContain("[local] 😀");
expect(harness.messages.join("\n")).not.toContain("");
});
it("refuses to retarget local commands after the working directory is deleted", async () => {
const harness = createShellHarness({ getCwd: () => undefined });
@@ -238,19 +204,16 @@ describe("createLocalShellRunner", () => {
harness.getLastSelector()?.onSelect?.({ value: "yes", label: "Yes" });
await run;
expect(harness.spawnCommand).not.toHaveBeenCalled();
expect(harness.supervisor.spawn).not.toHaveBeenCalled();
expect(harness.messages).toContain(
"local shell: working directory was deleted; cd to an existing directory first",
);
});
it("finishes a failed child before reporting the next local command", async () => {
const harness = createShellHarness({
spawnCommand: spawn,
getCwd: vi
.fn(() => process.cwd())
.mockReturnValueOnce(join(process.cwd(), ".missing-openclaw-local-shell-directory")),
});
it("finishes a failed run before reporting the next local command", async () => {
const spawn = createSettlingSpawn({ stdout: ["second\n"] });
spawn.mockRejectedValueOnce(new Error("synthetic spawn failure"));
const harness = createShellHarness({ spawn });
const failedRun = harness.runLocalShellLine("!echo first");
harness.getLastSelector()?.onSelect?.({ value: "yes", label: "Yes" });
@@ -266,29 +229,63 @@ describe("createLocalShellRunner", () => {
]);
});
it("does not crash when stdout or stderr emit an error event", async () => {
const stdout = new EventEmitter();
const stderr = new EventEmitter();
const spawnCommand = vi.fn(() => ({
stdout,
stderr,
on: (event: string, callback: (...args: unknown[]) => void) => {
if (event === "close") {
setImmediate(() => callback(0, null));
}
},
}));
const harness = createShellHarness({
spawnCommand: spawnCommand as unknown as typeof import("node:child_process").spawn,
});
it("reports a command result failure once", async () => {
const spawn = createSettlingSpawn({ error: new Error("synthetic failure") });
const harness = createShellHarness({ spawn });
const run = harness.runLocalShellLine("!cmd");
harness.getLastSelector()?.onSelect?.({ value: "yes", label: "Yes" });
await vi.waitFor(() => expect(spawnCommand).toHaveBeenCalledTimes(1));
stdout.emit("error", new Error("EPIPE"));
stderr.emit("error", new Error("EIO"));
await expect(run).resolves.toBeUndefined();
expect(harness.messages.some((message) => message.includes("exit 0"))).toBe(true);
expect(harness.messages.filter((message) => message.includes("synthetic failure"))).toEqual([
"[local] error: synthetic failure",
]);
});
it("fences a pending approval when shutdown begins", async () => {
const harness = createShellHarness();
const run = harness.runLocalShellLine("!echo late");
const selector = harness.getLastSelector();
await harness.shutdown();
selector?.onSelect?.({ value: "yes", label: "Yes" });
await run;
expect(harness.supervisor.spawn).not.toHaveBeenCalled();
expect(harness.supervisor.cancelScope).toHaveBeenCalledOnce();
expect(harness.supervisor.waitForScope).toHaveBeenCalledWith(
vi.mocked(harness.supervisor.cancelScope).mock.calls[0]?.[0],
);
expect(harness.closeOverlay).toHaveBeenCalledWith(harness.overlayHandle);
});
it("keeps another TUI instance's settled command scope alive during shutdown", async () => {
const spawn = createSettlingSpawn({});
const first = createShellHarness({ spawn });
const second = createShellHarness({ supervisor: first.supervisor });
for (const harness of [first, second]) {
const run = harness.runLocalShellLine("!echo alive");
harness.getLastSelector()?.onSelect?.({ value: "yes", label: "Yes" });
await run;
}
const firstScope = spawn.mock.calls[0]?.[0].scopeKey;
const secondScope = spawn.mock.calls[1]?.[0].scopeKey;
expect(firstScope).toBeDefined();
expect(secondScope).toBeDefined();
expect(firstScope).not.toBe(secondScope);
const liveScopes = new Set([firstScope, secondScope]);
vi.mocked(first.supervisor.cancelScope).mockImplementation((scopeKey) => {
liveScopes.delete(scopeKey);
});
const shutdown = first.shutdown();
expect(first.shutdown()).toBe(shutdown);
await shutdown;
expect(first.supervisor.cancelScope).toHaveBeenCalledOnce();
expect(first.supervisor.cancelScope).toHaveBeenCalledWith(firstScope);
expect(first.supervisor.waitForScope).toHaveBeenCalledWith(firstScope);
expect(liveScopes).toEqual(new Set([secondScope]));
});
});
+89 -78
View File
@@ -1,9 +1,9 @@
// Launches and manages the local shell process used by TUI local mode.
import { spawn } from "node:child_process";
import { StringDecoder } from "node:string_decoder";
import { randomUUID } from "node:crypto";
import type { Component, OverlayHandle, SelectItem } from "@earendil-works/pi-tui";
import { sliceUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { tryProcessCwd } from "../infra/safe-cwd.js";
import { getProcessSupervisor, type ManagedRun } from "../process/supervisor/index.js";
import { createSearchableSelectList } from "./components/selectors.js";
import { formatTuiErrorMessage } from "./tui-formatters.js";
@@ -23,7 +23,6 @@ type LocalShellDeps = {
onSelect?: (item: SelectItem) => void;
onCancel?: () => void;
};
spawnCommand?: typeof spawn;
getCwd?: () => string | undefined;
env?: NodeJS.ProcessEnv;
maxOutputChars?: number;
@@ -32,22 +31,28 @@ type LocalShellDeps = {
export function createLocalShellRunner(deps: LocalShellDeps) {
let localExecAsked = false;
let localExecAllowed = false;
let closing = false;
let shutdownPromise: Promise<void> | undefined;
let cancelPendingApproval: (() => void) | undefined;
const supervisor = getProcessSupervisor();
const waitForScope = supervisor.waitForScope;
if (!waitForScope) {
throw new Error("process supervisor must support scope extinction before running local shells");
}
const scopeKey = `tui-local:${randomUUID()}`;
const createSelector = deps.createSelector ?? createSearchableSelectList;
const spawnCommand = deps.spawnCommand ?? spawn;
const getCwd = deps.getCwd ?? tryProcessCwd;
const env = deps.env ?? process.env;
const maxChars = deps.maxOutputChars ?? 40_000;
const ensureLocalExecAllowed = async (): Promise<boolean> => {
if (localExecAllowed) {
return true;
}
if (localExecAsked) {
return false;
if (closing || localExecAsked) {
return localExecAllowed && !closing;
}
localExecAsked = true;
return await new Promise<boolean>((resolve) => {
let settled = false;
deps.chatLog.addSystem("Allow local shell commands for this session?");
deps.chatLog.addSystem(
"This runs commands on YOUR machine (not the gateway) and may delete files or reveal secrets.",
@@ -60,25 +65,30 @@ export function createLocalShellRunner(deps: LocalShellDeps) {
],
2,
);
selector.onSelect = (item: SelectItem) => {
deps.closeOverlay(overlayHandle);
if (item.value === "yes") {
localExecAllowed = true;
deps.chatLog.addSystem("local shell: enabled for this session");
resolve(true);
} else {
deps.chatLog.addSystem("local shell: not enabled");
resolve(false);
const finish = (allowed: boolean, message: string) => {
if (settled) {
return;
}
deps.tui.requestRender();
};
selector.onCancel = () => {
settled = true;
deps.closeOverlay(overlayHandle);
deps.chatLog.addSystem("local shell: cancelled");
cancelPendingApproval = undefined;
if (allowed) {
localExecAllowed = true;
}
deps.chatLog.addSystem(message);
deps.tui.requestRender();
resolve(false);
resolve(allowed);
};
const overlayHandle: OverlayHandle = deps.openOverlay(selector);
selector.onSelect = (item: SelectItem) => {
const allowed = item.value === "yes" && !closing;
finish(
allowed,
allowed ? "local shell: enabled for this session" : "local shell: not enabled",
);
};
selector.onCancel = () => finish(false, "local shell: cancelled");
const overlayHandle = deps.openOverlay(selector);
cancelPendingApproval = selector.onCancel;
deps.tui.requestRender();
});
};
@@ -98,7 +108,7 @@ export function createLocalShellRunner(deps: LocalShellDeps) {
}
const allowed = await ensureLocalExecAllowed();
if (!allowed) {
if (!allowed || closing) {
return;
}
@@ -115,63 +125,64 @@ export function createLocalShellRunner(deps: LocalShellDeps) {
deps.chatLog.addSystem(`[local] $ ${cmd}`);
deps.tui.requestRender();
const appendWithCap = (text: string, chunk: string) => {
const combined = text + chunk;
return combined.length > maxChars ? sliceUtf16Safe(combined, -maxChars) : combined;
};
await new Promise<void>((resolve) => {
const child = spawnCommand(cmd, {
// Intentionally a shell: this is an operator-only local TUI feature (prefixed with `!`)
// and is gated behind an explicit in-session approval prompt.
shell: true,
let stdout = "";
let stderr = "";
let error: unknown;
let result: Awaited<ReturnType<ManagedRun["wait"]>> | undefined;
let run: ManagedRun | undefined;
try {
run = await supervisor.spawn({
mode: "anchored-shell",
command: cmd,
sessionId: scopeKey,
backendId: "tui-local-shell",
scopeKey,
cwd,
env: { ...env, OPENCLAW_SHELL: "tui-local" },
captureOutput: false,
onStdout: (chunk) => {
stdout = sliceUtf16Safe(stdout + chunk, -maxChars);
},
onStderr: (chunk) => {
stderr = sliceUtf16Safe(stderr + chunk, -maxChars);
},
});
if (closing) {
return;
}
result = await run.wait();
} catch (caught) {
error = caught;
} finally {
run?.detachOutput?.();
}
// Keep the tail so a large stdout cannot evict a trailing stderr failure reason.
const combined = sliceUtf16Safe(
stdout + (stderr ? (stdout ? "\n" : "") + stderr : ""),
-maxChars,
).trimEnd();
let stdout = "";
let stderr = "";
let error: Error | undefined;
const stdoutDecoder = new StringDecoder("utf8");
const stderrDecoder = new StringDecoder("utf8");
// Pipe errors are incidental; close owns completion after any recorded spawn error.
const ignoreOutputStreamError = () => {};
child.stdout.on("error", ignoreOutputStreamError);
child.stderr.on("error", ignoreOutputStreamError);
child.stdout.on("data", (buf) => {
stdout = appendWithCap(stdout, stdoutDecoder.write(buf));
});
child.stderr.on("data", (buf) => {
stderr = appendWithCap(stderr, stderrDecoder.write(buf));
});
child.on("close", (code, signal) => {
stdout = appendWithCap(stdout, stdoutDecoder.end());
stderr = appendWithCap(stderr, stderrDecoder.end());
// Keep the tail (consistent with the streaming appendWithCap above) so a
// large stdout cannot evict stderr: the failure reason (FATAL etc.) at the
// end is what the operator needs most when output overflows the cap.
const combined = sliceUtf16Safe(
stdout + (stderr ? (stdout ? "\n" : "") + stderr : ""),
-maxChars,
).trimEnd();
if (combined) {
for (const lineLocal of combined.split("\n")) {
deps.chatLog.addSystem(`[local] ${lineLocal}`);
}
}
const status = error ? `error: ${formatTuiErrorMessage(error)}` : `exit ${code ?? "?"}`;
deps.chatLog.addSystem(`[local] ${status}${signal ? ` (signal ${signal})` : ""}`);
deps.tui.requestRender();
resolve();
});
child.on("error", (err) => {
error = err;
});
});
if (combined) {
for (const lineLocal of combined.split("\n")) {
deps.chatLog.addSystem(`[local] ${lineLocal}`);
}
}
const status = error
? `error: ${formatTuiErrorMessage(error)}`
: `exit ${result?.exitCode ?? "?"}`;
deps.chatLog.addSystem(
`[local] ${status}${result?.exitSignal ? ` (signal ${result.exitSignal})` : ""}`,
);
deps.tui.requestRender();
};
return { runLocalShellLine };
const shutdown = () =>
(shutdownPromise ??= (async () => {
closing = true;
cancelPendingApproval?.();
supervisor.cancelScope(scopeKey);
await waitForScope(scopeKey);
})());
return { runLocalShellLine, shutdown };
}
+31 -1
View File
@@ -11,6 +11,7 @@ import {
createOpenClawTestInstance,
type OpenClawTestInstance,
} from "../../test/helpers/openclaw-test-instance.js";
import { isProcessAlive, waitForPidFile } from "../../test/helpers/process-wait.js";
import { createDeferred } from "../../test/helpers/promise.js";
import { loadPersistedAuthProfileStore } from "../agents/auth-profiles/persisted.js";
import { resolveAgentModelPrimaryValue } from "../config/model-input.js";
@@ -18,6 +19,7 @@ import type { ModelProviderConfig } from "../config/types.models.js";
import type { OpenClawConfig } from "../config/types.openclaw.js";
import { connectGatewayClient } from "../gateway/test-helpers.e2e.js";
import { runExec } from "../process/exec.js";
import { killPidIfAlive } from "../test-utils/process-tree.js";
import { sleep } from "../utils/sleep.js";
import { GatewayChatClient } from "./gateway-chat.js";
import { extractTextFromMessage } from "./tui-formatters.js";
@@ -1352,10 +1354,30 @@ describe("TUI PTY real backends", () => {
);
it(
"confirms and renders local shell output and environment through a real local PTY",
"confirms and renders local shell output, then extinguishes descendants before TUI exit",
async ({ onTestFinished }) => {
const fixture = await startLocalModeTui(onTestFinished);
const rootPath = path.join(fixture.stateDir, "tui-local-owned-root.cjs");
const pidPath = path.join(fixture.stateDir, "tui-local-owned-descendant.pid");
let descendantPid: number | undefined;
try {
await writeFile(
rootPath,
`
const { spawn } = require("node:child_process");
const { writeFileSync } = require("node:fs");
const stdio = process.platform === "win32"
? ["ignore", "ignore", "ignore"]
: ["ignore", "ignore", "ignore", 3];
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
stdio,
detached: process.platform === "win32",
});
child.unref();
writeFileSync(${JSON.stringify(pidPath)}, String(child.pid));
`,
"utf8",
);
await fixture.run.waitForOutput("local ready", LOCAL_STARTUP_TIMEOUT_MS);
await fixture.run.write(
"!node -e \"console.log('T06_STDOUT'); console.error('T06_STDERR'); console.log('T06_ENV='+process.env.OPENCLAW_SHELL); process.exitCode=7\"\r",
@@ -1372,9 +1394,17 @@ describe("TUI PTY real backends", () => {
await fixture.run.waitForOutput("[local] T06_ENV=tui-local");
await fixture.run.waitForOutput("[local] exit 7");
const descendantCommandOffset = fixture.run.visibleOutput().length;
await fixture.run.write(`!node ${JSON.stringify(rootPath)}\r`);
await waitForOutputAfter(fixture.run, "[local] exit 0", descendantCommandOffset);
descendantPid = await waitForPidFile(pidPath, LOCAL_OUTPUT_TIMEOUT_MS);
expect(isProcessAlive(descendantPid)).toBe(true);
await fixture.run.write("/exit\r", { delay: false });
expect((await fixture.run.waitForExit()).exitCode).toBe(0);
expect(isProcessAlive(descendantPid)).toBe(false);
} finally {
killPidIfAlive(descendantPid);
await fixture.cleanup();
}
},
+12 -8
View File
@@ -984,13 +984,13 @@ describe("TUI shutdown safety", () => {
vi.useFakeTimers();
const calls: string[] = [];
const forceExit = vi.fn();
const recordPhase = (phase: string) => async () => {
calls.push(phase);
};
beginTestShutdown({
stopClient: async () => {
calls.push("client");
},
stopTui: async () => {
calls.push("tui");
},
stopCommandScopes: recordPhase("scopes"),
stopClient: recordPhase("client"),
stopTui: recordPhase("tui"),
disposeStatus: () => {
calls.push("status");
},
@@ -1001,7 +1001,7 @@ describe("TUI shutdown safety", () => {
});
await vi.advanceTimersByTimeAsync(0);
expect(calls).toEqual(["status", "client", "tui", "status", "finish"]);
expect(calls).toEqual(["status", "scopes", "client", "tui", "status", "finish"]);
expect(forceExit).not.toHaveBeenCalled();
});
@@ -1052,12 +1052,16 @@ describe("TUI shutdown safety", () => {
it("reports transport and terminal shutdown errors in phase order", async () => {
vi.useFakeTimers();
const scopeError = new Error("command scope stop failed");
const transportError = new Error("transport stop failed");
const terminalError = new Error("terminal stop failed");
const onError = vi.fn();
const requestFinish = vi.fn();
beginTestShutdown({
stopCommandScopes: async () => {
throw scopeError;
},
stopClient: async () => {
throw transportError;
},
@@ -1072,7 +1076,7 @@ describe("TUI shutdown safety", () => {
expect(onError).toHaveBeenCalledOnce();
const error = onError.mock.calls[0]?.[0];
expect(error).toBeInstanceOf(AggregateError);
expect((error as AggregateError).errors).toEqual([transportError, terminalError]);
expect((error as AggregateError).errors).toEqual([scopeError, transportError, terminalError]);
expect(requestFinish).toHaveBeenCalledOnce();
});
+16 -11
View File
@@ -504,6 +504,7 @@ type TuiProcessExitTimeout = (callback: () => void, delayMs: number) => TuiProce
type TuiShutdownTask = () => void | Promise<void>;
export function beginTuiShutdown(params: {
stopCommandScopes?: TuiShutdownTask;
stopClient: TuiShutdownTask;
stopTui: TuiShutdownTask;
disposeStatus: () => void;
@@ -525,10 +526,13 @@ export function beginTuiShutdown(params: {
void Promise.resolve()
.then(async () => {
const errors: unknown[] = [];
try {
await params.stopClient();
} catch (error) {
errors.push(error);
const runtimeTasks = [params.stopCommandScopes, params.stopClient].map(async (task) =>
task?.(),
);
for (const result of await Promise.allSettled(runtimeTasks)) {
if (result.status === "rejected") {
errors.push(result.reason);
}
}
// Terminal ownership must be released even when transport teardown fails.
try {
@@ -1496,6 +1500,12 @@ async function runTuiUnlocked(opts: RunTuiOptions): Promise<TuiResult> {
clearLocalBtwRunIds: localBtwRunIds.clear,
});
reconcileReconnectRun = reconnectStreamingWatchdog;
const localShell = createLocalShellRunner({
chatLog,
tui,
openOverlay,
closeOverlay,
});
invalidateSessionRunOwnership = () => {
disposeEventHandlers();
state.activeChatRunId = null;
@@ -1531,6 +1541,7 @@ async function runTuiUnlocked(opts: RunTuiOptions): Promise<TuiResult> {
pluginApprovals?.dispose();
taskSuggestions?.dispose();
beginTuiShutdown({
stopCommandScopes: () => localShell.shutdown(),
stopClient: () => client.stop(),
stopTui: () => drainAndStopTuiSafely(tui),
disposeStatus,
@@ -1592,12 +1603,6 @@ async function runTuiUnlocked(opts: RunTuiOptions): Promise<TuiResult> {
requestExit,
});
const { runLocalShellLine } = createLocalShellRunner({
chatLog,
tui,
openOverlay,
closeOverlay,
});
updateAutocompleteProvider();
const notifySubmitError = (action: TuiSubmitAction, error: unknown) => {
const message = formatTuiErrorMessage(error);
@@ -1608,7 +1613,7 @@ async function runTuiUnlocked(opts: RunTuiOptions): Promise<TuiResult> {
editor,
handleCommand,
sendMessage,
handleBangLine: runLocalShellLine,
handleBangLine: localShell.runLocalShellLine,
onSubmitError: notifySubmitError,
admitMessage: resolveMessageAdmission,
onBlockedMessageSubmit: reportBlockedMessageSubmit,
+2
View File
@@ -394,6 +394,8 @@ function buildCoreDistEntries(): Record<string, string> {
"process/supervisor/service-child-relay": "src/process/supervisor/service-child-relay.ts",
"process/supervisor/service-child-group-anchor":
"src/process/supervisor/service-child-group-anchor.ts",
"process/supervisor/service-child-windows-job-anchor":
"src/process/supervisor/service-child-windows-job-anchor.ts",
"telegram-ingress-worker.runtime": bundledPluginFile(
"telegram",
"src/telegram-ingress-worker.runtime.ts",