feat(lambda): run the test-link agent over the authenticated TCP link

The lambdatestlink agent now loads its identity, seals fw_cfg and dials the
host exactly as the normal agent does, then carries the test stream over
that mutual-TLS connection instead of a device path. Nothing tenant-facing
starts until the host's first valid frame: a refusal after the handshake,
an early close or a shutdown never starts the bootstrap. Each VM serves one
invocation and then stops.

The normal agent is unchanged and still accepts no invocations. The tagged
tests join the race tier for the two packages that carry the build tag.

Makefile

@@ -326,6 +326,7 @@ test-cover:
 test-race:
 	@echo -e "\n....Running tests with race detector for $(GO_PROJECT_NAME)...."
 	$(_Q)LOG_IGNORE=1 SPINIFEX_SQS_R3=1 go test $(GOTESTFLAGS) -race -timeout 300s ./spinifex/... ./cmd/... ./internal/... ./contracts/... ./architecture/... $(_RACEQ)
+	$(_Q)LOG_IGNORE=1 go test $(GOTESTFLAGS) -race -tags=lambdatestlink -timeout 120s ./cmd/lambda-agent/ ./spinifex/agents/lambda/hostlink/ $(_RACEQ)
 
 # Unit tests for in-repo GitHub Actions (e.g. .github/actions/e2e-analyze).
 # Kept out of `test-cover` so coverage % isn't diluted by CI-only tooling.

cmd/lambda-agent/connect.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+)
+
+// Exit statuses: the guest refused to start, the control connection failed,
+// or the host closed it.
+const (
+	exitRefused = 2
+	exitLink    = 1
+	exitClosed  = 0
+)
+
+type startup struct {
+	fwCfgDir  string
+	fwCfgRoot string
+	unload    func() error
+}
+
+func defaultStartup() startup {
+	return startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
+}
+
+// connect loads the identity, seals fw_cfg whether or not the load succeeded,
+// then dials the host. Under TLS 1.3 a host refusal is seen only on first read.
+func (s startup) connect(ctx context.Context) (*tls.Conn, int, error) {
+	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
+	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
+		return nil, exitRefused, errors.Join(loadErr, err)
+	}
+	if loadErr != nil {
+		return nil, exitRefused, loadErr
+	}
+	conn, err := hostlink.Dial(ctx, id)
+	if err != nil {
+		if ctx.Err() != nil {
+			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
+		}
+		return nil, exitLink, err
+	}
+	return conn, 0, nil
+}

cmd/lambda-agent/helpers_test.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
+	"github.com/stretchr/testify/require"
+)
+
+const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
+
+// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
+type guest struct {
+	root     string
+	unloads  int
+	sealedAt chan struct{}
+}
+
+func newGuest(t *testing.T, entries map[string][]byte) *guest {
+	t.Helper()
+	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
+	for name, b := range entries {
+		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
+		require.NoError(t, os.MkdirAll(dir, 0o700))
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
+	}
+	return g
+}
+
+func (g *guest) startup() startup {
+	return startup{
+		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
+		fwCfgRoot: g.root,
+		unload: func() error {
+			g.unloads++
+			g.sealedAt <- struct{}{}
+			return os.RemoveAll(g.root)
+		},
+	}
+}
+
+type host struct {
+	ln   *net.TCPListener
+	gen  *guestconn.Generation
+	conn chan accepted
+}
+
+func newHost(t *testing.T) *host {
+	t.Helper()
+	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	gen, err := guestconn.NewGeneration(scope, time.Hour)
+	require.NoError(t, err)
+	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
+}
+
+func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
+
+type accepted struct {
+	conn *tls.Conn
+	err  error
+}
+
+// accept runs the host end until ctx ends and reports what it accepted.
+func (h *host) accept(ctx context.Context) {
+	go func() {
+		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
+		h.conn <- accepted{c, err}
+	}()
+}
+
+// guest waits for the host to authenticate the guest.
+func (h *host) guest(t *testing.T) *tls.Conn {
+	t.Helper()
+	a := wait(t, h.conn, "host to authenticate the guest")
+	require.NoError(t, a.err)
+	return a.conn
+}
+
+// wait bounds every test wait so a stalled host or startup fails promptly.
+func wait[T any](t *testing.T, ch <-chan T, what string) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(10 * time.Second):
+		t.Fatalf("timed out waiting for %s", what)
+		panic("unreachable")
+	}
+}

cmd/lambda-agent/main.go

@@ -6,8 +6,8 @@
 // never starts the function's bootstrap: the message protocol and grant
 // fencing do not exist yet. The agent never listens for the host.
 //
-// A lambdatestlink build instead relays invocations over an unauthenticated
-// test carrier for local proofs; it is not the guest protocol.
+// A lambdatestlink build opens the same connection, then relays one invocation
+// over a test carrier for local proofs; the carrier is not the guest protocol.
 package main
 
 import (

cmd/lambda-agent/start_tcp.go

@@ -9,56 +9,28 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
-
-	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
-)
-
-// Exit statuses: the guest refused to start, the control connection failed,
-// or the host closed it.
-const (
-	exitRefused = 2
-	exitLink    = 1
-	exitClosed  = 0
 )
 
 // errNoProtocol is returned when the host sends anything: this build defines no
 // message protocol, so it accepts no work.
 var errNoProtocol = errors.New("host sent data, but this build accepts no invocations")
 
-type startup struct {
-	fwCfgDir  string
-	fwCfgRoot string
-	unload    func() error
-}
-
 func start(ctx context.Context) int {
-	s := startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
-	code, err := s.run(ctx)
+	code, err := defaultStartup().run(ctx)
 	if err != nil {
 		slog.Error("lambda-agent stopping", "exit", code, "err", err)
 	}
 	return code
 }
 
-// run loads the identity, seals fw_cfg whether or not the load succeeded,
-// then authenticates to the host and holds the connection until it closes.
+// run authenticates to the host and holds the connection until it closes.
 func (s startup) run(ctx context.Context) (int, error) {
-	id, loadErr := hostlink.LoadIdentity(s.fwCfgDir)
-	if err := hostlink.SealFwCfg(s.fwCfgRoot, s.unload); err != nil {
-		return exitRefused, errors.Join(loadErr, err)
-	}
-	if loadErr != nil {
-		return exitRefused, loadErr
-	}
-	conn, err := hostlink.Dial(ctx, id)
+	conn, code, err := s.connect(ctx)
 	if err != nil {
-		if ctx.Err() != nil {
-			err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
-		}
-		return exitLink, err
+		return code, err
 	}
 	defer conn.Close()
-	slog.Info("control connection open; invocations are not enabled in this build", "host", id.HostAddr)
+	slog.Info("control connection open; invocations are not enabled in this build", "host", conn.RemoteAddr().String())
 	if err := awaitHostClose(ctx, conn); err != nil {
 		return exitLink, err
 	}

cmd/lambda-agent/start_tcp_test.go

@@ -4,11 +4,9 @@ package main
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"net"
 	"os"
-	"path/filepath"
 	"testing"
 	"time"
 
@@ -18,89 +16,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const scope lambdav1.Scope = "ap-southeast-2/000000000001/fn-1/env-1/gen-1"
-
-// guest is a fake fw_cfg tree; unload removes it as unloading the module would.
-type guest struct {
-	root     string
-	unloads  int
-	sealedAt chan struct{}
-}
-
-func newGuest(t *testing.T, entries map[string][]byte) *guest {
-	t.Helper()
-	g := &guest{root: filepath.Join(t.TempDir(), "qemu_fw_cfg"), sealedAt: make(chan struct{}, 1)}
-	for name, b := range entries {
-		dir := filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix, name)
-		require.NoError(t, os.MkdirAll(dir, 0o700))
-		require.NoError(t, os.WriteFile(filepath.Join(dir, "raw"), b, 0o600))
-	}
-	return g
-}
-
-func (g *guest) startup() startup {
-	return startup{
-		fwCfgDir:  filepath.Join(g.root, "by_name", lambdav1.FwCfgPrefix),
-		fwCfgRoot: g.root,
-		unload: func() error {
-			g.unloads++
-			g.sealedAt <- struct{}{}
-			return os.RemoveAll(g.root)
-		},
-	}
-}
-
-type host struct {
-	ln   *net.TCPListener
-	gen  *guestconn.Generation
-	conn chan accepted
-}
-
-func newHost(t *testing.T) *host {
-	t.Helper()
-	ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
-	require.NoError(t, err)
-	t.Cleanup(func() { ln.Close() })
-	gen, err := guestconn.NewGeneration(scope, time.Hour)
-	require.NoError(t, err)
-	return &host{ln: ln, gen: gen, conn: make(chan accepted, 1)}
-}
-
-func (h *host) entries() map[string][]byte { return h.gen.GuestEntries(h.ln.Addr().String()) }
-
-type accepted struct {
-	conn *tls.Conn
-	err  error
-}
-
-// accept runs the host end until ctx ends and reports what it accepted.
-func (h *host) accept(ctx context.Context) {
-	go func() {
-		c, err := h.gen.AcceptGuest(ctx, h.ln, 2*time.Second)
-		h.conn <- accepted{c, err}
-	}()
-}
-
-// guest waits for the host to authenticate the guest.
-func (h *host) guest(t *testing.T) *tls.Conn {
-	t.Helper()
-	a := wait(t, h.conn, "host to authenticate the guest")
-	require.NoError(t, a.err)
-	return a.conn
-}
-
-// wait bounds every test wait so a stalled host or startup fails promptly.
-func wait[T any](t *testing.T, ch <-chan T, what string) T {
-	t.Helper()
-	select {
-	case v := <-ch:
-		return v
-	case <-time.After(10 * time.Second):
-		t.Fatalf("timed out waiting for %s", what)
-		panic("unreachable")
-	}
-}
-
 type result struct {
 	code int
 	err  error

cmd/lambda-agent/start_testlink.go

@@ -12,7 +12,6 @@ import (
 	"net/http"
 	"os"
 	"os/exec"
-	"syscall"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
@@ -25,34 +24,56 @@ const (
 	taskRoot    = "/var/task"
 )
 
-var (
-	linkPath = flag.String("link", os.Getenv("LAMBDA_AGENT_LINK"), "test carrier device path")
-	handler  = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
-)
+var handler = flag.String("handler", os.Getenv("_HANDLER"), "function handler passed to the bootstrap")
 
-func start(parent context.Context) int {
-	if *linkPath == "" {
-		slog.Error("no test carrier configured; refusing to start")
-		return 2
+// testlink relays one invocation over the test carrier on the authenticated
+// connection; the carrier is not the guest protocol.
+type testlink struct {
+	startup
+	runtimeAddr    string
+	startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
+}
+
+func start(ctx context.Context) int {
+	t := testlink{
+		startup:     defaultStartup(),
+		runtimeAddr: runtimeAddr,
+		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
+			return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
+		},
 	}
-	// The bootstrap is tenant code; fw_cfg must be unreadable before it runs.
-	if err := hostlink.SealFwCfg(hostlink.FwCfgRoot, hostlink.UnloadFwCfg); err != nil {
-		slog.Error("seal fw_cfg; refusing to start", "err", err)
-		return 2
-	}
-	dev, err := os.OpenFile(*linkPath, os.O_RDWR, 0)
+	code, err := t.run(ctx)
 	if err != nil {
-		slog.Error("open test carrier; refusing to start", "err", err)
-		return 2
+		slog.Error("lambda-agent stopping", "exit", code, "err", err)
+	}
+	return code
+}
+
+func (t testlink) run(parent context.Context) (int, error) {
+	conn, code, err := t.connect(parent)
+	if err != nil {
+		return code, err
+	}
+	defer conn.Close()
+	link := hostlink.NewTestStream(conn)
+
+	// The handshake can return before the host refuses the guest, so nothing
+	// tenant-facing starts until the host's first valid frame arrives.
+	first, err := link.Next(parent)
+	if errors.Is(err, hostlink.ErrShutdown) {
+		return exitClosed, nil
+	}
+	if err != nil {
+		return exitLink, fmt.Errorf("first frame from host: %w", err)
+	}
+	if parent.Err() != nil {
+		return exitLink, context.Cause(parent)
 	}
-	defer dev.Close()
-	link := hostlink.NewTestStream(dev)
 
 	srv := runtimeapi.New()
-	ln, err := net.Listen("tcp", runtimeAddr)
+	ln, err := net.Listen("tcp", t.runtimeAddr)
 	if err != nil {
-		slog.Error("listen runtime api", "err", err)
-		return 1
+		return exitLink, fmt.Errorf("listen runtime api: %w", err)
 	}
 	hs := &http.Server{Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}
 	go func() { _ = hs.Serve(ln) }()
@@ -60,20 +81,14 @@ func start(parent context.Context) int {
 
 	ctx, cancel := context.WithCancelCause(parent)
 	defer cancel(nil)
-
-	cmd := exec.CommandContext(ctx, taskRoot+"/bootstrap")
-	cmd.Env = []string{
-		"AWS_LAMBDA_RUNTIME_API=" + runtimeAddr,
+	cmd, err := t.startBootstrap(ctx, []string{
+		"AWS_LAMBDA_RUNTIME_API=" + ln.Addr().String(),
 		"_HANDLER=" + *handler,
 		"LAMBDA_TASK_ROOT=" + taskRoot,
 		"PATH=/bin:/sbin:/usr/bin",
-	}
-	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
-	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
-	cmd.WaitDelay = 5 * time.Second
-	if err := cmd.Start(); err != nil {
-		slog.Error("start bootstrap", "err", err)
-		return 1
+	})
+	if err != nil {
+		return exitLink, fmt.Errorf("start bootstrap: %w", err)
 	}
 	exited := make(chan struct{})
 	go func() {
@@ -82,12 +97,32 @@ func start(parent context.Context) int {
 		cancel(fmt.Errorf("bootstrap exited: %w", err))
 	}()
 
-	err = agent.Run(ctx, link, srv)
+	err = agent.Run(ctx, &oneInvocation{Link: link, first: &first}, srv)
 	cancel(errors.New("agent stopping"))
 	<-exited
 	if err != nil {
-		slog.Error("agent stopped", "err", err)
-		return 1
+		return exitLink, err
 	}
-	return 0
+	return exitClosed, nil
+}
+
+// oneInvocation serves the host's first request and then stops the agent:
+// each test VM runs a single invocation.
+type oneInvocation struct {
+	hostlink.Link
+	first *hostlink.Request
+}
+
+var _ hostlink.Link = (*oneInvocation)(nil)
+
+func (o *oneInvocation) Next(ctx context.Context) (hostlink.Request, error) {
+	if err := ctx.Err(); err != nil {
+		return hostlink.Request{}, err
+	}
+	if o.first == nil {
+		return hostlink.Request{}, hostlink.ErrShutdown
+	}
+	req := *o.first
+	o.first = nil
+	return req, nil
 }

cmd/lambda-agent/start_testlink_test.go

@@ -0,0 +1,179 @@
+//go:build lambdatestlink
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const helperEnv = "LAMBDA_AGENT_TEST_BOOTSTRAP"
+
+// TestHelperBootstrap is the fake function bootstrap, run as a subprocess: it
+// echoes one invocation through the Runtime API, then waits for the next.
+func TestHelperBootstrap(t *testing.T) {
+	if os.Getenv(helperEnv) != "1" {
+		return
+	}
+	base := "http://" + os.Getenv("AWS_LAMBDA_RUNTIME_API") + "/2018-06-01/runtime/invocation/"
+	resp, err := http.Get(base + "next")
+	require.NoError(t, err)
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	resp.Body.Close()
+	id := resp.Header.Get("Lambda-Runtime-Aws-Request-Id")
+	post, err := http.Post(base+id+"/response", "application/json", bytes.NewReader(body))
+	require.NoError(t, err)
+	post.Body.Close()
+	_, _ = http.Get(base + "next")
+}
+
+type bootstrapStarts struct {
+	n       int
+	started chan struct{}
+}
+
+func (b *bootstrapStarts) testlink(g *guest) testlink {
+	b.started = make(chan struct{}, 1)
+	return testlink{
+		startup:     g.startup(),
+		runtimeAddr: "127.0.0.1:0",
+		startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
+			b.n++
+			b.started <- struct{}{}
+			cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestHelperBootstrap$")
+			cmd.Env = append(env, helperEnv+"=1")
+			return cmd, cmd.Start()
+		},
+	}
+}
+
+func runTestlinkAsync(ctx context.Context, tl testlink) <-chan result {
+	out := make(chan result, 1)
+	go func() {
+		code, err := tl.run(ctx)
+		out <- result{code, err}
+	}()
+	return out
+}
+
+type result struct {
+	code int
+	err  error
+}
+
+func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	conn := h.guest(t)
+	defer conn.Close()
+	select {
+	case <-b.started:
+		t.Fatal("bootstrap started before the host sent a frame")
+	case <-time.After(100 * time.Millisecond):
+	}
+	hs := hostlink.NewTestStream(conn)
+	req := hostlink.Request{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{"a":40,"b":2}`)}
+	require.NoError(t, hs.HostSend(req))
+	got, err := hs.HostResult(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, hostlink.Response{GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", Payload: req.Payload}, got)
+
+	// A second invocation on the same VM is never served. The agent may
+	// already have closed the connection, so a failed send is a refusal too.
+	second := req
+	second.RequestID = "r-2"
+	sendErr := hs.HostSend(second)
+	res := wait(t, done, "agent to stop after one invocation")
+	assert.Equal(t, exitClosed, res.code)
+	assert.NoError(t, res.err)
+	if sendErr == nil {
+		_, err = hs.HostResult(ctx)
+		assert.Error(t, err, "agent answered a second invocation")
+	}
+	assert.Equal(t, 1, b.n)
+}
+
+func TestTestlinkShutdownBeforeInvocationNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	conn := h.guest(t)
+	defer conn.Close()
+	require.NoError(t, hostlink.NewTestStream(conn).HostShutdown())
+	res := wait(t, done, "agent to stop")
+	assert.Equal(t, exitClosed, res.code)
+	assert.NoError(t, res.err)
+	assert.Zero(t, b.n)
+}
+
+// The guest's handshake completes, but the host pins another generation's
+// guest; the refusal arrives on first read and the bootstrap never starts.
+func TestTestlinkHostRefusalNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	other, err := guestconn.NewGeneration(scope, time.Hour)
+	require.NoError(t, err)
+	entries := h.entries()
+	wrong := other.GuestEntries(h.ln.Addr().String())
+	entries[lambdav1.EntryGuestKey], entries[lambdav1.EntryGuestCert] = wrong[lambdav1.EntryGuestKey], wrong[lambdav1.EntryGuestCert]
+	var b bootstrapStarts
+
+	res := wait(t, runTestlinkAsync(ctx, b.testlink(newGuest(t, entries))), "agent to stop")
+	assert.Equal(t, exitLink, res.code)
+	require.Error(t, res.err)
+	assert.Contains(t, res.err.Error(), "bad certificate")
+	assert.Zero(t, b.n)
+}
+
+func TestTestlinkHostCloseBeforeFrameNeverStartsBootstrap(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	h := newHost(t)
+	h.accept(ctx)
+	var b bootstrapStarts
+	done := runTestlinkAsync(ctx, b.testlink(newGuest(t, h.entries())))
+
+	require.NoError(t, h.guest(t).Close())
+	res := wait(t, done, "agent to stop")
+	assert.Equal(t, exitLink, res.code)
+	assert.ErrorContains(t, res.err, "first frame from host")
+	assert.Zero(t, b.n)
+}
+
+// A cancelled context is reported before the buffered request is delivered,
+// and the request stays undelivered.
+func TestOneInvocationHonoursCancelledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	o := &oneInvocation{first: &hostlink.Request{RequestID: "r-1"}}
+	_, err := o.Next(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+
+	req, err := o.Next(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, "r-1", req.RequestID)
+}

spinifex/agents/lambda/agent/bootstrap.go

@@ -0,0 +1,54 @@
+//go:build linux
+
+package agent
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os/exec"
+	goruntime "runtime"
+	"syscall"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// sandboxID is the uid and gid the function's bootstrap runs as.
+const sandboxID = 993
+
+// StartBootstrap starts the function's bootstrap at path as the sandbox uid and
+// gid with no supplementary groups, with no_new_privs, and with exactly env.
+// The caller waits on the returned command.
+func StartBootstrap(ctx context.Context, path string, env []string, stdout, stderr io.Writer) (*exec.Cmd, error) {
+	cred := &syscall.Credential{Uid: sandboxID, Gid: sandboxID, Groups: []uint32{}}
+	return startBootstrap(ctx, path, env, stdout, stderr, cred)
+}
+
+func startBootstrap(ctx context.Context, path string, env []string, stdout, stderr io.Writer, cred *syscall.Credential) (*exec.Cmd, error) {
+	cmd := exec.CommandContext(ctx, path)
+	// A nil Env inherits the agent's environment; the bootstrap gets only env.
+	cmd.Env = append([]string{}, env...)
+	cmd.Stdout, cmd.Stderr = stdout, stderr
+	if cred != nil {
+		cmd.SysProcAttr = &syscall.SysProcAttr{Credential: cred}
+	}
+	cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) }
+	cmd.WaitDelay = 5 * time.Second
+
+	started := make(chan error, 1)
+	go func() {
+		// no_new_privs is per thread and the child inherits it from this one. The
+		// thread is never unlocked, so Go ends it when this goroutine returns.
+		goruntime.LockOSThread()
+		if err := unix.Prctl(unix.PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0); err != nil {
+			started <- fmt.Errorf("set no_new_privs: %w", err)
+			return
+		}
+		started <- cmd.Start()
+	}()
+	if err := <-started; err != nil {
+		return nil, err
+	}
+	return cmd, nil
+}

spinifex/agents/lambda/agent/bootstrap_test.go

@@ -0,0 +1,171 @@
+//go:build linux
+
+package agent
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	goruntime "runtime"
+	"strings"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// statusProbe writes an executable that prints its own credential lines.
+func statusProbe(t *testing.T) string {
+	t.Helper()
+	// The sandbox uid must be able to reach the probe through the test's dirs.
+	dir := t.TempDir()
+	require.NoError(t, os.Chmod(filepath.Dir(dir), 0o755))
+	require.NoError(t, os.Chmod(dir, 0o755))
+	path := filepath.Join(dir, "bootstrap")
+	script := "#!/bin/sh\ngrep -E '^(Uid|Gid|Groups|NoNewPrivs):' /proc/self/status\n"
+	require.NoError(t, os.WriteFile(path, []byte(script), 0o755))
+	return path
+}
+
+func statusLines(t *testing.T, out []byte) map[string]string {
+	t.Helper()
+	lines := map[string]string{}
+	for l := range strings.SplitSeq(strings.TrimSpace(string(out)), "\n") {
+		k, v, ok := strings.Cut(l, ":")
+		require.True(t, ok, "status line %q", l)
+		lines[k] = strings.Join(strings.Fields(v), " ")
+	}
+	return lines
+}
+
+// run starts a bootstrap with the given credential and returns its stdout.
+func run(t *testing.T, path string, env []string, sandbox bool) []byte {
+	t.Helper()
+	var out bytes.Buffer
+	start := startBootstrap
+	if sandbox {
+		start = func(ctx context.Context, p string, e []string, o, s io.Writer, _ *syscall.Credential) (*exec.Cmd, error) {
+			return StartBootstrap(ctx, p, e, o, s)
+		}
+	}
+	cmd, err := start(context.Background(), path, env, &out, os.Stderr, nil)
+	require.NoError(t, err)
+	require.NoError(t, cmd.Wait())
+	return out.Bytes()
+}
+
+// threadNoNewPrivs reads no_new_privs for the calling OS thread.
+func threadNoNewPrivs(t *testing.T) string {
+	t.Helper()
+	b, err := os.ReadFile("/proc/thread-self/status")
+	require.NoError(t, err)
+	return statusLines(t, grepLines(b, "NoNewPrivs:"))["NoNewPrivs"]
+}
+
+func grepLines(b []byte, prefix string) []byte {
+	var keep []string
+	for l := range strings.SplitSeq(string(b), "\n") {
+		if strings.HasPrefix(l, prefix) {
+			keep = append(keep, l)
+		}
+	}
+	return []byte(strings.Join(keep, "\n"))
+}
+
+func TestStartBootstrapSetsNoNewPrivsWithoutChangingTheCaller(t *testing.T) {
+	goruntime.LockOSThread()
+	defer goruntime.UnlockOSThread()
+	if threadNoNewPrivs(t) != "0" {
+		t.Skip("test thread already has no_new_privs; the child would inherit it")
+	}
+
+	out := run(t, statusProbe(t), nil, false)
+	assert.Equal(t, "1", statusLines(t, out)["NoNewPrivs"])
+	assert.Equal(t, "0", threadNoNewPrivs(t), "the calling thread must keep no_new_privs unset")
+}
+
+func TestStartBootstrapDropsToSandboxIdentity(t *testing.T) {
+	if os.Geteuid() != 0 {
+		t.Skip("dropping to the sandbox uid needs root")
+	}
+	lines := statusLines(t, run(t, statusProbe(t), nil, true))
+	assert.Equal(t, "993 993 993 993", lines["Uid"])
+	assert.Equal(t, "993 993 993 993", lines["Gid"])
+	assert.Empty(t, lines["Groups"])
+	assert.Equal(t, "1", lines["NoNewPrivs"])
+}
+
+// A tenant LD_PRELOAD constructor runs before the bootstrap's main; it must
+// already see no_new_privs, so no loader runs before the flag is set.
+func TestStartBootstrapPreloadConstructorSeesNoNewPrivs(t *testing.T) {
+	goruntime.LockOSThread()
+	defer goruntime.UnlockOSThread()
+	if threadNoNewPrivs(t) != "0" {
+		t.Skip("test thread already has no_new_privs; the check would pass vacuously")
+	}
+	cc, err := exec.LookPath("cc")
+	if err != nil {
+		t.Skip("no C compiler for the preload library")
+	}
+	bootstrap, err := exec.LookPath("true")
+	require.NoError(t, err)
+	dir := t.TempDir()
+	src := filepath.Join(dir, "preload.c")
+	lib := filepath.Join(dir, "preload.so")
+	require.NoError(t, os.WriteFile(src, []byte(preloadSource), 0o644))
+	build := exec.Command(cc, "-shared", "-fPIC", "-o", lib, src)
+	build.Stderr = os.Stderr
+	require.NoError(t, build.Run())
+
+	out := run(t, bootstrap, []string{"LD_PRELOAD=" + lib}, false)
+	assert.Equal(t, "preload NoNewPrivs: 1", strings.Join(strings.Fields(string(out)), " "))
+}
+
+const preloadSource = `#include <stdio.h>
+#include <string.h>
+
+__attribute__((constructor)) static void report(void) {
+	char line[256];
+	FILE *f = fopen("/proc/self/status", "r");
+	if (!f)
+		return;
+	while (fgets(line, sizeof line, f))
+		if (strncmp(line, "NoNewPrivs:", 11) == 0)
+			printf("preload %s", line);
+	fclose(f);
+	fflush(stdout);
+}
+`
+
+func TestStartBootstrapDoesNotInheritAgentEnvironment(t *testing.T) {
+	t.Setenv("ZZ_AGENT_SECRET", "must-not-leak")
+	probe, err := exec.LookPath("env")
+	require.NoError(t, err)
+
+	for name, env := range map[string][]string{"nil": nil, "empty": {}, "given": {"ONLY=this"}} {
+		out := run(t, probe, env, false)
+		// Never print the captured environment: an inherited one may hold secrets.
+		if strings.TrimSuffix(string(out), "\n") != strings.Join(env, "\n") {
+			t.Errorf("%s: bootstrap environment is not exactly the one given", name)
+		}
+	}
+}
+
+// Without root the sandbox credential cannot be applied, so the exported entry
+// point must refuse rather than run the bootstrap as the caller.
+func TestStartBootstrapAsNonRootRefusesInsteadOfKeepingCallerIdentity(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root can apply the sandbox credential")
+	}
+	_, err := StartBootstrap(context.Background(), statusProbe(t), nil, nil, nil)
+	assert.ErrorIs(t, err, syscall.EPERM)
+}
+
+func TestStartBootstrapMissingExecutableFails(t *testing.T) {
+	_, err := startBootstrap(context.Background(), filepath.Join(t.TempDir(), "bootstrap"), nil, nil, nil, nil)
+	assert.Error(t, err)
+}