review: ledger round — sdk/go, sandboxd, e2e, silkd, python (4 fixes, 9 reviews)

.github/workflows/python.yml

@@ -6,13 +6,17 @@ on:
       - "sdk/python/**"
       - "sdk/openai/**"
       - "sdk/langchain/**"
+      - "mcp/**/*.py"
+      - "ruff.toml"
       - "protocol/**"
       - ".github/workflows/python.yml"
   pull_request:
     paths:
       - "sdk/python/**"
       - "sdk/openai/**"
       - "sdk/langchain/**"
+      - "mcp/**/*.py"
+      - "ruff.toml"
       - "protocol/**"
 
 jobs:
@@ -24,9 +28,9 @@ jobs:
         with:
           python-version: "3.12"
       - name: Install tooling
-        run: pip install ruff pytest && pip install -e sdk/python -e sdk/openai -e sdk/langchain
+        run: pip install ruff==0.15.20 pytest==8.4.2 && pip install -e sdk/python -e sdk/openai -e sdk/langchain
       - name: Lint
-        run: ruff check sdk/python && ruff check sdk/openai && ruff check sdk/langchain
+        run: ruff format --check . && ruff check .
       - name: Test cocoonsandbox
         run: cd sdk/python && pytest -q
       - name: Test openai adapter

boot/init/src/boot.rs

@@ -122,7 +122,7 @@ fn assemble(cfg: &BootCfg, marks: &mut Marks) -> Result<(), String> {
     if let Some(hostname) = &cfg.hostname {
         sys::sethostname(hostname)?;
     }
-    // Empty machine-id => systemd generates a fresh one per VM (clone identity).
+    // empty machine-id => systemd generates a fresh one per VM (clone identity).
     let _ = fs::write(format!("{NEWROOT}/etc/machine-id"), "");
 
     persist_network(cfg);
@@ -139,7 +139,7 @@ fn assemble(cfg: &BootCfg, marks: &mut Marks) -> Result<(), String> {
     Ok(())
 }
 
-/// Persists kernel ip= params as MAC-matched networkd units in the new root; a missing NIC degrades to the DHCP fallback.
+/// Persists kernel ip= params as MAC-matched networkd units in the new root.
 fn persist_network(cfg: &BootCfg) {
     if cfg.ips.is_empty() {
         return;
@@ -228,7 +228,7 @@ fn scan_serials(ids: &[&str], found: &mut [Option<String>]) {
         if !name.starts_with("vd") {
             continue;
         }
-        // The serial attribute location varies by kernel version.
+        // the serial attribute location varies by kernel version.
         let paths = [
             format!("/sys/block/{name}/serial"),
             format!("/sys/block/{name}/device/serial"),

boot/init/src/cfg.rs

@@ -59,7 +59,7 @@ pub fn parse(cmdline: &str) -> Result<BootCfg, String> {
                     .collect();
             }
             "cocoon.cow" => cfg.cow = val.to_string(),
-            // A junk value keeps the default, matching the old initramfs hook.
+            // a junk value keeps the default, matching the old initramfs hook.
             "cocoon.timeout" => {
                 if let Ok(secs) = val.parse::<u64>() {
                     cfg.timeout = Duration::from_secs(secs.min(MAX_TIMEOUT_SECS));
@@ -155,7 +155,7 @@ fn parse_ip_param(val: &str) -> Option<IpParam> {
 fn mask_to_prefix(mask: &str) -> Option<u8> {
     let bits = u32::from(mask.parse::<std::net::Ipv4Addr>().ok()?);
     let prefix = bits.leading_ones();
-    // Reject non-contiguous masks.
+    // reject non-contiguous masks.
     (bits == u32::MAX.checked_shl(32 - prefix).unwrap_or(0)).then_some(prefix as u8)
 }
 

docs/langchain.md

@@ -19,7 +19,7 @@ schemas, sync-native with `asyncio.to_thread` async bridges):
 
 | tool | what it does |
 |---|---|
-| `sandbox_exec` | run a shell command, cut off after 5 minutes; stdout/stderr/exit code; disk state persists across calls |
+| `sandbox_exec` | run a shell command, cut off after 5 minutes without output; stdout/stderr/exit code; disk state persists across calls |
 | `sandbox_write_file` | write a text file (atomic on the guest) |
 | `sandbox_read_file` | read a text file |
 | `sandbox_list_dir` | list a directory as JSON |

e2e/cmd/crossnode/main.go

@@ -55,6 +55,7 @@ func run(addrA, addrB, token, template string) error {
 	if err != nil {
 		return fmt.Errorf("checkpoint on A: %w", err)
 	}
+	defer func() { _ = ck.Delete(context.WithoutCancel(ctx)) }()
 	fmt.Printf("  A: checkpoint %s published to the store in %.1fs\n", ck.ID, time.Since(t0).Seconds())
 
 	cb, err := sandbox.Connect(addrB, sandbox.WithAPIToken(token))

e2e/cmd/egresssmoke/main.go

@@ -14,6 +14,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -49,6 +50,9 @@ func main() {
 }
 
 func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string, guarded bool) error {
+	if guarded && wantToken == "" {
+		return errors.New("-secret is required for the injection check")
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 

e2e/cmd/interceptsmoke/main.go

@@ -7,6 +7,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -34,6 +35,9 @@ func main() {
 }
 
 func run(addr, token, template, echo, secret, issuer string) error {
+	if secret == "" {
+		return errors.New("-secret is required for the injection check")
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 

e2e/cmd/rpcbench/main.go

@@ -141,6 +141,8 @@ func dialAgent(ctx context.Context, addr, id, token string) (net.Conn, error) {
 	if err != nil {
 		return nil, err
 	}
+	stop := context.AfterFunc(ctx, func() { _ = raw.Close() })
+	defer stop()
 	req, err := http.NewRequest(http.MethodGet, "http://"+addr+"/v1/sandboxes/"+id+"/agent", nil)
 	if err != nil {
 		_ = raw.Close()

e2e/cmd/smoke/main.go

@@ -14,6 +14,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"net/textproto"
 	"os"
 	"slices"
 	"strconv"
@@ -274,8 +275,11 @@ func smokeGit(ctx context.Context, sb *sandbox.Sandbox) error {
 	if err = sb.GitCheckout(ctx, "/work", "feature"); err != nil {
 		return err
 	}
-	if br, err = sb.GitBranches(ctx, "/work"); err != nil || br.Current != "feature" {
-		return fmt.Errorf("after checkout: current=%q err=%v", br.Current, err)
+	if br, err = sb.GitBranches(ctx, "/work"); err != nil {
+		return fmt.Errorf("branches after checkout: %w", err)
+	}
+	if br.Current != "feature" {
+		return fmt.Errorf("after checkout: current=%q, want feature", br.Current)
 	}
 
 	// This sandbox is on the no-network lane: push must fail with the typed
@@ -706,25 +710,15 @@ func lspWrite(w io.Writer, body string) error {
 // id arrives (server-initiated requests also carry ids but have a method;
 // notifications have no id — both are skipped), returning its result.
 func lspReadResponse(r *bufio.Reader, id int) (json.RawMessage, error) {
+	tp := textproto.NewReader(r)
 	for {
-		length := 0
-		for {
-			line, err := r.ReadString('\n')
-			if err != nil {
-				return nil, err
-			}
-			line = strings.TrimRight(line, "\r\n")
-			if line == "" {
-				break
-			}
-			if v, ok := strings.CutPrefix(line, "Content-Length: "); ok {
-				if length, err = strconv.Atoi(v); err != nil {
-					return nil, fmt.Errorf("content-length %q: %w", v, err)
-				}
-			}
+		hdr, err := tp.ReadMIMEHeader()
+		if err != nil {
+			return nil, err
 		}
-		if length <= 0 {
-			return nil, fmt.Errorf("lsp frame without content-length")
+		length, err := strconv.Atoi(hdr.Get("Content-Length"))
+		if err != nil || length <= 0 {
+			return nil, fmt.Errorf("lsp frame content-length %q", hdr.Get("Content-Length"))
 		}
 		body := make([]byte, length)
 		if _, err := io.ReadFull(r, body); err != nil {

mcp/e2e.py

@@ -5,20 +5,26 @@
     python3 e2e.py --bin ./sandbox-mcp --addr 127.0.0.1:7777 --token e2e --template rt2:24.04
 """
 
+from __future__ import annotations
+
 import argparse
 import json
 import subprocess
 import sys
 
 
 class McpClient:
-    def __init__(self, argv):
-        self.proc = subprocess.Popen(
-            argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE
-        )
+    def __init__(self, argv: list[str]) -> None:
+        self.proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
         self.seq = 0
 
-    def call(self, method, params=None):
+    def __enter__(self) -> McpClient:
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.close()
+
+    def call(self, method: str, params: dict | None = None) -> dict:
         self.seq += 1
         req = {"jsonrpc": "2.0", "id": self.seq, "method": method}
         if params is not None:
@@ -31,7 +37,7 @@ def call(self, method, params=None):
         assert "error" not in resp, resp
         return resp["result"]
 
-    def tool(self, tool_name, **arguments):
+    def tool(self, tool_name: str, **arguments: object) -> object:
         result = self.call("tools/call", {"name": tool_name, "arguments": arguments})
         text = result["content"][0]["text"]
         if result.get("isError"):
@@ -41,9 +47,14 @@ def tool(self, tool_name, **arguments):
         except ValueError:
             return text
 
-    def close(self):
+    def close(self) -> None:
         self.proc.stdin.close()
-        self.proc.wait(timeout=10)
+        self.proc.stdout.close()
+        try:
+            self.proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+            self.proc.wait()
 
 
 def main() -> int:
@@ -54,18 +65,11 @@ def main() -> int:
     parser.add_argument("--template", default="rt:24.04")
     args = parser.parse_args()
 
-    mcp = McpClient(
-        [args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template]
-    )
-    try:
-        init = mcp.call(
-            "initialize", {"protocolVersion": "2024-11-05", "capabilities": {}}
-        )
+    with McpClient([args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template]) as mcp:
+        init = mcp.call("initialize", {"protocolVersion": "2024-11-05", "capabilities": {}})
         assert init["serverInfo"]["name"] == "sandbox-mcp", init
         tools = {t["name"] for t in mcp.call("tools/list")["tools"]}
-        assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, (
-            tools
-        )
+        assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, tools
         print(f"  initialize + tools/list ok ({len(tools)} tools)")
 
         sandbox_id = mcp.tool("create_sandbox")["sandbox_id"]
@@ -75,15 +79,11 @@ def main() -> int:
 
         mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v1")
         assert mcp.tool("read_file", sandbox_id=sandbox_id, path="/root/m.txt") == "v1"
-        names = {
-            e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root")
-        }
+        names = {e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root")}
         assert "m.txt" in names, names
         print("  files ok")
 
-        ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")[
-            "checkpoint_id"
-        ]
+        ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")["checkpoint_id"]
         mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v2")
         branch = mcp.tool("branch_checkpoint", checkpoint_id=ckpt)["sandbox_id"]
         assert mcp.tool("read_file", sandbox_id=branch, path="/root/m.txt") == "v1"
@@ -109,8 +109,6 @@ def main() -> int:
         info = mcp.tool("node_info")
         assert "pools" in info, info
         print("  cleanup + node_info ok")
-    finally:
-        mcp.close()
     print("MCP-E2E PASS")
     return 0
 

mcp/tools.go

@@ -438,11 +438,7 @@ func jsonText(v any) string {
 type props map[string]map[string]any
 
 func str(description string) map[string]any {
-	p := map[string]any{"type": "string"}
-	if description != "" {
-		p["description"] = description
-	}
-	return p
+	return map[string]any{"type": "string", "description": description}
 }
 
 func integer(description string) map[string]any {

protocol/wire/frame.go

@@ -22,6 +22,10 @@ const (
 	ProtoVersion = 1
 	// MaxFrame mirrors silkd's frame cap.
 	MaxFrame = 8 << 20
+	// BulkChunk mirrors silkd's BULK_CHUNK, the payload of one bulk data frame.
+	BulkChunk = 256 * 1024
+	// PortWriteChunk keeps a port data frame (payload x4/3 base64 plus envelope) well under MaxFrame.
+	PortWriteChunk = 1 << 20
 
 	// GitBranch.Action values (silkd's GitBranchOp).
 	BranchList     = "list"

ruff.toml

@@ -0,0 +1,5 @@
+line-length = 120
+target-version = "py310"
+
+[lint]
+select = ["E", "F", "W", "I", "UP", "B", "SIM"]

sandboxd/engine/installca.go

@@ -41,7 +41,7 @@ func (e *Engine) silkdWriteFile(ctx context.Context, vsockSocket, path string, m
 		if serr := s.send(wire.FsWrite{Path: path, Mode: &mode}); serr != nil {
 			return serr
 		}
-		for chunk := range slices.Chunk(data, silkdChunk) {
+		for chunk := range slices.Chunk(data, wire.BulkChunk) {
 			if serr := s.send(wire.Data{Data: chunk}); serr != nil {
 				return serr
 			}

sandboxd/engine/portconn.go

@@ -13,9 +13,6 @@ import (
 )
 
 const (
-	// portWriteChunk keeps each data frame well under silkd's 8MiB frame cap.
-	portWriteChunk = 1 << 20
-
 	// portReadBuf fits silkd's data frames in one buffered read.
 	portReadBuf = 64 << 10
 )
@@ -69,7 +66,7 @@ func (g *guestPortConn) Read(p []byte) (int, error) {
 func (g *guestPortConn) Write(p []byte) (int, error) {
 	written := 0
 	for len(p) > 0 {
-		n := min(len(p), portWriteChunk)
+		n := min(len(p), wire.PortWriteChunk)
 		// the hot relay path reuses one buffer instead of allocating per chunk.
 		g.wbuf = wire.AppendBulkRequest(g.wbuf, "data", p[:n])
 		if _, err := g.Conn.Write(g.wbuf); err != nil {

sandboxd/engine/silkd.go

@@ -10,8 +10,6 @@ import (
 	"github.com/cocoonstack/sandbox/protocol/wire"
 )
 
-const silkdChunk = 256 * 1024
-
 // silkdSession is a dialed silkd conn bound to a ctx, with wire-typed request/reply helpers.
 type silkdSession struct {
 	conn net.Conn

sandboxd/pool/hibernate.go

@@ -247,7 +247,7 @@ func (m *Manager) commitTransition(ctx context.Context, sb *types.Sandbox, snap,
 // syncClaims flushes a lagging journal so a hibernate retry cannot report a false success.
 func (m *Manager) syncClaims(ctx context.Context, sb *types.Sandbox) error {
 	if !m.store.synced() {
-		if err := m.store.commit(m.claimsSnapshot()); err != nil {
+		if err := m.store.commit(m.store.mark()); err != nil {
 			return err
 		}
 	}