review: batch-end quality round

boot/init/src/boot.rs

@@ -15,6 +15,32 @@ const POLL_INTERVAL: Duration = Duration::from_millis(2);
 // fallback, not stall rootfs handoff for the full disk budget (10s).
 const NIC_TIMEOUT: Duration = Duration::from_millis(200);
 
+/// Cumulative µs checkpoints since sandbox-init start.
+struct Marks {
+    start: Instant,
+    points: Vec<(&'static str, u128)>,
+}
+
+impl Marks {
+    fn new() -> Self {
+        Marks {
+            start: Instant::now(),
+            points: Vec::new(),
+        }
+    }
+
+    fn mark(&mut self, label: &'static str) {
+        self.points.push((label, self.start.elapsed().as_micros()));
+    }
+
+    fn render(&self) -> String {
+        self.points
+            .iter()
+            .map(|(label, us)| format!(" {label}@{us}us"))
+            .collect()
+    }
+}
+
 pub fn run() -> ! {
     // Best-effort: if devtmpfs fails there is no console either; later
     // failures then power off silently, which is still the right end state.
@@ -59,32 +85,6 @@ pub fn run() -> ! {
     sys::fatal(&format!("exec {}: {err}", cfg.init), cfg.debug)
 }
 
-/// Cumulative µs checkpoints since sandbox-init start.
-struct Marks {
-    start: Instant,
-    points: Vec<(&'static str, u128)>,
-}
-
-impl Marks {
-    fn new() -> Self {
-        Marks {
-            start: Instant::now(),
-            points: Vec::new(),
-        }
-    }
-
-    fn mark(&mut self, label: &'static str) {
-        self.points.push((label, self.start.elapsed().as_micros()));
-    }
-
-    fn render(&self) -> String {
-        self.points
-            .iter()
-            .map(|(label, us)| format!(" {label}@{us}us"))
-            .collect()
-    }
-}
-
 fn assemble(cfg: &BootCfg, marks: &mut Marks) -> Result<(), String> {
     let ids: Vec<&str> = cfg
         .layers

docs/silkd.md

@@ -30,7 +30,7 @@ a frame only one side can parse fails CI.
 
 | group | ops | response flow |
 |---|---|---|
-| exec | `exec {argv, cwd?, env?, user?, detach?, session?}` | `started{pid}` → `stdout/stderr{data}`… → `exit{code}`; the client may stream `stdin{data}` / `stdin_close`. `detach` returns after `started`; the process keeps a bounded output ring for later `logs`/`attach` |
+| exec | `exec {argv, cwd?, env?, user?, detach?, session?}` | `started{pid}` → `stdout/stderr{data}`… → `exit{code}`; the client may stream `stdin{data}` / `stdin_close`. `detach` returns after `started`; the process keeps a bounded output ring for later `logs`/`attach`; not combinable with `session` |
 | procs | `ps` / `kill {pid, signal?}` / `attach {pid}` / `logs {pid}` | handles are guest pids; any connection can list, signal, replay, or re-attach live |
 | sessions | `session_create {id?, cwd?, env?}` / `session_list` / `session_rm {id}` | a session is a real persistent bash; `exec` with `session` runs inside it. Idle sessions are reaped after 30 minutes |
 | fs | `fs_write {path, mode?}` (+`data`/`data_end` frames) / `fs_read` / `fs_list` / `fs_stat` / `fs_mkdir {parents?}` / `fs_rm {recursive?}` / `fs_rename {from, to}` | streaming both directions; write commits atomically via temp+rename and inherits an overwritten file's mode; `fs_list` streams 4096-entry batches |

e2e/cmd/egresssmoke/main.go

@@ -103,7 +103,7 @@ func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string
 	if !strings.Contains(seen, wantToken) {
 		return fmt.Errorf("origin %s did not echo injected token %q: %q", echo, wantToken, strings.TrimSpace(seen))
 	}
-	fmt.Printf("  allowed origin reached; injected credential observed host-side\n")
+	fmt.Println("  allowed origin reached; injected credential observed host-side")
 
 	if out, _ := sb.Exec(ctx, "sh", "-c", "env; cat /proc/1/environ 2>/dev/null | tr '\\0' '\\n'"); wantToken != "" && strings.Contains(out, wantToken) {
 		return fmt.Errorf("secret value leaked into the guest")

mcp/e2e.py

@@ -44,7 +44,7 @@ def close(self):
         self.proc.wait(timeout=10)
 
 
-def main():
+def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--bin", required=True)
     parser.add_argument("--addr", default="127.0.0.1:7777")

mcp/tools.go

@@ -83,7 +83,6 @@ type tool struct {
 	handler     func(context.Context, *server, json.RawMessage) (string, error)
 }
 
-// toolSpecs renders the tools/list payload from the single table.
 func toolSpecs() []map[string]any {
 	specs := make([]map[string]any, len(tools))
 	for i, t := range tools {

sandboxd/engine/portconn.go

@@ -70,6 +70,22 @@ func (g *guestPortConn) Read(p []byte) (int, error) {
 	return n, nil
 }
 
+func (g *guestPortConn) Write(p []byte) (int, error) {
+	written := 0
+	for len(p) > 0 {
+		n := min(len(p), portWriteChunk)
+		// The hot relay path renders into one reused buffer instead of
+		// allocating two frame-sized slices per chunk.
+		g.wbuf = wire.AppendBulkRequest(g.wbuf, "data", p[:n])
+		if _, err := g.Conn.Write(g.wbuf); err != nil {
+			return written, err
+		}
+		p = p[n:]
+		written += n
+	}
+	return written, nil
+}
+
 // fastPortData slices the canonical data frame's base64 out without a JSON
 // parse — json.Unmarshal otherwise dominates the download relay, and the
 // SDK's fastBulk sets the contract: only the exact canonical shape takes the
@@ -90,19 +106,3 @@ func fastPortData(line []byte) ([]byte, bool) {
 	}
 	return out[:n], true
 }
-
-func (g *guestPortConn) Write(p []byte) (int, error) {
-	written := 0
-	for len(p) > 0 {
-		n := min(len(p), portWriteChunk)
-		// The hot relay path renders into one reused buffer instead of
-		// allocating two frame-sized slices per chunk.
-		g.wbuf = wire.AppendBulkRequest(g.wbuf, "data", p[:n])
-		if _, err := g.Conn.Write(g.wbuf); err != nil {
-			return written, err
-		}
-		p = p[n:]
-		written += n
-	}
-	return written, nil
-}

sandboxd/mesh/mesh.go

@@ -48,9 +48,6 @@ type Mesh struct {
 	mu   sync.Mutex
 	self NodeState
 	view map[string]NodeState // node_id → latest known state (includes self)
-
-	epochMu      sync.Mutex
-	epochWritten uint64 // highest epoch on disk; guarded by epochMu
 }
 
 // New starts a mesh member listening per cfg. selfAddr is the data-plane
@@ -245,19 +242,9 @@ func (m *Mesh) Shutdown() error {
 	return m.ml.Shutdown()
 }
 
-// persistEpoch durably records the epoch, serialized so a slower concurrent
-// UpdateSelf's lower value cannot overwrite a higher one already on disk.
+// persistEpoch durably records the epoch.
 func (m *Mesh) persistEpoch(epoch uint64) error {
-	m.epochMu.Lock()
-	defer m.epochMu.Unlock()
-	if epoch <= m.epochWritten {
-		return nil
-	}
-	if err := storeEpoch(m.epochPath, epoch); err != nil {
-		return err
-	}
-	m.epochWritten = epoch
-	return nil
+	return storeEpoch(m.epochPath, epoch)
 }
 
 // forget drops a departed node from the placement view so redirects stop

sandboxd/mesh/state_test.go

@@ -84,29 +84,6 @@ func TestUpdateSelfPersistsEpoch(t *testing.T) {
 	}
 }
 
-func TestPersistEpochMonotonic(t *testing.T) {
-	dir := t.TempDir()
-	m := newBoundMesh(t, dir)
-	base := m.self.Epoch // New already seeded a wall-clock floor
-	high := base + 1000
-	if err := m.persistEpoch(high); err != nil {
-		t.Fatalf("persist high: %v", err)
-	}
-	// Concurrent lower values must never regress the on-disk floor below high.
-	var wg sync.WaitGroup
-	for e := base + 1; e < high; e++ {
-		wg.Go(func() {
-			if err := m.persistEpoch(e); err != nil {
-				t.Errorf("persist %d: %v", e, err)
-			}
-		})
-	}
-	wg.Wait()
-	if got := loadEpoch(filepath.Join(dir, "mesh-epoch")); got != high {
-		t.Errorf("persisted epoch %d, want the %d floor held against lower concurrent writes", got, high)
-	}
-}
-
 func TestUpdateSelfConcurrentDropsNothing(t *testing.T) {
 	m := newBoundMesh(t, t.TempDir())
 	base := m.self.Epoch

sandboxd/server/preview.go

@@ -190,8 +190,13 @@ func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	ttl := time.Duration(req.TTLSeconds) * time.Second
-	if lease := time.Until(deadline); ttl <= 0 || ttl > lease {
-		ttl = lease // never outlive the claim
+	switch {
+	case !deadline.IsZero():
+		if lease := time.Until(deadline); ttl <= 0 || ttl > lease {
+			ttl = lease // never outlive the claim
+		}
+	case ttl <= 0:
+		ttl = previewTTL
 	}
 	writeJSON(w, http.StatusOK, types.PreviewResponse{URL: s.preview.Mint(id, req.Port, ttl)})
 }

sandboxd/server/server.go

@@ -27,6 +27,7 @@ import (
 const (
 	upgradeProto = "silkd"
 	maxBodyBytes = 1 << 20
+	previewTTL   = time.Hour
 )
 
 // poolErrHTTP maps pool sentinels to their HTTP replies; an empty msg

sandboxd/server/server_test.go

@@ -822,6 +822,41 @@ func TestOwnerEndpoint(t *testing.T) {
 	}
 }
 
+// TestPreviewHandlerZeroDeadlineMintsLiveToken covers a claim with no
+// deadline (e.g. archived with ArchiveDeleteAfterSeconds=0): the minted
+// preview token must still verify, not be stamped already-expired.
+func TestPreviewHandlerZeroDeadlineMintsLiveToken(t *testing.T) {
+	mgr := &fakeManager{claimDeadline: func(string, string) (time.Time, error) {
+		return time.Time{}, nil
+	}}
+	ps := NewPreviewServer("secret", "node:7777", &fakePreviewMgr{})
+	srv := New("", nil, "node:7777", mgr, &fakeDialer{}, nil, ps)
+	ts := httptest.NewServer(srv.Handler())
+	t.Cleanup(func() { ts.Close(); srv.CloseRelays() })
+
+	resp, err := http.Post(ts.URL+"/v1/sandboxes/sb_1/preview", "application/json",
+		strings.NewReader(`{"token":"tok","port":8080,"ttl_seconds":300}`))
+	if err != nil {
+		t.Fatalf("preview: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status %d, want 200", resp.StatusCode)
+	}
+	var out types.PreviewResponse
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	token := strings.TrimSuffix(out.URL[strings.Index(out.URL, "/p/")+3:], "/")
+	claims, ok := ps.verify(token)
+	if !ok {
+		t.Fatalf("minted token %q does not verify", token)
+	}
+	if claims.Exp <= time.Now().Unix() {
+		t.Errorf("exp %d, want in the future", claims.Exp)
+	}
+}
+
 func newTestServer(t *testing.T, apiToken string, mgr Manager, dialer Dialer) *httptest.Server {
 	t.Helper()
 	return newTenantTestServer(t, apiToken, nil, mgr, dialer)
@@ -863,6 +898,7 @@ type fakeManager struct {
 	deleteCheckpoint func(ckptID string) error
 	setPools         func(pools []config.PoolSpec) error
 	infoPools        []pool.PoolInfo
+	claimDeadline    func(id, token string) (time.Time, error)
 
 	gotTenant    string
 	tenantClaims map[string]int
@@ -938,6 +974,9 @@ func (f *fakeManager) HasGolden(context.Context, types.PoolKey) bool {
 }
 
 func (f *fakeManager) ClaimDeadline(id, token string) (time.Time, error) {
+	if f.claimDeadline != nil {
+		return f.claimDeadline(id, token)
+	}
 	if f.socket == nil {
 		return time.Now().Add(time.Hour), nil
 	}

sandboxd/store/s3/s3_test.go

@@ -9,7 +9,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -67,8 +67,8 @@ func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			}
 			result.Contents = append(result.Contents, object{Key: k, Size: len(v)})
 		}
-		sort.Slice(result.Contents, func(i, j int) bool { return result.Contents[i].Key < result.Contents[j].Key })
-		sort.Slice(result.CommonPrefixes, func(i, j int) bool { return result.CommonPrefixes[i].Prefix < result.CommonPrefixes[j].Prefix })
+		slices.SortFunc(result.Contents, func(a, b object) int { return cmp.Compare(a.Key, b.Key) })
+		slices.SortFunc(result.CommonPrefixes, func(a, b commonPrefix) int { return cmp.Compare(a.Prefix, b.Prefix) })
 		w.Header().Set("Content-Type", "application/xml")
 		_ = xml.NewEncoder(w).Encode(result)
 	case r.Method == http.MethodHead:

sdk/go/proc.go

@@ -11,12 +11,16 @@ import (
 
 // Spawn starts cmd detached: it returns the guest pid as soon as the
 // process starts, and the process keeps running with a bounded output ring
-// readable later via Logs or Attach. cmd's Stdin/Stdout/Stderr are ignored.
+// readable later via Logs or Attach. cmd's Stdin/Stdout/Stderr are ignored;
+// cmd.Session must be empty — the session exec path cannot detach.
 func (s *Sandbox) Spawn(ctx context.Context, cmd Cmd) (uint32, error) {
 	if len(cmd.Argv) == 0 {
 		return 0, fmt.Errorf("empty argv")
 	}
-	req := &wire.Exec{Argv: cmd.Argv, Cwd: cmd.Cwd, Env: cmd.Env, User: cmd.User, Detach: true, Session: cmd.Session}
+	if cmd.Session != "" {
+		return 0, fmt.Errorf("spawn does not support session")
+	}
+	req := &wire.Exec{Argv: cmd.Argv, Cwd: cmd.Cwd, Env: cmd.Env, User: cmd.User, Detach: true}
 	started, err := oneShotRPC[wire.Started](ctx, s, req)
 	if err != nil {
 		return 0, err

sdk/go/proc_test.go

@@ -32,6 +32,10 @@ func TestProcVerbs(t *testing.T) {
 		t.Fatalf("Spawn: pid %d, %v", pid, err)
 	}
 
+	if _, err = sb.Spawn(ctx, Cmd{Argv: []string{"true"}, Session: "s1"}); err == nil || !strings.Contains(err.Error(), "session") {
+		t.Fatalf("Spawn with session: want session error, got %v", err)
+	}
+
 	procs, err := sb.Ps(ctx)
 	if err != nil || len(procs) != 1 || procs[0].PID != 41 || !procs[0].Detached || procs[0].State != "running" {
 		t.Fatalf("Ps: %+v, %v", procs, err)

sdk/python/cocoonsandbox/frames.py

@@ -32,12 +32,10 @@ def encode_request(op: str, **fields) -> bytes:
 def decode_response(line: bytes) -> dict:
     """Parses one response frame; the returned dict carries its tag under
     "type" and any binary payload decoded under "data"."""
-    # Bulk frames (stdout/stderr/data) are the throughput path where json.loads
-    # of the big base64 string dominates. base64 is JSON-escape-free and these
-    # frames carry exactly {"type":...,"data":...}, so slice both out and skip
-    # json — but only for that exact shape; anything else (extra fields,
-    # trailing bytes, non-alphabet bytes in the segment) gets the full parse
-    # instead of a silently wrong slice.
+    # base64 is JSON-escape-free, so a frame shaped exactly {"type":...,
+    # "data":...} can be sliced directly, skipping json.loads; any other
+    # shape (extra fields, trailing bytes, non-alphabet bytes) falls through
+    # to the full parse instead of risking a silently wrong slice.
     if line.startswith(b'{"type":"'):
         te = line.find(b'"', 9)
         if te > 0 and line[9:te] in (b"stdout", b"stderr", b"data") and line.startswith(b'","data":"', te):

sdk/python/cocoonsandbox/sandbox.py

@@ -67,12 +67,11 @@ def run(self, argv: list[str], cwd: str = "", env: dict | None = None,
         return code
 
     def spawn(self, *argv: str, cwd: str = "", env: dict | None = None,
-              user: str = "", session: str = "") -> int:
+              user: str = "") -> int:
         """Starts argv detached, returning its pid immediately; the process
         keeps a bounded output ring readable later via logs()/attach()."""
         started = self._call("exec", "started", argv=list(argv), cwd=cwd or None,
-                             env=env, user=user or None, session=session or None,
-                             detach=True)
+                             env=env, user=user or None, detach=True)
         return started["pid"]
 
     def ps(self) -> list[dict]:

silkd/src/server.rs

@@ -67,6 +67,14 @@ impl State {
                     )
                     .await;
                 }
+                if e.detach && e.session.is_some() {
+                    return proto::error_frame(
+                        &mut writer,
+                        ErrorKind::BadRequest,
+                        "detach is not supported with session",
+                    )
+                    .await;
+                }
                 if let Some(sid) = e.session.as_deref() {
                     let Some(sess) = self.sessions.get(sid) else {
                         return proto::error_frame(