From d563099c857ae895ddf461699904d4ae56632447 Mon Sep 17 00:00:00 2001
From: Kit <344842+roodboi@users.noreply.github.com>
Date: Fri, 13 Nov 2026 10:12:56 +0000
Subject: [PATCH 01/13] feat(native): add interactive exec with retained
session cleanup
---
.github/workflows/ci.yml | 9 +
docs/diagrams/interactive-session.mmd | 35 +
docs/diagrams/interactive-session.svg | 66 ++
docs/guides/native-candidate.md | 49 +-
docs/reference/native-authored-commands.md | 53 +-
docs/reference/v5-model-coverage.md | 2 +
.../guest/environment-launcher.zig | 108 +++
packages/runtime-core/src/error.rs | 12 +
.../runtime-core/src/interactive_exec_cli.rs | 126 +++
packages/runtime-core/src/main.rs | 16 +
packages/runtime-core/src/provider/engine.rs | 14 +
.../src/provider/engine/interactive_exec.rs | 401 ++++++++++
.../src/provider/engine/relay_exec.rs | 34 +-
.../src/provider/environment_recovery.rs | 2 +-
.../provider/graph/foreground/transport.rs | 2 +-
.../src/provider/graph/interactive_exec.rs | 207 +++++
.../runtime-core/src/provider/graph/mod.rs | 3 +
.../src/provider/graph/native/foreground.rs | 155 +++-
.../src/provider/graph/native/runtime.rs | 4 +
.../src/provider/graph/native/runtime/exec.rs | 70 ++
.../src/provider/graph/service_exec.rs | 124 +++
.../src/provider/interactive_handoff.rs | 331 ++++++++
.../src/provider/interactive_session.rs | 719 ++++++++++++++++++
.../src/provider/interactive_terminal.rs | 593 +++++++++++++++
.../runtime-core/src/provider/lifecycle.rs | 2 +
.../src/provider/managed_environment.rs | 18 +
packages/runtime-core/src/provider/mod.rs | 6 +
.../src/provider/private_input.rs | 17 +-
scripts/lib/tla-runtime-models.ts | 60 ++
src/backends/native-authored-project-exec.ts | 47 ++
src/backends/native-interactive-exec.ts | 378 +++++++++
src/backends/native-project-exec.ts | 57 +-
src/lib/native-authored-command.ts | 6 +-
src/lib/tty-process-group.ts | 7 +-
tests/fixtures/environment-launcher-tty.py | 98 +++
tests/fixtures/native-interactive-frontend.py | 73 ++
tests/fixtures/native-interactive-frontend.ts | 19 +
tests/fixtures/native-interactive-runtime.py | 44 ++
tests/fixtures/native-interactive-terminal.py | 94 +++
.../tla/interactive-envelope/COVERAGE.md | 21 +
.../tla/interactive-envelope/Envelope.tla | 41 +
.../tla/interactive-envelope/negative.cfg | 4 +
.../tla/interactive-envelope/positive.cfg | 4 +
.../tla/interactive-session/COVERAGE.md | 28 +
.../tla/interactive-session/Session.tla | 57 ++
.../interactive-session/eof-retirement.cfg | 6 +
.../tla/interactive-session/negative.cfg | 6 +
.../tla/interactive-session/positive.cfg | 6 +
.../tla/interactive-session/wrong-boot.cfg | 6 +
.../native-authored-command-dispatch.test.ts | 2 +-
tests/native-authored-command.test.ts | 2 +-
tests/native-interactive-exec.test.ts | 117 +++
tests/native-interactive-launcher.test.ts | 68 ++
53 files changed, 4391 insertions(+), 38 deletions(-)
create mode 100644 docs/diagrams/interactive-session.mmd
create mode 100644 docs/diagrams/interactive-session.svg
create mode 100644 packages/runtime-core/src/interactive_exec_cli.rs
create mode 100644 packages/runtime-core/src/provider/engine/interactive_exec.rs
create mode 100644 packages/runtime-core/src/provider/graph/interactive_exec.rs
create mode 100644 packages/runtime-core/src/provider/interactive_handoff.rs
create mode 100644 packages/runtime-core/src/provider/interactive_session.rs
create mode 100644 packages/runtime-core/src/provider/interactive_terminal.rs
create mode 100644 src/backends/native-interactive-exec.ts
create mode 100644 tests/fixtures/environment-launcher-tty.py
create mode 100644 tests/fixtures/native-interactive-frontend.py
create mode 100644 tests/fixtures/native-interactive-frontend.ts
create mode 100644 tests/fixtures/native-interactive-runtime.py
create mode 100644 tests/fixtures/native-interactive-terminal.py
create mode 100644 tests/models/tla/interactive-envelope/COVERAGE.md
create mode 100644 tests/models/tla/interactive-envelope/Envelope.tla
create mode 100644 tests/models/tla/interactive-envelope/negative.cfg
create mode 100644 tests/models/tla/interactive-envelope/positive.cfg
create mode 100644 tests/models/tla/interactive-session/COVERAGE.md
create mode 100644 tests/models/tla/interactive-session/Session.tla
create mode 100644 tests/models/tla/interactive-session/eof-retirement.cfg
create mode 100644 tests/models/tla/interactive-session/negative.cfg
create mode 100644 tests/models/tla/interactive-session/positive.cfg
create mode 100644 tests/models/tla/interactive-session/wrong-boot.cfg
create mode 100644 tests/native-interactive-exec.test.ts
create mode 100644 tests/native-interactive-launcher.test.ts
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7bcd3d5d9..8ea6d903e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -311,6 +311,15 @@ jobs:
(cd "$destination" && echo "$checksum $archive" | shasum -a 256 --check)
tar -xf "$destination/$archive" -C "$destination"
echo "$destination/zig-${platform}-0.15.2" >> "$GITHUB_PATH"
+ - name: Test interactive guest launcher on a real Linux PTY
+ if: runner.os == 'Linux'
+ shell: bash
+ run: |
+ set -euo pipefail
+ test "$(zig version)" = '0.15.2'
+ command -v python3
+ zig build-exe packages/runtime-core/guest/environment-launcher.zig -target x86_64-linux-musl -O ReleaseSmall -femit-bin="$RUNNER_TEMP/interactive-environment-launcher"
+ python3 tests/fixtures/environment-launcher-tty.py "$RUNNER_TEMP/interactive-environment-launcher"
- name: Check candidate formatting
run: cargo fmt --manifest-path packages/runtime-core/Cargo.toml --check
- name: Lint candidate core
diff --git a/docs/diagrams/interactive-session.mmd b/docs/diagrams/interactive-session.mmd
new file mode 100644
index 000000000..781ea40a3
--- /dev/null
+++ b/docs/diagrams/interactive-session.mmd
@@ -0,0 +1,35 @@
+---
+title: Session · admission and cleanup proof
+---
+stateDiagram-v2
+ direction TB
+ [*] --> new
+ new --> admitted : Admit
+ admitted --> withdrawn : Withdraw [undispatched]
+ admitted --> created : Create
+ created --> recorded : Record
+ recorded --> started : Start [fresh]
+ started --> attached : Attach
+ started --> ended : EOF
+ attached --> ended : EOF
+ admitted --> abandoned : Crash
+ created --> abandoned : Crash
+ recorded --> abandoned : Crash
+ started --> abandoned : Crash
+ attached --> abandoned : Crash
+ ended --> abandoned : Crash
+ ended --> retired : Retire [exact proof]
+ abandoned --> retired : Retire [exact proof]
+ note right of recorded
+ INV WriteAhead
+ Intent and exec ID durable before Start
+ end note
+ note right of abandoned
+ INV NoReplay
+ No recovered admission capability
+ end note
+ note right of retired
+ INV RetiredWithProof
+ Same-incarnation container absence
+ or proven bound VM boot stop
+ end note
diff --git a/docs/diagrams/interactive-session.svg b/docs/diagrams/interactive-session.svg
new file mode 100644
index 000000000..8b48514c2
--- /dev/null
+++ b/docs/diagrams/interactive-session.svg
@@ -0,0 +1,66 @@
+
\ No newline at end of file
diff --git a/docs/guides/native-candidate.md b/docs/guides/native-candidate.md
index b6565a195..63e6523f3 100644
--- a/docs/guides/native-candidate.md
+++ b/docs/guides/native-candidate.md
@@ -3980,8 +3980,9 @@ are allowed.
With the same explicit native binary/home selection used for `up`, run
`hack exec --path /absolute/project SERVICE -- /bin/echo hello`. The command targets
the current owned project/branch mapping, preserves binary stdout/stderr and the
-command exit code, and never falls back to Compose. It is noninteractive, with a
-30-second observation budget; a timeout may leave the command running and is not
+command exit code, and never falls back to Compose. With piped input/output or
+`HACK_NO_INTERACTIVE=1`, it uses finite exec with a 30-second observation budget;
+a timeout may leave the command running and is not
retried. `--env` and `--profile` changes are refused. A command name such as `bun`
uses the selected container's PATH; no host PATH or implicit shell is used. Normal
`hack exec` resolves the saved startup overlay and AWS profile again, and delivers
@@ -3997,6 +3998,50 @@ refuse before execution; restart with the matching candidate to adopt this path.
Services without managed values use ordinary exec. The native live qualification
of this fresh-delivery path remains separate from its unit and transport tests.
+With terminal stdin and stdout, `hack exec SERVICE -- /bin/sh` selects interactive
+exec. It streams both guest output channels through the terminal, forwards resize
+and terminal control bytes, and releases the provider mutation lease after attach.
+`down` and `restart` can proceed while the session is open. Authored projects use
+their foreground owner for admission and inherit startup environment values.
+
+For adopted services with managed values, the pinned guest launcher disables echo
+and enters raw mode before announcing readiness. Hack then sends a bounded base64
+private envelope, waits for the launcher's acknowledgement after terminal
+restoration, and only then forwards user input. The envelope cannot contain Docker
+detach-control bytes. Old launcher mounts refuse before any value is sent; restart
+the service with the matching runtime to enable this mode. The launcher returns
+125 if admission or exec fails, including an exec failure after acknowledgement.
+This does not isolate values from other processes running as the same container
+user, which already share access to the command's environment.
+
+Interactive managed-env exec refuses if fresh delivery is unavailable; it does
+not reuse mounted startup values. Finite exec retains its existing behavior.
+
+An interactive command's exit code does not prove its descendants are gone. Hack
+retains a value-free session receipt and prints `session_cleanup_uncertain` with
+its UUID and exact `hack down` remedy until the same container incarnation is
+absent or its VM boot is proven stopped. EOF, detach, a dead exec process and a
+changed boot alone are not cleanup proof. No uncertain command is replayed.
+
+### Interactive exec acceptance
+
+On the built current-branch runtime and CLI, exercise authored and adopted services:
+
+- A shell with foreground/background jobs; Ctrl-C, Ctrl-Z, `fg` and resize.
+- Ctrl-P Ctrl-Q disconnect, retaining its UUID and cleanup remedy.
+- A background process that survives shell exit; receipt remains until `down`.
+- `down` and restart from another terminal while the session is active; the owner
+ stays responsive and the old session never attaches to the new generation.
+- Native child death and frontend death, including an idle terminal; attributes
+ restore, owned children settle, and uncertain sessions remain discoverable.
+- VM loss, followed by exact stopped-boot proof; a new boot alone cannot retire
+ an older receipt.
+- Fresh adopted env with a synthetic canary and detach-control bytes; no private
+ bytes echo, the intended command receives the values, and an old launcher refuses
+ before delivery.
+
+Linux PTY and fake-Engine tests do not qualify these real Engine/VM effects.
+
Managed HTTPS authority startup uses the same five-second provider-lock admission
budget as ordinary pool startup. This lets a brief concurrent graph inspection
diff --git a/docs/reference/native-authored-commands.md b/docs/reference/native-authored-commands.md
index 6115bd2f3..211c17374 100644
--- a/docs/reference/native-authored-commands.md
+++ b/docs/reference/native-authored-commands.md
@@ -4,7 +4,7 @@ An explicit `HACK_RUNTIME_BACKEND=native` selection delegates an authored
`.hack/hack.project.json` project to the separate native foreground owner. The
current command slice supports whole-project foreground or detached `up` and foreground `restart`,
owner-mediated `down` and explicit stored-generation `down --recover`, saved-run
-`ps`, live service `logs --follow`, finite `logs --no-follow`, noninteractive service `exec` and one-off
+`ps`, live service `logs --follow`, finite `logs --no-follow`, terminal or finite service `exec` and one-off
service `run` on macOS with an absolute `HACK_NATIVE_BINARY` and private canonical
`HACK_NATIVE_HOME`. It requires
the caller's separately prepared native provider pool and matching runtime binary.
@@ -75,7 +75,7 @@ provider contact. A refusal names no option value, path or project detail.
| `down --recover` | Explicit retirement of a complete stored generation whose owners are dead, or of a pending startup intent that never reserved runtime state; omit `--profile` | `--json`, `--detach`, explicit empty or named `--profile`, `--env` overrides, service subsets, single-service, command, workdir, follow, tail and format options |
| `ps` | `ps` and `ps --json` for the exact saved project and explicit `--branch` | `--profile`, `--env`, service subsets or a single service, `--recover`, `--detach`, command, workdir, follow, tail and format options |
| `logs` | `logs SERVICE --follow` (the default), or `--no-follow`; `--tail` from 0 through 1000 with follow, 1 through 1000 with no-follow (default 200), plain or `--json`; `--since` and `--timestamps` with follow | Omitted, multiple or malformed service selection, non-integer or out-of-range `--tail`, pretty format, `--until`, query filters, `--profile`, `--env`, `--recover`, `--detach`, command and workdir options; TTY-configured services in follow mode |
-| `exec` | `exec SERVICE -- COMMAND ARG...` with closed stdin, no TTY and an optional absolute `--workdir` | Empty command or empty first argument, NUL bytes, more than 256 arguments, an argument over 16 KiB or 64 KiB in total, relative `--workdir`, interactive or TTY input, `--json`, `--profile`, `--env`, service subsets, follow, tail and format options, `--recover`, `--detach` |
+| `exec` | `exec SERVICE -- COMMAND ARG...`; terminal input/output selects a TTY session, otherwise finite exec with closed stdin; optional absolute `--workdir` | Empty command or empty first argument, NUL bytes, more than 256 arguments, an argument over 16 KiB or 64 KiB in total, relative `--workdir`, `--json`, `--profile`, `--env`, service subsets, follow, tail and format options, `--recover`, `--detach` |
| `run` | `run SERVICE -- COMMAND ARG...` through a ready foreground owner, closed stdin, no TTY, optional absolute `--workdir`; member must have no mounts | The `exec` option refusals above; source, persistent-data or private-env mounts refuse before the one-off effect |
| `restart` | Whole-project proven stop followed by fresh foreground `up`; `--branch`, `--profile` and `--env` select the fresh generation | Service subsets, `--json`, `--detach`, `--target`, recovery and other unsupported foreground options; missing or unproven stop prevents startup |
@@ -498,8 +498,8 @@ not qualify a live provider or application log behavior.
`exec SERVICE -- COMMAND ARG...` sends one command to an already running authored
service through its authenticated foreground owner. Arguments are literal, including
empty arguments, spaces and shell characters. `--workdir` selects an absolute guest
-directory; the service's existing environment and user remain in force. Stdin is
-closed and no TTY is allocated. Interactive input, TTY control, env/profile overrides,
+directory; the service's existing environment and user remain in force. Without
+terminal input and output, stdin is closed and no TTY is allocated. Env/profile overrides,
multiple services and JSON formatting are unsupported. No shell is inserted.
The owner verifies the admitted service, full resource ownership, guest boot, image,
@@ -520,6 +520,51 @@ project is not stopped. Inspect the service before
deciding whether another command is appropriate. Offline controls do not qualify
installed native exec or live application behavior.
+## Interactive service exec
+
+When both stdin and stdout are terminals, `hack exec SERVICE -- /bin/sh` opens a
+TTY session. `HACK_NO_INTERACTIVE=1` selects the finite path instead. The shell is
+explicit: Hack never inserts one. Authored commands inherit the service's startup
+environment. The authenticated owner checks the same service and generation,
+creates and starts the exec, then hands its connected Engine stream to the native
+terminal child. The owner remains available for `down` and `restart`; no mutation
+lease is held during the session.
+
+The native home holds at most 1024 retained sessions across all projects.
+`session_receipt_limit` refuses further admission without discarding evidence;
+run `hack down` for a project with retained sessions to release its receipts.
+This includes normally exited commands whose background descendants remain
+unproven. The command's known exit status is still preserved.
+
+Terminal output combines the command's stdout and stderr. Ctrl-C, Ctrl-Z and other
+terminal bytes go to the guest PTY; window-size changes are forwarded. Docker's
+Ctrl-P Ctrl-Q sequence disconnects the transport. Disconnecting does not prove
+that the command or its descendants stopped. The native child restores terminal
+attributes on exit or frontend loss; the frontend has a second snapshot for native
+child failure. If restoration fails, Hack names `stty sane` as the repair.
+
+Before create, Hack durably records a value-free session UUID bound to the exact
+owner, run, boot, container and start generation; it records the exec ID before
+start. A known refusal before create dispatch withdraws only its own intent;
+uncertain dispatch retains it. Neither operation is replayed. An exec exit code is returned when
+known, but EOF or process exit cannot retire the session receipt because background
+descendants may remain. Retained evidence produces `session_cleanup_uncertain`,
+the UUID and an exact project/branch `hack down` command. Only same-incarnation
+container absence or a proven stop of the bound VM boot retires it. A new boot
+alone is not proof. Receipt bookkeeping failures do not block container cleanup.
+
+
+
+The [diagram source](../diagrams/interactive-session.mmd) and
+[model coverage](../../tests/models/tla/interactive-session/COVERAGE.md) name the
+guards and finite proof limits. Retirement is also allowed from any earlier
+intent-bearing state once the same exact cleanup proof holds.
+
+The PTY fixtures qualify local input/output, cancellation, private delivery and
+restoration boundaries. Real Engine job control, detach keys, macOS descriptor
+handoff, down/restart during a session and VM loss require the real-VM acceptance
+run described in the [candidate guide](../guides/native-candidate.md#interactive-exec-acceptance).
+
## Finite one-off service run
`run SERVICE -- COMMAND ARG...` sends one command to the authenticated foreground
diff --git a/docs/reference/v5-model-coverage.md b/docs/reference/v5-model-coverage.md
index 30f5db75a..e6866cee9 100644
--- a/docs/reference/v5-model-coverage.md
+++ b/docs/reference/v5-model-coverage.md
@@ -36,6 +36,8 @@ must be attached to the eventual exact integration revision.
| Property and critical states | Model coverage | Implementation/regression boundary | Remaining qualification |
| --- | --- | --- | --- |
+| Interactive exec write-ahead admission, crash, EOF, replacement boot and retirement | interactive-session (180 states); no-replay, EOF-retirement and wrong-boot negative controls | `interactive_session.rs` atomic intent/exec records, exact stopped-boot proof, known-undispatched withdrawal and lock-free observation/revalidation tests; Engine effect fences; authored owner start and SCM_RIGHTS handoff | One session, two boot identities, two possible starts. Atomic durable publication and exact identity/absence proofs are assumptions; real fsync/power loss, real owner/Engine handoff and VM loss need Mac acceptance. No descendant cleanup inferred from exec exit. |
+| Private interactive envelope, echo state and bytes still in flight on failure | interactive-envelope (15 states); restore-on-error negative control | HKEE2 launcher and host gate, Linux PTY seven-case matrix, bounded marker/ACK and coalesced-output tests | Two abstract private bytes, one handshake. PTY fixtures prove local termios ordering; Docker detach scanning, real guest wiring, same-user access and VM effects remain separate. Source/compiled Bun PTY and native-process fixtures cover lifetime/terminal boundaries, not model liveness. |
| Native log follower versus lifecycle mutation, frontend loss, stopped container and transport loss | No new state model: the follower observes without a mutation lease and grants no cleanup authority | `provider/graph/log_follow.rs` and its native counterpart pin Ready admission, container and start generation; `provider/engine/log_follow.rs` bounds framing and distinguishes real EOF from stopped-service idle cutoff. `lifecycle_log_follow_tests.rs` keeps the production transport on a fake Engine socket active while acquiring startup and operation leases, with a same-lock contention control and output while each lease is held. `native-log-follow-source.test.ts` guards the observer-only route; process tests cover cancellation, frontend loss, real output-pipe closure while idle, replacement-container restart (including retained adopted run/owner IDs) with one marker, NDJSON admission failures, truncated/partial output and slow idle-end delivery. `native-log-restart.test.ts` covers bounded re-admission, unchanged-container refusal and prompt down; `native-log-selection.test.ts` covers strict finite admission responses and owned-child cancellation/timeouts; advisory markers grant no authority | 4 KiB payload envelopes, 8 KiB wire lines, 64 Ki-character retained payload prefixes excluding Engine timestamps, 30-second restart wait, 1000-line new-container backlog. The portable lease test fakes identity/state checks; it does not exercise host process/disk auditing, graph admission or the VM relay. Real relay EOF on stop/down/restart, VM loss and concurrent lifecycle latency require Apple Silicon acceptance. No performance or whole-product parity claim. |
| Foreground group interrupt; publisher exit before authenticated stop; dead owner's pending cleanup enrollment and bridge release | native-publisher-stop | [Process adapter](../../src/backends/native-project-process.ts) isolates Compose owners regardless of hooks; Rust `process::publisher_command` isolates publishers; `legacy_cleanup_enrollment` binds the old coordinator and inventories before a distinct recovery intent. [Group regression](../../tests/native-foreground-group-interrupt.test.ts) signals the whole group for fresh/restore owners; Rust real-publisher process, frontend-loss, exact child-reaping and enrollment tests check grouping, lifetime wakeups and stale selectors. | One graph, one publisher, one crash, one identity replacement and one audited successor boot; 168 states. Publisher exit observation is separate from lease-bound reaping/proof commitment; `NoUncommittedExit` rejects owner cleanup while proof is pending. A busy lease stutters without delaying stop control. SIGHUP requests retaining cleanup; a reaped publisher enters an observable publisher-down state until explicit stop. Native process tests and frontend framing/ps tests cover signal delivery and notices. Dependency release is a separate post-completion step; `NoStrandedReservation` rejects success with a held reservation. Real reservation-file regressions check retry, capacity reuse and refusal fences. Separate stopped-receipt and intent-completion commits are collapsed; TypeScript dispatch and Rust completion-digest regressions cover that interruption window. Publication-registry sync/rename is abstracted and covered by portable commit-interruption tests. Kernel identity, socket/inode proofs, guest observations, durable storage and existing next.13 home recovery remain implementation/Apple Silicon VM gates. No eventual recovery is asserted. |
| Exclusive claims; competing writers; partial owner publication | graph-admission, registry-writer, publication-journal, missing-publication-lock, native-frontend-recovery | Admission/registry locks, pending publication and exact saved owner; related runtime ownership tests | Real filesystems, boot/incarnation identity and unsupported writers remain independent. |
diff --git a/packages/runtime-core/guest/environment-launcher.zig b/packages/runtime-core/guest/environment-launcher.zig
index 64f0008db..cfa063501 100644
--- a/packages/runtime-core/guest/environment-launcher.zig
+++ b/packages/runtime-core/guest/environment-launcher.zig
@@ -108,10 +108,118 @@ fn freshExec(args: []const [:0]u8) !void {
// stdin was fully consumed to EOF; the command receives no secret input bytes.
return execCommand(args[2..], &env);
}
+
+/// The private envelope uses only printable base64, so Docker's detach-key
+/// scanner cannot interpret its contents. Nothing is sent until echo is off;
+/// user input is withheld until the restored-terminal acknowledgement.
+fn interactiveFreshExec(args: []const [:0]u8) !void {
+ if (args.len < 3 or args.len > 4099) return error.Arguments;
+ const started = try bootSeconds();
+ const saved = try std.posix.tcgetattr(0);
+ var raw = saved;
+ raw.iflag.BRKINT = false;
+ raw.iflag.ICRNL = false;
+ raw.iflag.INLCR = false;
+ raw.iflag.IGNCR = false;
+ raw.iflag.ISTRIP = false;
+ raw.iflag.IXON = false;
+ raw.iflag.PARMRK = false;
+ raw.oflag.OPOST = false;
+ raw.lflag.ECHO = false;
+ raw.lflag.ECHONL = false;
+ raw.lflag.ICANON = false;
+ raw.lflag.ISIG = false;
+ raw.lflag.IEXTEN = false;
+ raw.cflag.CSIZE = .CS8;
+ raw.cflag.PARENB = false;
+ raw.cc[@intFromEnum(std.posix.V.MIN)] = 1;
+ raw.cc[@intFromEnum(std.posix.V.TIME)] = 0;
+ try std.posix.tcsetattr(0, .NOW, raw);
+ // On failure leave echo disabled, including for bytes still in flight.
+ // The guest launcher targets Linux; Zig has no portable tcflush wrapper.
+ errdefer _ = std.os.linux.ioctl(0, std.os.linux.T.CFLSH, 0); // TCIFLUSH
+ var hello: [64]u8 = undefined;
+ const marker = try std.fmt.bufPrint(&hello, "HKEE2 {d}\n", .{started});
+ if (try std.posix.write(1, marker) != marker.len) return error.Transport;
+ var env = try interactiveEnvelope(started);
+ defer env.deinit();
+ try std.posix.tcsetattr(0, .DRAIN, saved);
+ const ack = "HKEE2 OK\n";
+ if (try std.posix.write(1, ack) != ack.len) return error.Transport;
+ return execCommand(args[2..], &env);
+}
+
+fn privateRead(destination: []u8, started: u64) !void {
+ var used: usize = 0;
+ while (used < destination.len) {
+ const now = try bootSeconds();
+ if (now >= started + 5) return error.Expired;
+ var fds = [_]std.posix.pollfd{.{ .fd = 0, .events = std.posix.POLL.IN, .revents = 0 }};
+ if (try std.posix.poll(&fds, @intCast((started + 5 - now) * 1000)) == 0) return error.Expired;
+ const n = try std.posix.read(0, destination[used..]);
+ if (n == 0) return error.Transport;
+ used += n;
+ }
+}
+
+fn interactiveEnvelope(started: u64) !std.process.EnvMap {
+ var header: [32]u8 = undefined;
+ defer std.crypto.secureZero(u8, &header);
+ var used: usize = 0;
+ while (true) {
+ if (used == header.len) return error.Payload;
+ try privateRead(header[used..][0..1], started);
+ used += 1;
+ if (header[used - 1] == '\n') break;
+ }
+ if (!std.mem.startsWith(u8, header[0..used], "HKEE2 ")) return error.Payload;
+ const digits = header[6 .. used - 1];
+ if (digits.len == 0 or digits[0] == '0') return error.Payload;
+ for (digits) |c| if (!std.ascii.isDigit(c)) return error.Payload;
+ const length = try std.fmt.parseInt(usize, digits, 10);
+ const encoded = try allocator.alloc(u8, 87384);
+ defer {
+ std.crypto.secureZero(u8, encoded);
+ allocator.free(encoded);
+ }
+ if (length == 0 or length > encoded.len) return error.Payload;
+ try privateRead(encoded[0..length], started);
+ const decoder = std.base64.standard.Decoder;
+ const decoded_length = try decoder.calcSizeForSlice(encoded[0..length]);
+ if (decoded_length == 0 or decoded_length > 65536) return error.Payload;
+ const decoded = try allocator.alloc(u8, 65536);
+ defer {
+ std.crypto.secureZero(u8, decoded);
+ allocator.free(decoded);
+ }
+ try decoder.decode(decoded[0..decoded_length], encoded[0..length]);
+ // JSON may allocate decoded escape sequences. Keep all parse storage in a
+ // bounded arena which is wiped before the acknowledgement, too.
+ const storage = try allocator.alloc(u8, 512 * 1024);
+ defer {
+ std.crypto.secureZero(u8, storage);
+ allocator.free(storage);
+ }
+ var fixed = std.heap.FixedBufferAllocator.init(storage);
+ const parsed = try std.json.parseFromSlice(std.json.Value, fixed.allocator(), decoded[0..decoded_length], .{ .duplicate_field_behavior = .@"error" });
+ defer parsed.deinit();
+ const values = try freshValues(parsed.value, started, try bootSeconds());
+ var env = try std.process.getEnvMap(allocator);
+ errdefer env.deinit();
+ var entries = values.iterator();
+ while (entries.next()) |entry| try env.put(entry.key_ptr.*, entry.value_ptr.string);
+ // The host is still gating keystrokes. Pending bytes mean a bad envelope,
+ // not command input; never restore echo over them.
+ var fds = [_]std.posix.pollfd{.{ .fd = 0, .events = std.posix.POLL.IN, .revents = 0 }};
+ if (try std.posix.poll(&fds, 0) != 0) return error.Payload;
+ _ = try freshValues(parsed.value, started, try bootSeconds());
+ return env;
+}
fn run() !void {
try std.posix.setrlimit(.CORE, .{ .cur = 0, .max = 0 });
const raw = try std.process.argsAlloc(allocator);
if (raw.len > 1 and std.mem.eql(u8, raw[1], "--exec-environment-stdin-v1")) return freshExec(raw);
+ if (raw.len > 1 and std.mem.eql(u8, raw[1], "--exec-environment-tty-v2")) return interactiveFreshExec(raw);
const health = raw.len > 1 and std.mem.eql(u8, raw[1], "--health");
const mounted_exec = raw.len > 1 and std.mem.eql(u8, raw[1], "--exec-mounted-v1");
if (health) {
diff --git a/packages/runtime-core/src/error.rs b/packages/runtime-core/src/error.rs
index 7ddf9b888..d2f4605e4 100644
--- a/packages/runtime-core/src/error.rs
+++ b/packages/runtime-core/src/error.rs
@@ -135,6 +135,9 @@ pub struct PlanChangeDiagnostic {
/// `process_refusals` and `plan_change`.
#[derive(Debug, Default, Serialize)]
pub struct RefusalDiagnostics {
+ /// Value-free durable interactive-session UUID, never argv or environment.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub session: Option,
/// Every service whose selected process a private wrapper refused, found in
/// one pass before any runtime state is reserved.
#[serde(skip_serializing_if = "Option::is_none")]
@@ -190,6 +193,15 @@ pub struct CandidateError {
}
impl CandidateError {
+ pub fn with_session(mut self, session: &str) -> Self {
+ if crate::provider::environment_recovery::uuid(session) {
+ self.refusals.get_or_insert_default().session = Some(session.into());
+ }
+ self
+ }
+ pub fn session(&self) -> Option<&str> {
+ self.refusals.as_ref()?.session.as_deref()
+ }
pub fn new(code: &'static str, message: impl Into) -> Self {
let message = message.into();
Self {
diff --git a/packages/runtime-core/src/interactive_exec_cli.rs b/packages/runtime-core/src/interactive_exec_cli.rs
new file mode 100644
index 000000000..393e663f8
--- /dev/null
+++ b/packages/runtime-core/src/interactive_exec_cli.rs
@@ -0,0 +1,126 @@
+//! The terminal is inherited unchanged; fd 3 is the private duplex control
+//! channel. Its input is length-delimited and its output is value-free JSON.
+use hack_runtime_core::{
+ Candidate, CandidateError,
+ provider::graph::interactive_exec::{self, Options},
+};
+use std::{
+ collections::BTreeMap,
+ fs::File,
+ io::Write,
+ os::fd::{FromRawFd, OwnedFd},
+};
+
+fn invalid() -> CandidateError {
+ CandidateError::new(
+ "interactive_exec_arguments",
+ "Use graph exec-interactive --run-id ID --service NAME --expect-owner ID --expect-plan SHA --expect-namespace SHA [--native] [--workdir DIR] [--expect-container ID --expect-generation SHA] -- PROGRAM [ARG...].",
+ )
+}
+
+pub fn command(candidate: &Candidate, args: &[&str]) -> Result {
+ let mut values = BTreeMap::new();
+ let mut native = false;
+ let mut index = 0;
+ let mut argv = Vec::new();
+ while index < args.len() {
+ let key = args[index];
+ index += 1;
+ if key == "--" {
+ argv = args[index..].iter().map(|v| (*v).to_owned()).collect();
+ break;
+ }
+ if key == "--native" && !native {
+ native = true;
+ continue;
+ }
+ if ![
+ "--run-id",
+ "--service",
+ "--expect-owner",
+ "--expect-plan",
+ "--expect-namespace",
+ "--workdir",
+ "--expect-container",
+ "--expect-generation",
+ ]
+ .contains(&key)
+ {
+ return Err(invalid());
+ }
+ let value = *args.get(index).ok_or_else(invalid)?;
+ index += 1;
+ if values.insert(key, value).is_some() {
+ return Err(invalid());
+ }
+ }
+ let get = |key| values.get(key).copied().ok_or_else(invalid);
+ let options = Options {
+ run: get("--run-id")?,
+ service: get("--service")?,
+ owner: get("--expect-owner")?,
+ plan: get("--expect-plan")?,
+ namespace: get("--expect-namespace")?,
+ native,
+ argv: &argv,
+ workdir: values.get("--workdir").copied(),
+ expected_container: values.get("--expect-container").copied(),
+ expected_generation: values.get("--expect-generation").copied(),
+ };
+ options.validate()?;
+ // SAFETY: this dedicated entrypoint owns the explicitly inherited fd 3.
+ // Duplicate the report end before the terminal session consumes its leash.
+ if unsafe { libc::fcntl(3, libc::F_GETFD) } < 0 {
+ return Err(invalid());
+ }
+ let frontend = File::from(unsafe { OwnedFd::from_raw_fd(3) });
+ hack_runtime_core::provider::private_input::validate_duplex(&frontend)?;
+ let mut report = frontend.try_clone().map_err(|_| invalid())?;
+ let result = interactive_exec::execute(candidate, &options, frontend.into());
+ let (code, value) = match result {
+ Ok(completion) => (
+ completion.exit_code,
+ serde_json::json!({"version":1,"outcome":"completed","completion":completion}),
+ ),
+ Err(error) => (
+ 1,
+ serde_json::json!({"version":1,"outcome":"refused","code":error.code,"session":error.session()}),
+ ),
+ };
+ let bytes = serde_json::to_vec(&value).map_err(|_| invalid())?;
+ // The report is under 1 KiB and never contains command output or values.
+ report.write_all(&bytes).map_err(|_| invalid())?;
+ report.write_all(b"\n").map_err(|_| invalid())?;
+ Ok(code)
+}
+
+pub fn sessions(candidate: &Candidate, run: &str, service: &str) -> Result<(), CandidateError> {
+ if run.len() != 32
+ || !run
+ .bytes()
+ .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+ || service.is_empty()
+ || service.len() > 128
+ || !service
+ .bytes()
+ .all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b))
+ {
+ return Err(invalid());
+ }
+ #[cfg(target_os = "macos")]
+ {
+ let records = hack_runtime_core::provider::interactive_session::retained(candidate)?;
+ let sessions: Vec<_> = records
+ .into_iter()
+ .filter(|r| r.binding.run == run && r.binding.service == service)
+ .map(|r| r.session)
+ .collect();
+ println!("{}", serde_json::json!({"version":1,"sessions":sessions}));
+ Ok(())
+ }
+ #[cfg(not(target_os = "macos"))]
+ {
+ let _ = (candidate, run, service);
+ Err(invalid())
+ }
+}
diff --git a/packages/runtime-core/src/main.rs b/packages/runtime-core/src/main.rs
index 4006d5540..775242237 100644
--- a/packages/runtime-core/src/main.rs
+++ b/packages/runtime-core/src/main.rs
@@ -1,4 +1,5 @@
mod graph_cli;
+mod interactive_exec_cli;
mod log_follow_cli;
mod normalized_cli;
mod prepared_base_cli;
@@ -368,6 +369,21 @@ fn run() -> Result<(), CandidateError> {
["graph", "logs-follow", arguments @ ..] => {
log_follow_cli::command(&discover_candidate(&requested)?, arguments)?;
}
+ ["graph", "exec-interactive", arguments @ ..] => {
+ let code = interactive_exec_cli::command(&discover_candidate(&requested)?, arguments)?;
+ std::process::exit(code);
+ }
+ [
+ "graph",
+ "interactive-sessions",
+ "--run-id",
+ run,
+ "--service",
+ service,
+ "--json",
+ ] => {
+ interactive_exec_cli::sessions(&discover_candidate(&requested)?, run, service)?;
+ }
["graph", arguments @ ..] => {
let candidate = discover_candidate(&requested)?;
let result = graph_cli::command(&candidate, arguments)?;
diff --git a/packages/runtime-core/src/provider/engine.rs b/packages/runtime-core/src/provider/engine.rs
index 1a5376d5e..6573feb67 100644
--- a/packages/runtime-core/src/provider/engine.rs
+++ b/packages/runtime-core/src/provider/engine.rs
@@ -12,6 +12,8 @@ use std::io::Read;
use std::path::Path;
use std::time::{Duration, Instant};
+#[cfg(any(target_os = "macos", test))]
+pub(in crate::provider) mod interactive_exec;
#[cfg(any(target_os = "macos", test))]
pub(in crate::provider) mod log_follow;
#[cfg(any(all(target_os = "macos", feature = "environment-launcher"), test))]
@@ -705,6 +707,18 @@ impl<'a> Observer<'a> {
Ok(value)
}
+ /// Recheck after failures too: a new-boot 404 cannot prove old-boot absence.
+ #[cfg(target_os = "macos")]
+ pub(super) fn get_until(&self, path: &str, deadline: Instant) -> Result {
+ self.guest.verify()?;
+ let bytes = self
+ .transport
+ .request_bytes_until(Method::GET, path, None, Some(deadline));
+ self.guest.verify()?;
+ serde_json::from_slice(&bytes?)
+ .map_err(|_| failure("Malformed engine response; values omitted."))
+ }
+
pub(super) fn storage_usage(&self) -> Result {
self.info()?;
self.guest.verify()?;
diff --git a/packages/runtime-core/src/provider/engine/interactive_exec.rs b/packages/runtime-core/src/provider/engine/interactive_exec.rs
new file mode 100644
index 000000000..03e1e5be1
--- /dev/null
+++ b/packages/runtime-core/src/provider/engine/interactive_exec.rs
@@ -0,0 +1,401 @@
+//! One non-replayed TTY start. The caller retains mutation authority only until
+//! upgrade; streaming and exact exec inspection use a lease-free connection.
+#[cfg(target_os = "macos")]
+use super::*;
+use crate::CandidateError;
+#[cfg(target_os = "macos")]
+use crate::provider::interactive_session::{self, Binding, Receipt};
+use base64::Engine as _;
+use std::{
+ collections::BTreeMap,
+ io::{Read, Write},
+ os::unix::net::UnixStream,
+ time::{Duration, Instant},
+};
+use zeroize::Zeroizing;
+
+fn uncertain() -> CandidateError {
+ CandidateError::new(
+ "session_cleanup_uncertain",
+ "Interactive session completion or cleanup is unconfirmed; its durable receipt was retained. No request was replayed.",
+ )
+}
+fn remaining(deadline: Instant) -> Result {
+ deadline
+ .checked_duration_since(Instant::now())
+ .filter(|v| !v.is_zero())
+ .ok_or_else(uncertain)
+}
+
+#[cfg(target_os = "macos")]
+pub(in crate::provider) struct Attached {
+ pub receipt: Receipt,
+ pub stream: UnixStream,
+}
+
+#[cfg(target_os = "macos")]
+impl Engine<'_> {
+ pub(in crate::provider) fn interactive_start(
+ &self,
+ candidate: &Candidate,
+ binding: Binding,
+ argv: &[String],
+ workdir: Option<&str>,
+ deadline: Instant,
+ fresh: &dyn Fn() -> Result<(), CandidateError>,
+ ) -> Result {
+ if self.cleanup_only || !binding.valid() {
+ return Err(uncertain());
+ }
+ fresh()?;
+ self.guest.verify()?;
+ self.guest.before_effect()?;
+ let intent = interactive_session::admit(candidate, binding.clone())?;
+ let session = intent.receipt().session.clone();
+ let mut intent = Some(intent);
+ let mut dispatched = false;
+ let effect = || -> Result<(), CandidateError> {
+ fresh()?;
+ self.guest.verify()?;
+ self.guest.before_effect()?;
+ remaining(deadline)?;
+ Ok(())
+ };
+ let result: Result = (|| {
+ effect()?;
+ let mut body = serde_json::json!({"AttachStdin":true,"AttachStdout":true,"AttachStderr":true,"Tty":true,"Privileged":false,"DetachKeys":"ctrl-p,ctrl-q","Cmd":argv});
+ if let Some(path) = workdir {
+ body["WorkingDir"] = Value::String(path.into());
+ }
+ let transport = Transport::new(&self.guest.engine_socket()?, remaining(deadline)?)?;
+ effect()?;
+ dispatched = true;
+ let created = transport.request(
+ Method::POST,
+ &format!("/v1.53/containers/{}/exec", binding.container),
+ Some(&body),
+ )?;
+ let id = created["Id"]
+ .as_str()
+ .filter(|id| {
+ id.len() == 64
+ && id
+ .bytes()
+ .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+ })
+ .ok_or_else(uncertain)?;
+ let receipt = intent
+ .take()
+ .ok_or_else(uncertain)?
+ .record_exec(candidate, id)?;
+ effect()?;
+ let mut stream =
+ crate::provider::agent::connect_until(&self.guest.engine_socket()?, deadline)
+ .map_err(|_| uncertain())?;
+ // Recheck after connect, immediately before the non-replayed start.
+ effect()?;
+ super::relay_exec::start_mode(&mut stream, id, deadline, true)?;
+ fresh()?;
+ self.guest.verify()?;
+ Ok(Attached { receipt, stream })
+ })();
+ match result {
+ Err(error) if !dispatched => {
+ if intent.ok_or_else(uncertain)?.withdraw(candidate).is_err() {
+ Err(uncertain().with_session(&session))
+ } else {
+ Err(error)
+ }
+ }
+ Err(_) => Err(uncertain().with_session(&session)),
+ Ok(attached) => Ok(attached),
+ }
+ }
+}
+
+/// Values never enter Engine create configuration. Both markers are stripped
+/// byte-by-byte, preserving any command output coalesced with the ACK.
+pub(in crate::provider) fn deliver(
+ stream: &mut UnixStream,
+ values: &BTreeMap,
+ deadline: Instant,
+ live: &dyn Fn() -> Result<(), CandidateError>,
+) -> Result<(), CandidateError> {
+ let deadline = deadline.min(Instant::now() + Duration::from_secs(5));
+ let marker = line(stream, deadline, live)?;
+ let now = std::str::from_utf8(&marker)
+ .ok()
+ .and_then(|s| s.strip_prefix("HKEE2 "))
+ .and_then(|s| s.strip_suffix('\n'))
+ .filter(|s| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()))
+ .and_then(|s| s.parse::().ok())
+ .ok_or_else(uncertain)?;
+ let expires = now
+ .checked_add(remaining(deadline)?.as_secs().max(1))
+ .ok_or_else(uncertain)?;
+ #[derive(serde::Serialize)]
+ struct Envelope<'a> {
+ version: u8,
+ expires: u64,
+ values: &'a BTreeMap,
+ }
+ // Fixed storage avoids leaving private bytes in abandoned Vec allocations.
+ let mut json = Zeroizing::new(vec![0u8; 65536]);
+ let mut writer = std::io::Cursor::new(json.as_mut_slice());
+ serde_json::to_writer(
+ &mut writer,
+ &Envelope {
+ version: 1,
+ expires,
+ values,
+ },
+ )
+ .map_err(|_| uncertain())?;
+ let length = writer.position() as usize;
+ json.truncate(length);
+ let encoded = Zeroizing::new(base64::engine::general_purpose::STANDARD.encode(&*json));
+ let header = format!("HKEE2 {}\n", encoded.len());
+ for mut bytes in [header.as_bytes(), encoded.as_bytes()] {
+ while !bytes.is_empty() {
+ live()?;
+ stream
+ .set_write_timeout(Some(remaining(deadline)?.min(Duration::from_millis(100))))
+ .map_err(|_| uncertain())?;
+ match stream.write(bytes) {
+ Ok(0) => return Err(uncertain()),
+ Ok(n) => bytes = &bytes[n..],
+ Err(e)
+ if matches!(
+ e.kind(),
+ std::io::ErrorKind::WouldBlock
+ | std::io::ErrorKind::TimedOut
+ | std::io::ErrorKind::Interrupted
+ ) => {}
+ Err(_) => return Err(uncertain()),
+ }
+ }
+ }
+ let ack = line(stream, deadline, live)?;
+ if ack != b"HKEE2 OK\n" && ack != b"HKEE2 OK\r\n" {
+ return Err(uncertain());
+ }
+ live()?;
+ remaining(deadline)?;
+ Ok(())
+}
+fn line(
+ stream: &mut UnixStream,
+ deadline: Instant,
+ live: &dyn Fn() -> Result<(), CandidateError>,
+) -> Result, CandidateError> {
+ let mut bytes = Vec::with_capacity(64);
+ loop {
+ live()?;
+ if bytes.len() == 64 {
+ return Err(uncertain());
+ }
+ stream
+ .set_read_timeout(Some(remaining(deadline)?.min(Duration::from_millis(100))))
+ .map_err(|_| uncertain())?;
+ let mut byte = [0];
+ match stream.read(&mut byte) {
+ Ok(0) => return Err(uncertain()),
+ Ok(_) => {
+ bytes.push(byte[0]);
+ if byte[0] == b'\n' {
+ return Ok(bytes);
+ }
+ }
+ Err(e)
+ if matches!(
+ e.kind(),
+ std::io::ErrorKind::WouldBlock
+ | std::io::ErrorKind::TimedOut
+ | std::io::ErrorKind::Interrupted
+ ) => {}
+ Err(_) => return Err(uncertain()),
+ }
+ }
+}
+
+#[cfg(target_os = "macos")]
+impl Observer<'_> {
+ /// Narrow capability: resize only an already-created exec in this exact
+ /// running container generation. No arbitrary lease-free Engine POST.
+ pub(in crate::provider) fn interactive_resize(
+ &self,
+ receipt: &Receipt,
+ rows: u16,
+ cols: u16,
+ ) -> Result<(), CandidateError> {
+ let deadline = Instant::now() + Duration::from_millis(750);
+ self.interactive_binding(receipt, deadline)?;
+ let id = receipt.exec.as_deref().ok_or_else(uncertain)?;
+ let transport = Transport::new(&self.guest.engine_socket()?, Duration::from_millis(750))?;
+ transport.request_bytes_until(
+ Method::POST,
+ &format!("/v1.53/exec/{id}/resize?h={rows}&w={cols}"),
+ None,
+ Some(deadline),
+ )?;
+ self.guest.verify()?;
+ Ok(())
+ }
+ fn interactive_binding(
+ &self,
+ receipt: &Receipt,
+ deadline: Instant,
+ ) -> Result {
+ let b = &receipt.binding;
+ if !receipt.valid() || self.incarnation() != b.owner || self.guest.boot_id()? != b.boot {
+ return Err(uncertain());
+ }
+ let value = self.get_until(&format!("/v1.53/containers/{}/json", b.container), deadline)?;
+ if value["Id"] != b.container
+ || value["State"]["StartedAt"] != b.generation
+ || value["State"]["Running"] != true
+ {
+ return Err(uncertain());
+ }
+ let id = receipt.exec.as_deref().ok_or_else(uncertain)?;
+ let exec = self.get_until(&format!("/v1.53/exec/{id}/json"), deadline)?;
+ if exec["ID"] != id || exec["ContainerID"] != b.container {
+ return Err(uncertain());
+ }
+ Ok(exec)
+ }
+ pub(in crate::provider) fn interactive_exit(
+ &self,
+ receipt: &Receipt,
+ ) -> Result {
+ let deadline = Instant::now() + Duration::from_secs(2);
+ wait_exit(deadline, &mut || {
+ self.interactive_binding(receipt, deadline)
+ })
+ }
+}
+
+fn wait_exit(
+ deadline: Instant,
+ inspect: &mut dyn FnMut() -> Result,
+) -> Result {
+ loop {
+ remaining(deadline)?;
+ let value = inspect()?;
+ remaining(deadline)?;
+ if value["Running"] == false {
+ return value["ExitCode"]
+ .as_i64()
+ .filter(|c| (0..=255).contains(c))
+ .map(|c| c as i32)
+ .ok_or_else(uncertain);
+ }
+ if value["Running"] != true {
+ return Err(uncertain());
+ }
+ std::thread::sleep(remaining(deadline)?.min(Duration::from_millis(50)));
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ #[test]
+ fn eof_can_precede_exit_publication_but_polling_is_bounded_and_fail_closed() {
+ let mut calls = 0;
+ assert_eq!(
+ wait_exit(Instant::now() + Duration::from_secs(1), &mut || {
+ calls += 1;
+ Ok(serde_json::json!({"Running":calls < 3,"ExitCode":17}))
+ })
+ .unwrap(),
+ 17
+ );
+ assert_eq!(calls, 3);
+ assert_eq!(
+ wait_exit(Instant::now() + Duration::from_millis(10), &mut || Ok(
+ serde_json::json!({"Running":true})
+ ))
+ .unwrap_err()
+ .code,
+ "session_cleanup_uncertain"
+ );
+ let mut calls = 0;
+ assert_eq!(
+ wait_exit(Instant::now() + Duration::from_secs(1), &mut || {
+ calls += 1;
+ Err(CandidateError::new("selection_changed", "Fixed refusal."))
+ })
+ .unwrap_err()
+ .code,
+ "selection_changed"
+ );
+ assert_eq!(calls, 1);
+ }
+ #[test]
+ fn private_handshake_preserves_coalesced_command_output() {
+ for ack in [b"HKEE2 OK\n".as_slice(), b"HKEE2 OK\r\n"] {
+ let (mut client, mut server) = UnixStream::pair().unwrap();
+ let ack = ack.to_vec();
+ let worker = std::thread::spawn(move || {
+ server.write_all(b"HKEE2 123\n").unwrap();
+ let header = line(
+ &mut server,
+ Instant::now() + Duration::from_secs(2),
+ &|| Ok(()),
+ )
+ .unwrap();
+ let length: usize = std::str::from_utf8(&header)
+ .unwrap()
+ .trim()
+ .strip_prefix("HKEE2 ")
+ .unwrap()
+ .parse()
+ .unwrap();
+ let mut encoded = vec![0; length];
+ server.read_exact(&mut encoded).unwrap();
+ assert!(!encoded.contains(&0x10) && !encoded.contains(&0x11));
+ let json = base64::engine::general_purpose::STANDARD
+ .decode(encoded)
+ .unwrap();
+ let value: serde_json::Value = serde_json::from_slice(&json).unwrap();
+ assert_eq!(value["values"]["TOKEN"], "synthetic\u{10}\u{11}");
+ server
+ .write_all(&[ack, b"early-output".to_vec()].concat())
+ .unwrap();
+ });
+ deliver(
+ &mut client,
+ &BTreeMap::from([("TOKEN".into(), "synthetic\u{10}\u{11}".into())]),
+ Instant::now() + Duration::from_secs(2),
+ &|| Ok(()),
+ )
+ .unwrap();
+ let mut output = [0; 12];
+ client.read_exact(&mut output).unwrap();
+ assert_eq!(&output, b"early-output");
+ worker.join().unwrap();
+ }
+ }
+ #[test]
+ fn old_or_malformed_launcher_receives_no_private_bytes() {
+ for marker in [
+ b"HKEE1 123\n".as_slice(),
+ b"bad\n",
+ b"HKEE2 private-canary\n",
+ ] {
+ let (mut client, mut server) = UnixStream::pair().unwrap();
+ server.write_all(marker).unwrap();
+ let error = deliver(
+ &mut client,
+ &BTreeMap::new(),
+ Instant::now() + Duration::from_secs(1),
+ &|| Ok(()),
+ )
+ .unwrap_err();
+ assert!(!error.message.contains("canary"));
+ client.shutdown(std::net::Shutdown::Both).unwrap();
+ assert_eq!(server.read(&mut [0]).unwrap(), 0);
+ }
+ }
+}
diff --git a/packages/runtime-core/src/provider/engine/relay_exec.rs b/packages/runtime-core/src/provider/engine/relay_exec.rs
index 0910f4f0c..a4790b839 100644
--- a/packages/runtime-core/src/provider/engine/relay_exec.rs
+++ b/packages/runtime-core/src/provider/engine/relay_exec.rs
@@ -122,7 +122,19 @@ pub(super) fn start(
id: &str,
deadline: Instant,
) -> Result<(), CandidateError> {
- let body = r#"{"Detach":false,"Tty":false}"#;
+ start_mode(stream, id, deadline, false)
+}
+pub(super) fn start_mode(
+ stream: &mut UnixStream,
+ id: &str,
+ deadline: Instant,
+ tty: bool,
+) -> Result<(), CandidateError> {
+ let body = if tty {
+ r#"{"Detach":false,"Tty":true}"#
+ } else {
+ r#"{"Detach":false,"Tty":false}"#
+ };
let request = format!(
"POST /v1.53/exec/{id}/start HTTP/1.1\r\nHost: hack-local\r\nConnection: Upgrade\r\nUpgrade: tcp\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
body.len()
@@ -238,6 +250,26 @@ mod tests {
bytes
}
#[test]
+ fn interactive_upgrade_preserves_output_in_the_header_read() {
+ let (mut client, mut server) = UnixStream::pair().unwrap();
+ let worker = std::thread::spawn(move || {
+ let wire = request(&mut server);
+ assert!(wire.ends_with(br#"{"Detach":false,"Tty":true}"#));
+ server.write_all(b"HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: tcp\r\n\r\nfirst-output").unwrap();
+ });
+ start_mode(
+ &mut client,
+ &"a".repeat(64),
+ Instant::now() + Duration::from_secs(2),
+ true,
+ )
+ .unwrap();
+ let mut output = Vec::new();
+ client.read_to_end(&mut output).unwrap();
+ assert_eq!(output, b"first-output");
+ worker.join().unwrap();
+ }
+ #[test]
fn real_upgrade_keeps_private_stdin_out_of_http_and_reads_fragmented_output() {
let (mut client, mut server) = UnixStream::pair().unwrap();
let worker = std::thread::spawn(move || {
diff --git a/packages/runtime-core/src/provider/environment_recovery.rs b/packages/runtime-core/src/provider/environment_recovery.rs
index 9d6d82295..47e61ead5 100644
--- a/packages/runtime-core/src/provider/environment_recovery.rs
+++ b/packages/runtime-core/src/provider/environment_recovery.rs
@@ -57,7 +57,7 @@ fn hex(value: &str, len: usize) -> bool {
.bytes()
.all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c))
}
-pub(super) fn uuid(value: &str) -> bool {
+pub(crate) fn uuid(value: &str) -> bool {
value.len() == 36
&& value.bytes().enumerate().all(|(i, c)| {
if [8, 13, 18, 23].contains(&i) {
diff --git a/packages/runtime-core/src/provider/graph/foreground/transport.rs b/packages/runtime-core/src/provider/graph/foreground/transport.rs
index 58f4434a4..bb2cbd4ae 100644
--- a/packages/runtime-core/src/provider/graph/foreground/transport.rs
+++ b/packages/runtime-core/src/provider/graph/foreground/transport.rs
@@ -1525,7 +1525,7 @@ pub(in crate::provider::graph) fn resume_recovered_publisher(
/// The framed client shuts down only its write side. A read-side EOF therefore
/// cannot prove abandonment; EVFILT_WRITE reports EV_EOF when its read side closes.
-pub(super) struct ClientWatch {
+pub(in crate::provider::graph) struct ClientWatch {
queue: OwnedFd,
}
impl ClientWatch {
diff --git a/packages/runtime-core/src/provider/graph/interactive_exec.rs b/packages/runtime-core/src/provider/graph/interactive_exec.rs
new file mode 100644
index 000000000..434aaca75
--- /dev/null
+++ b/packages/runtime-core/src/provider/graph/interactive_exec.rs
@@ -0,0 +1,207 @@
+//! Exact interactive admission followed by a lease-free terminal lifetime.
+use crate::{Candidate, CandidateError};
+use std::os::fd::OwnedFd;
+
+pub struct Options<'a> {
+ pub run: &'a str,
+ pub service: &'a str,
+ pub owner: &'a str,
+ pub plan: &'a str,
+ pub namespace: &'a str,
+ pub native: bool,
+ pub argv: &'a [String],
+ pub workdir: Option<&'a str>,
+ pub expected_container: Option<&'a str>,
+ pub expected_generation: Option<&'a str>,
+}
+fn refused() -> CandidateError {
+ CandidateError::new(
+ "interactive_exec_refused",
+ "Interactive exec requires a current ready service, exact ownership and a foreground terminal; no fallback was attempted.",
+ )
+}
+impl Options<'_> {
+ pub fn validate(&self) -> Result<(), CandidateError> {
+ if !super::hex(self.run, 32)
+ || !super::hex(self.owner, 32)
+ || !super::hex(self.plan, 64)
+ || !super::hex(self.namespace, 64)
+ || self.service.is_empty()
+ || self.service.len() > 128
+ || !self
+ .service
+ .bytes()
+ .all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b))
+ || self.argv.is_empty()
+ || self.argv.len() > 256
+ || self.argv[0].is_empty()
+ || self
+ .argv
+ .iter()
+ .any(|s| s.len() > 16384 || s.contains('\0'))
+ || self.argv.iter().map(String::len).sum::() > 65536
+ || self
+ .workdir
+ .is_some_and(|s| !s.starts_with('/') || s.len() > 4096 || s.contains('\0'))
+ || self
+ .expected_container
+ .is_some_and(|id| !super::hex(id, 64))
+ || self
+ .expected_generation
+ .is_some_and(|id| !super::hex(id, 64))
+ || self.expected_container.is_some() != self.expected_generation.is_some()
+ {
+ return Err(refused());
+ }
+ Ok(())
+ }
+}
+
+#[derive(serde::Serialize)]
+pub struct Completion {
+ pub exit_code: i32,
+ pub session: String,
+ pub retained: bool,
+}
+pub fn execute(
+ candidate: &Candidate,
+ options: &Options<'_>,
+ frontend: OwnedFd,
+) -> Result {
+ options.validate()?;
+ #[cfg(target_os = "macos")]
+ {
+ native(candidate, options, frontend)
+ }
+ #[cfg(not(target_os = "macos"))]
+ {
+ let _ = (candidate, frontend);
+ Err(CandidateError::new(
+ "interactive_exec_platform",
+ "Native interactive exec requires macOS.",
+ ))
+ }
+}
+
+#[cfg(target_os = "macos")]
+fn native(
+ candidate: &Candidate,
+ options: &Options<'_>,
+ frontend: OwnedFd,
+) -> Result {
+ use crate::provider::{
+ engine::{Observer, interactive_exec},
+ interactive_session,
+ interactive_terminal::{Leash, Terminal},
+ managed_environment,
+ };
+ use std::time::{Duration, Instant};
+ let mut terminal = Terminal::open()?;
+ let ingress = Instant::now();
+ let (leash, payload) = Leash::receive(frontend)?;
+ if options.native && !payload.is_empty() {
+ return Err(refused());
+ }
+ let managed = if payload.is_empty() {
+ None
+ } else {
+ if options.expected_container.is_none() {
+ return Err(refused());
+ }
+ Some(managed_environment::receive_interactive_bytes(
+ &payload,
+ options.plan,
+ options.run,
+ ingress,
+ )?)
+ };
+ drop(payload);
+ leash.check()?;
+ let _ = interactive_session::reconcile_observed(candidate);
+ let admission = if options.native {
+ #[cfg(feature = "native-config-plan")]
+ {
+ super::native::foreground::interactive(
+ candidate,
+ options.run,
+ options.service,
+ options.argv,
+ options.workdir,
+ (options.owner, options.plan, options.namespace),
+ &leash,
+ )
+ }
+ #[cfg(not(feature = "native-config-plan"))]
+ {
+ return Err(refused());
+ }
+ } else {
+ let selected = super::service_selection(candidate, options.run, options.service)?;
+ if selected.owner != options.owner
+ || selected.plan != options.plan
+ || selected.namespace != options.namespace
+ || options
+ .expected_container
+ .is_some_and(|v| v != selected.container)
+ || options
+ .expected_generation
+ .is_some_and(|v| v != selected.generation)
+ {
+ return Err(refused());
+ }
+ super::service_exec::interactive(
+ candidate,
+ &super::ServiceExecOptions {
+ run: options.run,
+ service: options.service,
+ expected_container: &selected.container,
+ expected_boot: &selected.boot,
+ expected_generation: &selected.generation,
+ argv: options.argv,
+ workdir: options.workdir,
+ timeout: Duration::from_secs(30),
+ },
+ managed.as_ref(),
+ &|| leash.check(),
+ )
+ };
+ let mut attached = admission?;
+ let receipt = &attached.receipt;
+ let result = (|| {
+ leash.check()?;
+ if let Some(managed) = &managed {
+ interactive_exec::deliver(
+ &mut attached.stream,
+ managed.values().get(options.service).ok_or_else(refused)?,
+ managed.deadline(),
+ &|| leash.check(),
+ )?;
+ }
+ drop(managed);
+ let observer = Observer::connect(candidate)?;
+ crate::provider::interactive_terminal::with_resize_worker(
+ |rows, cols| {
+ let _ = observer.interactive_resize(receipt, rows, cols);
+ },
+ |resize| {
+ terminal.relay(&mut attached.stream, &leash, &mut |rows, cols| {
+ resize(rows, cols)
+ })
+ },
+ )?;
+ observer.interactive_exit(receipt)
+ })();
+ terminal.restore()?;
+ drop(attached.stream);
+ let _ = interactive_session::reconcile_observed(candidate);
+ let remains = interactive_session::retained(candidate)?
+ .iter()
+ .any(|r| r.session == receipt.session);
+ result
+ .map(|exit_code| Completion {
+ exit_code,
+ session: receipt.session.clone(),
+ retained: remains,
+ })
+ .map_err(|error| error.with_session(&receipt.session))
+}
diff --git a/packages/runtime-core/src/provider/graph/mod.rs b/packages/runtime-core/src/provider/graph/mod.rs
index c2b6862f8..0557197c4 100644
--- a/packages/runtime-core/src/provider/graph/mod.rs
+++ b/packages/runtime-core/src/provider/graph/mod.rs
@@ -1,4 +1,5 @@
//! Fresh owned graph attempts. Recovery only observes or cleans recorded resources; never replay.
+pub mod interactive_exec;
pub mod one_off;
mod service_exec;
pub use service_exec::{
@@ -2234,6 +2235,8 @@ fn cleanup_owned_fenced(
.into();
state::write(&root.join("state.json"), &receipt)?;
fence()?;
+ #[cfg(target_os = "macos")]
+ crate::provider::interactive_session::after_cleanup(candidate);
Ok(receipt)
}
diff --git a/packages/runtime-core/src/provider/graph/native/foreground.rs b/packages/runtime-core/src/provider/graph/native/foreground.rs
index 4b9462592..b9eaad96c 100644
--- a/packages/runtime-core/src/provider/graph/native/foreground.rs
+++ b/packages/runtime-core/src/provider/graph/native/foreground.rs
@@ -32,6 +32,7 @@ pub enum Action {
Cleanup,
Logs,
Exec,
+ InteractiveExec,
Run,
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -78,7 +79,7 @@ impl Request {
|| self.review != review.review_id()
|| match (&self.action, &self.logs, &self.exec, &self.one_off) {
(Action::Logs, Some(logs), None, None) => !logs.valid(),
- (Action::Exec, None, Some(exec), None) => !exec.valid(),
+ (Action::Exec | Action::InteractiveExec, None, Some(exec), None) => !exec.valid(),
(Action::Run, None, None, Some(one_off)) => !one_off.valid(),
(Action::Status | Action::Cleanup, None, None, None) => false,
_ => true,
@@ -106,6 +107,13 @@ struct Reply {
#[derive(Serialize, Deserialize)]
#[serde(tag = "outcome", rename_all = "kebab-case", deny_unknown_fields)]
enum Outcome {
+ Interactive {
+ receipt: Receipt,
+ session: crate::provider::interactive_session::Receipt,
+ },
+ InteractiveRefused {
+ session: String,
+ },
Exec {
exec: runtime::ServiceExec,
},
@@ -163,6 +171,37 @@ impl Reply {
return Err(refused());
}
let receipt = match &self.result {
+ Outcome::Interactive { receipt, session } => {
+ let selected = exec_selected.ok_or_else(refused)?;
+ let binding = &session.binding;
+ if action != Action::InteractiveExec
+ || !session.valid()
+ || session.exec.is_none()
+ || binding.backend != crate::provider::interactive_session::Backend::Authored
+ || binding.run != review.scope().run
+ || binding.namespace != review.scope().namespace
+ || binding.owner != receipt.owner
+ || binding.boot != receipt.boot
+ || binding.service != selected.service
+ || receipt.phase != Phase::ReadyObserved
+ || receipt
+ .resources
+ .get(&format!("container:{}", selected.service))
+ .and_then(|r| r.id.as_deref())
+ != Some(binding.container.as_str())
+ {
+ return Err(refused());
+ }
+ receipt
+ }
+ Outcome::InteractiveRefused { session } => {
+ if action != Action::InteractiveExec
+ || !crate::provider::environment_recovery::uuid(session)
+ {
+ return Err(refused());
+ }
+ return Ok(());
+ }
Outcome::Exec { exec } => {
if action != Action::Exec || !exec.valid(exec_selected.ok_or_else(refused)?) {
return Err(refused());
@@ -359,7 +398,32 @@ pub fn serve_with_storage_tool(
continue;
}
publication.verify()?;
+ let mut attached = None;
let result = match request.action {
+ Action::InteractiveExec => {
+ let selected = request.exec.as_ref().ok_or_else(refused)?;
+ let client = transport::ClientWatch::new(&stream)?;
+ let check = || {
+ publication.verify()?;
+ if signals.pending() || client.disconnected() {
+ return Err(refused());
+ }
+ read_pin.verify(candidate)
+ };
+ match runtime::interactive_exec(candidate, &receipt, selected, &check) {
+ Ok(session) => {
+ attached = Some(session.stream);
+ Ok(Outcome::Interactive {
+ receipt: receipt.clone(),
+ session: session.receipt,
+ })
+ }
+ Err(error) if error.session().is_some() => Ok(Outcome::InteractiveRefused {
+ session: error.session().ok_or_else(refused)?.to_owned(),
+ }),
+ Err(error) => Err(error),
+ }
+ }
Action::Status => inspect(candidate, &run).map(|snapshot| Outcome::Status { snapshot }),
Action::Cleanup => clean().map(|receipt| Outcome::Cleaned { receipt }),
Action::Exec => {
@@ -425,6 +489,20 @@ pub fn serve_with_storage_tool(
};
return Ok(receipt);
}
+ if request.action == Action::InteractiveExec {
+ // The connected stream is handed over before the bound reply; its
+ // recipient validates that reply before using the fd. Sending
+ // consumes our copy even on error, so the owner never holds a shell.
+ if crate::provider::interactive_handoff::send(
+ &stream,
+ attached,
+ std::time::Instant::now() + Duration::from_secs(5),
+ )
+ .is_err()
+ {
+ continue;
+ }
+ }
let _ = transport::write(&mut stream, &reply, Duration::from_secs(5));
}
}
@@ -438,7 +516,10 @@ pub fn request(
candidate: &Candidate,
options: RequestOptions<'_>,
) -> Result {
- if matches!(options.action, Action::Logs | Action::Exec | Action::Run) {
+ if matches!(
+ options.action,
+ Action::Logs | Action::Exec | Action::InteractiveExec | Action::Run
+ ) {
return Err(refused());
}
request_selected(candidate, options.run, options.action, None, None, None)
@@ -496,6 +577,76 @@ pub fn one_off(
}
request_selected(candidate, run, Action::Run, None, None, Some(selected))
}
+/// The live owner creates and starts under its normal guards, then returns the
+/// connected fd. The caller streams without retaining an owner/provider lease.
+pub(in crate::provider) fn interactive(
+ candidate: &Candidate,
+ run: &str,
+ service: &str,
+ argv: &[String],
+ workdir: Option<&str>,
+ pins: (&str, &str, &str),
+ leash: &crate::provider::interactive_terminal::Leash,
+) -> Result {
+ let selected = runtime::ExecSelection {
+ service: service.into(),
+ argv: argv.to_vec(),
+ workdir: workdir.map(str::to_owned),
+ };
+ if !selected.valid() {
+ return Err(refused());
+ }
+ let pin = owner::Pin::load(candidate, run)?;
+ let review = pin.review();
+ let expected = journal::read_control(candidate, review)?;
+ if (
+ expected.owner.as_str(),
+ review.review_id(),
+ review.scope().namespace,
+ ) != pins
+ {
+ return Err(refused());
+ }
+ let mut stream = pin.connect()?;
+ let _lifetime = leash.watch_control(&stream)?;
+ let input = Request {
+ version: 2,
+ kind: RequestKind::NativeGraphControl,
+ run: run.into(),
+ review: review.review_id().into(),
+ action: Action::InteractiveExec,
+ logs: None,
+ exec: Some(selected.clone()),
+ one_off: None,
+ };
+ transport::write(&mut stream, &input, Duration::from_secs(5))?;
+ let descriptor = crate::provider::interactive_handoff::receive(
+ &stream,
+ std::time::Instant::now() + Duration::from_secs(40),
+ )?;
+ let reply: Reply = transport::read(&mut stream, Duration::from_secs(5), LIMIT)?;
+ reply.validate(
+ &expected,
+ Action::InteractiveExec,
+ None,
+ Some(&selected),
+ None,
+ )?;
+ match reply.result {
+ Outcome::Interactive { session, .. } => {
+ let result = (|| {
+ journal::read_control(candidate, review)?.check_binding(&expected)?;
+ pin.verify()?;
+ let stream = descriptor.ok_or_else(refused)?;
+ Ok(crate::provider::engine::interactive_exec::Attached { receipt: session.clone(), stream })
+ })();
+ result.map_err(|_: CandidateError| CandidateError::new("session_cleanup_uncertain", "Interactive session handoff changed; the session was retained and no start was replayed.").with_session(&session.session))
+ }
+ Outcome::InteractiveRefused { session } => Err(CandidateError::new("session_cleanup_uncertain", "Interactive admission ended after its durable intent; the session was retained and no start was replayed.").with_session(&session)),
+ Outcome::Refused { code, .. } => Err(refused().with_cause_code(code)),
+ _ => Err(refused()),
+ }
+}
fn request_selected(
candidate: &Candidate,
run: &str,
diff --git a/packages/runtime-core/src/provider/graph/native/runtime.rs b/packages/runtime-core/src/provider/graph/native/runtime.rs
index 47db7c101..a0d39cd43 100644
--- a/packages/runtime-core/src/provider/graph/native/runtime.rs
+++ b/packages/runtime-core/src/provider/graph/native/runtime.rs
@@ -11,6 +11,8 @@ pub(super) use exec::ExecSelection;
pub(super) use exec::ServiceExec;
#[cfg(target_os = "macos")]
pub(super) use exec::execute as service_exec;
+#[cfg(target_os = "macos")]
+pub(super) use exec::interactive as interactive_exec;
#[cfg(any(target_os = "macos", test))]
mod run;
#[cfg(target_os = "macos")]
@@ -1498,6 +1500,8 @@ fn cleanup_inner(
finish(&observed)?;
check_startup(guard)?;
}
+ #[cfg(target_os = "macos")]
+ crate::provider::interactive_session::after_cleanup(candidate);
Ok(receipt)
}
#[cfg(test)]
diff --git a/packages/runtime-core/src/provider/graph/native/runtime/exec.rs b/packages/runtime-core/src/provider/graph/native/runtime/exec.rs
index 717fb2fb8..6d40fdb08 100644
--- a/packages/runtime-core/src/provider/graph/native/runtime/exec.rs
+++ b/packages/runtime-core/src/provider/graph/native/runtime/exec.rs
@@ -172,3 +172,73 @@ pub(super) fn execute_with(
}
Ok(result)
}
+
+#[cfg(target_os = "macos")]
+pub(in crate::provider::graph::native) fn interactive(
+ candidate: &Candidate,
+ expected: &Receipt,
+ selected: &ExecSelection,
+ guard: &dyn Fn() -> Result<(), CandidateError>,
+) -> Result {
+ use crate::provider::interactive_session::{Backend as SessionBackend, Binding};
+ if !selected.valid() {
+ return Err(refused());
+ }
+ guard()?;
+ let deadline = Instant::now() + Duration::from_secs(30);
+ let engine = Engine::connect_command_until(candidate, deadline)?;
+ let (receipt, _) = journal::load(
+ candidate,
+ expected.review.scope().run,
+ engine.guest().incarnation(),
+ engine.guest().boot_id(),
+ )?;
+ receipt.check_binding(expected)?;
+ if receipt.phase != Phase::ReadyObserved || !receipt.readiness.contains_key(&selected.service) {
+ return Err(refused());
+ }
+ let owned = OwnedBackend {
+ engine,
+ launcher: None,
+ leases: BTreeMap::new(),
+ };
+ let backend = GuardedBackend {
+ backend: &owned,
+ guard: Some(guard),
+ };
+ let resource = receipt
+ .resources
+ .get(&format!("container:{}", selected.service))
+ .filter(|r| r.kind == Kind::Container)
+ .ok_or_else(refused)?;
+ let container = resource.id.clone().ok_or_else(refused)?;
+ let generation = member(&backend, &receipt, resource)?;
+ let fresh = || {
+ guard()?;
+ managed_environment::remaining_until(deadline)?;
+ journal::read_control(candidate, &expected.review)?.check_binding(expected)?;
+ snapshot_until(&backend, receipt.clone(), Some(deadline))?;
+ if member(&backend, &receipt, resource)? != generation {
+ return Err(refused());
+ }
+ Ok(())
+ };
+ fresh()?;
+ owned.engine.interactive_start(
+ candidate,
+ Binding {
+ backend: SessionBackend::Authored,
+ namespace: receipt.review.scope().namespace.to_owned(),
+ run: receipt.review.scope().run.to_owned(),
+ owner: receipt.owner.clone(),
+ boot: receipt.boot.clone(),
+ container,
+ generation: generation.clone(),
+ service: selected.service.clone(),
+ },
+ &selected.argv,
+ selected.workdir.as_deref(),
+ deadline,
+ &fresh,
+ )
+}
diff --git a/packages/runtime-core/src/provider/graph/service_exec.rs b/packages/runtime-core/src/provider/graph/service_exec.rs
index 1a0702a5a..0a0412051 100644
--- a/packages/runtime-core/src/provider/graph/service_exec.rs
+++ b/packages/runtime-core/src/provider/graph/service_exec.rs
@@ -67,6 +67,130 @@ fn validate(options: &ServiceExecOptions<'_>) -> Result<(), CandidateError> {
Ok(())
}
+#[cfg(target_os = "macos")]
+pub(super) fn interactive(
+ candidate: &Candidate,
+ options: &ServiceExecOptions<'_>,
+ managed: Option<&crate::provider::managed_environment::Managed>,
+ live: &dyn Fn() -> Result<(), CandidateError>,
+) -> Result {
+ use crate::provider::interactive_session::{Backend, Binding};
+ use std::time::Instant;
+ validate(options)?;
+ live()?;
+ let deadline = Instant::now() + options.timeout;
+ let engine = Engine::connect_command_until(candidate, deadline)?;
+ let (receipt, root) = load(candidate, &engine, options.run)?;
+ if receipt.environment_attached && managed.is_none() {
+ return Err(error(
+ "interactive_exec_environment_required",
+ "Interactive exec requires fresh selected-service environment delivery; no startup-environment fallback was attempted.",
+ ));
+ }
+ super::startup::require_dependency_rebind_complete(&root, &receipt)?;
+ let pending = root
+ .join("state.pending")
+ .try_exists()
+ .map_err(|_| refused())?;
+ let resource = binding(&receipt, engine.guest().boot_id(), pending, options)?;
+ let container = inspect_resource(&engine, &receipt, resource)?.ok_or_else(refused)?;
+ let generation = container["State"]["StartedAt"]
+ .as_str()
+ .filter(|s| crate::provider::engine::log_follow::valid_timestamp(s))
+ .ok_or_else(refused)?
+ .to_owned();
+ let slot = if receipt.environment_attached {
+ crate::provider::environment_recovery::active_exec_slot(
+ candidate,
+ engine.guest(),
+ &receipt.run,
+ options.service,
+ &resource.name,
+ mounted_environment_slot(&container)?.as_deref(),
+ )?
+ } else {
+ None
+ };
+ if managed.is_none() && slot.is_some() && !options.argv[0].starts_with('/') {
+ #[cfg(feature = "environment-launcher")]
+ require_current_launcher(&container, &super::launcher::current_source())?;
+ #[cfg(not(feature = "environment-launcher"))]
+ return Err(refused());
+ }
+ let argv = if let Some(managed) = managed {
+ #[cfg(feature = "environment-launcher")]
+ {
+ managed.validate_binding(&receipt.plan_id, &receipt.run)?;
+ if !receipt.environment_attached
+ || managed.values().len() != 1
+ || managed
+ .values()
+ .get(options.service)
+ .is_none_or(|v| v.is_empty())
+ {
+ return Err(refused());
+ }
+ require_current_launcher(&container, &super::launcher::current_source())?;
+ let mut argv = vec![
+ "/run/hack-environment-launcher".into(),
+ "--exec-environment-tty-v2".into(),
+ ];
+ argv.extend_from_slice(options.argv);
+ argv
+ }
+ #[cfg(not(feature = "environment-launcher"))]
+ {
+ let _ = managed;
+ return Err(refused());
+ }
+ } else {
+ managed_exec_argv(&container, slot.as_deref(), options.argv)?
+ };
+ let fresh = || {
+ live()?;
+ crate::provider::managed_environment::remaining_until(deadline)?;
+ let (current, root) = load(candidate, &engine, options.run)?;
+ super::startup::require_dependency_rebind_complete(&root, ¤t)?;
+ binding(
+ ¤t,
+ engine.guest().boot_id(),
+ root.join("state.pending")
+ .try_exists()
+ .map_err(|_| refused())?,
+ options,
+ )?;
+ let value = inspect_resource(&engine, &receipt, resource)?.ok_or_else(refused)?;
+ if value["State"]["Running"] != true
+ || value["State"]["Paused"] != false
+ || value["State"]["Restarting"] != false
+ || value["State"]["StartedAt"] != generation
+ {
+ return Err(refused());
+ }
+ Ok(())
+ };
+ fresh()?;
+ let attached = engine.interactive_start(
+ candidate,
+ Binding {
+ backend: Backend::Adopted,
+ namespace: receipt.namespace.clone(),
+ run: receipt.run.clone(),
+ owner: receipt.owner.clone(),
+ boot: engine.guest().boot_id().to_owned(),
+ container: options.expected_container.to_owned(),
+ generation: generation.clone(),
+ service: options.service.to_owned(),
+ },
+ &argv,
+ options.workdir,
+ deadline,
+ &fresh,
+ )?;
+ // No provider mutation lease crosses the return into the terminal loop.
+ Ok(attached)
+}
+
/// Execute argv once with stdin closed and TTY disabled, retaining the VM mutation
/// lease throughout. A timeout or interrupted response is uncertain: the command
/// may still run and must not be replayed automatically. Inherits container user,
diff --git a/packages/runtime-core/src/provider/interactive_handoff.rs b/packages/runtime-core/src/provider/interactive_handoff.rs
new file mode 100644
index 000000000..90ed83cea
--- /dev/null
+++ b/packages/runtime-core/src/provider/interactive_handoff.rs
@@ -0,0 +1,331 @@
+//! One descriptor on the already authenticated owner connection. The receiver
+//! must validate the following bound reply before using the returned stream.
+use crate::CandidateError;
+use std::{
+ io,
+ mem::{size_of, zeroed},
+ os::fd::{AsRawFd, FromRawFd, OwnedFd},
+ os::unix::net::UnixStream,
+ time::Instant,
+};
+
+fn refused() -> CandidateError {
+ CandidateError::new(
+ "session_cleanup_uncertain",
+ "Interactive session handoff was not confirmed; its durable receipt was retained. No start was replayed.",
+ )
+}
+
+fn wait(stream: &UnixStream, events: i16, deadline: Instant) -> Result<(), CandidateError> {
+ loop {
+ let left = deadline
+ .checked_duration_since(Instant::now())
+ .ok_or_else(refused)?;
+ let mut fd = libc::pollfd {
+ fd: stream.as_raw_fd(),
+ events,
+ revents: 0,
+ };
+ // SAFETY: fd is one initialized poll entry, retained for this call.
+ let count = unsafe { libc::poll(&mut fd, 1, left.as_millis().clamp(1, 1000) as i32) };
+ if count < 0 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
+ continue;
+ }
+ if count < 0 {
+ return Err(refused());
+ }
+ if count > 0 {
+ if fd.revents & libc::POLLNVAL != 0 {
+ return Err(refused());
+ }
+ return Ok(());
+ }
+ }
+}
+
+/// Consumes the sender's copy on *every* outcome. There is never a retry after
+/// successful sendmsg; a lost reply is an uncertain session, not a new admission.
+pub(super) fn send(
+ control: &UnixStream,
+ session: Option,
+ deadline: Instant,
+) -> Result<(), CandidateError> {
+ let mut tag = u8::from(session.is_some());
+ let mut iov = libc::iovec {
+ iov_base: (&mut tag as *mut u8).cast(),
+ iov_len: 1,
+ };
+ // usize storage provides cmsghdr alignment on both supported platforms.
+ let mut ancillary = [0usize; 16];
+ // SAFETY: msghdr contains integer/pointer fields with valid zero values.
+ let mut message: libc::msghdr = unsafe { zeroed() };
+ message.msg_iov = &mut iov;
+ message.msg_iovlen = 1;
+ if let Some(session) = &session {
+ message.msg_control = ancillary.as_mut_ptr().cast();
+ // SAFETY: control storage exceeds CMSG_SPACE for one int; FIRSTHDR and
+ // DATA point within it. write_unaligned avoids relying on data alignment.
+ unsafe {
+ message.msg_controllen = libc::CMSG_SPACE(size_of::() as u32) as _;
+ let header = libc::CMSG_FIRSTHDR(&message);
+ (*header).cmsg_level = libc::SOL_SOCKET;
+ (*header).cmsg_type = libc::SCM_RIGHTS;
+ (*header).cmsg_len = libc::CMSG_LEN(size_of::() as u32) as _;
+ libc::CMSG_DATA(header)
+ .cast::()
+ .write_unaligned(session.as_raw_fd());
+ }
+ }
+ loop {
+ wait(control, libc::POLLOUT, deadline)?;
+ // SAFETY: all buffers live through sendmsg; the kernel copies the fd,
+ // which remains owned by session until this function returns.
+ let count = unsafe {
+ libc::sendmsg(
+ control.as_raw_fd(),
+ &message,
+ libc::MSG_DONTWAIT | libc::MSG_NOSIGNAL,
+ )
+ };
+ if count == 1 {
+ return Ok(());
+ }
+ if count < 0
+ && matches!(
+ io::Error::last_os_error().kind(),
+ io::ErrorKind::Interrupted | io::ErrorKind::WouldBlock
+ )
+ {
+ continue;
+ }
+ return Err(refused());
+ }
+}
+
+/// Received descriptors are owned immediately, including on truncated/invalid
+/// messages. No descriptor is exposed until its count, type and CLOEXEC are valid.
+pub(super) fn receive(
+ control: &UnixStream,
+ deadline: Instant,
+) -> Result