review: whole-repo round after #329 (style, cuts −44 prod, OpenPty takes wire.PtyOpen)

docs/sdk.md

@@ -645,7 +645,7 @@ error pointing at `Push`.
 ## Terminals
 
 ```go
-pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 120, Rows: 40})
+pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 120, Rows: 40})
 defer pty.Close()
 pty.Write([]byte("make test\n"))
 io.Copy(os.Stdout, pty)                   // EOF when the shell exits

e2e/cmd/smoke/main.go

@@ -417,7 +417,7 @@ func smokePromote(ctx context.Context, client *sandbox.Client, sb *sandbox.Sandb
 }
 
 func smokePty(ctx context.Context, sb *sandbox.Sandbox) error {
-	pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 80, Rows: 24})
+	pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 80, Rows: 24})
 	if err != nil {
 		return err
 	}

protocol/wire/frame.go

@@ -143,9 +143,11 @@ type Response interface{ RespType() string }
 
 type respDecoder func([]byte) (Response, error)
 
+type bulkMaker func([]byte) Response
+
 type bulkResponse struct {
 	head []byte
-	mk   func([]byte) Response
+	mk   bulkMaker
 }
 
 // reqPtr and respPtr turn a mistyped decoder table entry into a compile error.
@@ -159,15 +161,13 @@ type respPtr[T any] interface {
 	Response
 }
 
-// Exec starts a process; with Session set it runs inside that persistent
-// shell instead. Detach is emitted even when false — it is part of the
-// fixture corpus shape.
+// Exec starts a process; with Session set it runs inside that persistent shell instead.
 type Exec struct {
 	Argv    []string          `json:"argv"`
 	Cwd     string            `json:"cwd,omitempty"`
 	Env     map[string]string `json:"env,omitempty"`
 	User    string            `json:"user,omitempty"`
-	Detach  bool              `json:"detach"`
+	Detach  bool              `json:"detach"` // emitted even when false: the fixture corpus carries it
 	Session string            `json:"session,omitempty"`
 }
 
@@ -679,12 +679,25 @@ func DecodeRequest(line []byte) (Request, error) {
 // DecodeResponse parses one frame into its type's concrete Go type. Byte
 // fields are freshly allocated per frame, so callers may retain them.
 func DecodeResponse(line []byte) (Response, error) {
-	return decodeResponse(line, nil)
+	return DecodeResponseInto(line, nil)
 }
 
 // DecodeResponseInto is DecodeResponse for a streaming consumer: a bulk frame's bytes land in *scratch, grown as needed, and stay valid only until the next call with it.
 func DecodeResponseInto(line []byte, scratch *[]byte) (Response, error) {
-	return decodeResponse(line, scratch)
+	typ, err := frameTag(line, respTagHead, "type")
+	if err != nil {
+		return nil, fmt.Errorf("parse response frame: %w", err)
+	}
+	if bulk, ok := bulkResponses[string(typ)]; ok {
+		if data, ok := decodeBulk(line, bulk.head, scratch); ok {
+			return bulk.mk(data), nil
+		}
+	}
+	dec, ok := responseDecoders[string(typ)]
+	if !ok {
+		return nil, fmt.Errorf("unknown response type %q", typ)
+	}
+	return dec(line)
 }
 
 // NewFrameScanner wraps r for newline-delimited frames capped at MaxFrame.
@@ -716,7 +729,7 @@ func IsContinuation(line []byte) bool {
 	return false
 }
 
-func bulkDecoders(mks ...func([]byte) Response) map[string]bulkResponse {
+func bulkDecoders(mks ...bulkMaker) map[string]bulkResponse {
 	out := make(map[string]bulkResponse, len(mks))
 	for _, mk := range mks {
 		tag := mk(nil).RespType()
@@ -725,23 +738,6 @@ func bulkDecoders(mks ...func([]byte) Response) map[string]bulkResponse {
 	return out
 }
 
-func decodeResponse(line []byte, scratch *[]byte) (Response, error) {
-	typ, err := frameTag(line, respTagHead, "type")
-	if err != nil {
-		return nil, fmt.Errorf("parse response frame: %w", err)
-	}
-	if bulk, ok := bulkResponses[string(typ)]; ok {
-		if data, ok := decodeBulk(line, bulk.head, scratch); ok {
-			return bulk.mk(data), nil
-		}
-	}
-	dec, ok := responseDecoders[string(typ)]
-	if !ok {
-		return nil, fmt.Errorf("unknown response type %q", typ)
-	}
-	return dec(line)
-}
-
 // decodeBulk decodes a canonical bulk frame's payload into scratch, or a fresh buffer when scratch is nil; ok is false for any other shape.
 func decodeBulk(line, head []byte, scratch *[]byte) ([]byte, bool) {
 	after, ok := bytes.CutPrefix(line, head)
@@ -826,26 +822,18 @@ func scanTag(line []byte, key string) (string, error) {
 	return "", nil
 }
 
-func decodeAs[T any](line []byte) (*T, error) {
-	v := new(T)
+func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) {
+	v := PT(new(T))
 	if err := json.Unmarshal(line, v); err != nil {
 		return nil, err
 	}
 	return v, nil
 }
 
-func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) {
-	v, err := decodeAs[T](line)
-	if err != nil {
-		return nil, err
-	}
-	return PT(v), nil
-}
-
 func decodeResp[T any, PT respPtr[T]](line []byte) (Response, error) {
-	v, err := decodeAs[T](line)
-	if err != nil {
+	v := PT(new(T))
+	if err := json.Unmarshal(line, v); err != nil {
 		return nil, err
 	}
-	return PT(v), nil
+	return v, nil
 }

sandboxd/engine/engine.go

@@ -101,7 +101,7 @@ func (e *Engine) VersionWarning(ctx context.Context) (version, warning string) {
 	if err != nil {
 		return "", fmt.Sprintf("cannot determine cocoon version (need >= %s): %v", RequiredCocoon, err)
 	}
-	if below, comparable := belowFloor(v); comparable && below {
+	if below, known := belowFloor(v); known && below {
 		return v, fmt.Sprintf("cocoon %s is below the required %s — upgrade cocoon", v, RequiredCocoon)
 	}
 	return v, ""
@@ -493,8 +493,8 @@ func respFail(resp wire.Response) string {
 	return "unexpected frame " + resp.RespType()
 }
 
-// belowFloor reports whether v is below RequiredCocoon; comparable is false for a dev build.
-func belowFloor(v string) (below, comparable bool) {
+// belowFloor reports whether v is below RequiredCocoon; known is false for a dev build.
+func belowFloor(v string) (below, known bool) {
 	v = "v" + strings.TrimPrefix(v, "v")
 	if semver.Canonical(v) != v {
 		return false, false

sandboxd/pool/archive.go

@@ -33,10 +33,6 @@ func (m *Manager) archiveDeleteFor(key types.PoolKey) time.Duration {
 	return m.archiveDeleteDefault
 }
 
-func (m *Manager) archiveEnabledFor(key types.PoolKey) bool {
-	return m.archiveAfterFor(key) > 0
-}
-
 func (m *Manager) archiveOnce(ctx context.Context) {
 	if !m.archiveSweep.CompareAndSwap(false, true) {
 		return // the previous sweep's archives are still draining

sandboxd/pool/archive_test.go

@@ -562,7 +562,7 @@ func TestArchiveRemovalCommitPinsCheckpoint(t *testing.T) {
 			if !ckExists(t, m, ck) {
 				t.Error("retry deleted the checkpoint before the claims commit")
 			}
-			m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json")
+			breakStore(t, m)
 			m.store.writeMu.Unlock()
 			locked = false
 			if err := <-done; action == testArchiveRelease && err == nil {
@@ -667,7 +667,7 @@ func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) {
 		defer m.recLocksMu.Unlock()
 		return m.recRefs[ck] >= 2
 	})
-	m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json")
+	breakStore(t, m)
 	m.store.writeMu.Unlock()
 	locked = false
 	if err := <-woke; err == nil {

sandboxd/pool/claim.go

@@ -502,7 +502,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 	}
 	m.mu.Lock()
 	var expired []victim
-	var dropped []string
+	var dropped, purges []string
 	for id, sb := range m.claimed {
 		// a zero deadline means no expiry (an archived claim kept forever).
 		if sb.Deadline.IsZero() || !now.After(sb.Deadline) {
@@ -511,6 +511,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 		switch {
 		case sb.ArchiveCk != "":
 			expired = append(expired, victim{action: reapPurge, id: id, ck: sb.ArchiveCk, tenant: sb.Tenant, sb: sb})
+			purges = append(purges, sb.ArchiveCk)
 		case m.archivesAtDeadline(sb):
 			// archive instead of destroy, kept in m.claimed for archive() to move
 			if _, busy := m.archiving[id]; busy {
@@ -538,19 +539,13 @@ func (m *Manager) reapOnce(ctx context.Context) {
 	}
 
 	logger := log.WithFunc("pool.reapOnce")
-	var purges []string
-	for _, v := range expired {
-		if v.action == reapPurge {
-			purges = append(purges, v.ck)
-		}
-	}
 	failed := m.markArchiveCks(purges)
 	expired = slices.DeleteFunc(expired, func(v victim) bool {
 		markErr, bad := failed[v.ck]
-		if bad && v.action == reapPurge {
+		if bad {
 			logger.Errorf(ctx, markErr, "mark archive ck %s; keeping %s", v.ck, v.id)
 		}
-		return bad && v.action == reapPurge
+		return bad
 	})
 	if len(expired) == 0 {
 		return
@@ -629,7 +624,7 @@ func (m *Manager) reapOnce(ctx context.Context) {
 }
 
 func (m *Manager) archivesAtDeadline(sb *types.Sandbox) bool {
-	return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveEnabledFor(sb.Key))
+	return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveAfterFor(sb.Key) > 0)
 }
 
 func (m *Manager) pauseExpired(ctx context.Context, sb *types.Sandbox) error {

sandboxd/pool/liveness_test.go

@@ -524,7 +524,6 @@ func TestReconcileKeepsRunningClaimsTheHostLost(t *testing.T) {
 	}
 }
 
-// assertRestartFailsUntilBudget runs recoveries against a guest setup step that always fails.
 func assertRestartFailsUntilBudget(t *testing.T, m *Manager, eng *fakeEngine, sb *types.Sandbox) {
 	t.Helper()
 	for attempt := range vmmRestartBudget {

sandboxd/pool/pool_test.go

@@ -426,14 +426,7 @@ func TestRefillServicesEveryPoolUnderTightBudget(t *testing.T) {
 			infos, _ := m.Info()
 			return len(infos) == 2 && infos[0].Warm == 3 && infos[1].Warm == 3
 		}
-		deadline := time.Now().Add(3 * time.Second)
-		for !both() && time.Now().Before(deadline) {
-			time.Sleep(5 * time.Millisecond)
-		}
-		if !both() {
-			infos, _ := m.Info()
-			t.Fatalf("one refill tick did not chain both pools to target: %+v", infos)
-		}
+		waitFor(t, both)
 	})
 }
 

sandboxd/server/relay.go

@@ -34,6 +34,11 @@ var (
 	silkdEnds = relayEnds{guest: func(c net.Conn) { _ = c.Close() }, client: drainClient}
 )
 
+// relayEnds is what a finished direction means to the peer that did not end it.
+type relayEnds struct {
+	guest, client func(net.Conn)
+}
+
 // CloseRelays force-closes in-flight relays; http.Server.Shutdown does not track hijacked conns.
 func (s *Server) CloseRelays() {
 	s.relayMu.Lock()
@@ -94,12 +99,6 @@ func (s *Server) relay(ctx context.Context, id string, client net.Conn, clientBu
 	s.splice(client, clientR, guest, switchingProtocols, silkdEnds)
 }
 
-// relayEnds is what a finished direction means to the peer that did not end it.
-type relayEnds struct {
-	guest  func(net.Conn)
-	client func(net.Conn)
-}
-
 func (s *Server) splice(client net.Conn, clientR io.Reader, guest net.Conn, hello []byte, ends relayEnds) {
 	release, ok := s.trackRelay(client, guest)
 	if !ok {
@@ -157,33 +156,6 @@ func (s *Server) trackRelay(client, guest net.Conn) (func(), bool) {
 	}, true
 }
 
-func keepAlive(client net.Conn) {
-	tcp, ok := client.(*net.TCPConn)
-	if !ok {
-		return
-	}
-	_ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{
-		Enable:   true,
-		Idle:     keepAliveIdle,
-		Interval: keepAliveInterval,
-		Count:    keepAliveCount,
-	})
-}
-
-// drainClient half-closes; the read deadline is what unblocks the splice goroutine.
-func drainClient(client net.Conn) {
-	utils.CloseWrite(client)
-	_ = client.SetReadDeadline(time.Now().Add(drainGrace))
-}
-
-// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads.
-func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader {
-	if n := clientBuf.Buffered(); n > 0 {
-		return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client)
-	}
-	return client
-}
-
 // auditTee records each request line it relays; input frames and the bytes past the cap pass unrecorded.
 type auditTee struct {
 	r      io.Reader
@@ -230,3 +202,30 @@ func (t *auditTee) endLine() {
 	t.buf = t.buf[:0]
 	t.skip = false
 }
+
+func keepAlive(client net.Conn) {
+	tcp, ok := client.(*net.TCPConn)
+	if !ok {
+		return
+	}
+	_ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{
+		Enable:   true,
+		Idle:     keepAliveIdle,
+		Interval: keepAliveInterval,
+		Count:    keepAliveCount,
+	})
+}
+
+// drainClient half-closes; the read deadline is what unblocks the splice goroutine.
+func drainClient(client net.Conn) {
+	utils.CloseWrite(client)
+	_ = client.SetReadDeadline(time.Now().Add(drainGrace))
+}
+
+// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads.
+func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader {
+	if n := clientBuf.Buffered(); n > 0 {
+		return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client)
+	}
+	return client
+}

sdk/go/files.go

@@ -8,23 +8,18 @@ import (
 	"github.com/cocoonstack/sandbox/protocol/wire"
 )
 
-// WriteFile writes data to path in the sandbox, atomically (silkd renames a
-// temp file into place). mode, when non-nil, sets the file's permission bits.
+// WriteFile atomically writes data to path in the sandbox; a non-nil mode sets the file's permission bits.
 func (s *Sandbox) WriteFile(ctx context.Context, path string, data []byte, mode *uint32) error {
 	return s.uploadRPC(ctx, &wire.FsWrite{Path: path, Mode: mode}, bytes.NewReader(data))
 }
 
 // ReadFile returns the contents of path.
 func (s *Sandbox) ReadFile(ctx context.Context, path string) ([]byte, error) {
-	var out []byte
-	err := s.downloadRPC(ctx, &wire.FsRead{Path: path}, func(b []byte) error {
-		out = append(out, b...)
-		return nil
-	})
-	if err != nil {
+	var buf bytes.Buffer
+	if err := s.ReadFileTo(ctx, path, &buf); err != nil {
 		return nil, err
 	}
-	return out, nil
+	return buf.Bytes(), nil
 }
 
 // ReadFileTo streams path into w; an error from w ends the read there.