From 8679ef337f749f17c2e77981a8a4d8aafd736e2d Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:53:06 +0900 Subject: [PATCH 01/14] fix: checkpoint claims reject WithClaimRef instead of dropping it Checkpoint.New applied every option into a full claim request but only the TTL reached the wire, so WithClaimRef was accepted and silently lost while WithVolumes, WithNetwork and WithSize were rejected. The claim_ref is what the k8s-side reverse lookup keys on; a silent drop leaves the sandbox unfindable. Reject it locally like the other inapplicable options; the server-side checkpoint claim carries no claim_ref field. --- sdk/go/checkpoint.go | 4 ++-- sdk/go/checkpoint_test.go | 11 +++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index cf452213..7cd534ae 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -32,8 +32,8 @@ func (ck *Checkpoint) New(ctx context.Context, opts ...Option) (*Sandbox, error) for _, opt := range opts { opt(&claim) } - if len(claim.Volumes) > 0 { - return nil, errors.New("checkpoint claims do not accept WithVolumes") + if len(claim.Volumes) > 0 || claim.ClaimRef != "" { + return nil, errors.New("checkpoint claims do not accept WithVolumes or WithClaimRef") } if err := claim.rejectPinnedAxes(); err != nil { return nil, err diff --git a/sdk/go/checkpoint_test.go b/sdk/go/checkpoint_test.go index 705ef5cf..493c24a3 100644 --- a/sdk/go/checkpoint_test.go +++ b/sdk/go/checkpoint_test.go @@ -45,6 +45,17 @@ func TestCheckpointNewFollowsRedirect(t *testing.T) { } } +func TestCheckpointNewRejectsClaimRefLocally(t *testing.T) { + ck := &Checkpoint{} + sb, err := ck.New(t.Context(), WithClaimRef("ns/claim")) + if err == nil || !strings.Contains(err.Error(), "WithClaimRef") { + t.Errorf("err %v, want local WithClaimRef rejection", err) + } + if sb != nil { + t.Errorf("sandbox %+v, want nil", sb) + } +} + func TestCheckpointNewRedirectAllCandidatesFail(t *testing.T) { broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) From c45922f609962474879d04766917f75f8d622c7a Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:54:46 +0900 Subject: [PATCH 02/14] =?UTF-8?q?review:=20sdk/go=20=E2=80=94=20one=20stdi?= =?UTF-8?q?o=20pump,=20template=20claims=20through=20claimFollow,=20stdlib?= =?UTF-8?q?=20folds?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit drainProc and Run carried the same recv/switch loop; pumpStdio holds the union of the arms and Run keeps rejecting a bare done frame. Template.New had a hand-rolled no-volumes fast path that duplicated claimFollow's first contact with no_redirect pinned; the pin now rides the encoder closure, so a non-compliant redirect is rejected instead of yielding an empty handle. SetPoolsCluster dedupes with slices.Compact over a sorted copy, StartLsp uses oneShotRPC like every other one-reply verb, the peers forwarder folds into its one caller, and dialAgent drops the socket deadline that duplicated the AfterFunc close on the same ctx. --- sdk/go/client.go | 3 ++- sdk/go/info.go | 9 +-------- sdk/go/lsp.go | 7 +------ sdk/go/pools.go | 11 ++--------- sdk/go/proc.go | 30 +----------------------------- sdk/go/sandbox.go | 32 +++++++------------------------- sdk/go/template.go | 15 ++------------- sdk/go/upgrade.go | 5 ----- sdk/go/utils.go | 32 ++++++++++++++++++++++++++++++++ 9 files changed, 48 insertions(+), 96 deletions(-) diff --git a/sdk/go/client.go b/sdk/go/client.go index 89823837..8bbecd6e 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -107,7 +107,8 @@ func (c *Client) Lookup(ctx context.Context, id, token string) (*Sandbox, error) if owner, err := c.ownerAt(ctx, c.addr, id, token); err == nil { return &Sandbox{ID: id, token: token, c: c, owner: owner}, nil } - owner, ok := scatter(ctx, c.peers(ctx), func(ctx context.Context, addr string) (string, error) { + addrs, _ := c.peersOrErr(ctx) + owner, ok := scatter(ctx, addrs, func(ctx context.Context, addr string) (string, error) { return c.ownerAt(ctx, addr, id, token) }) if !ok { diff --git a/sdk/go/info.go b/sdk/go/info.go index c0f44ff1..ba23c3c2 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -6,8 +6,7 @@ import ( "time" ) -// peersTimeout bounds the best-effort peer discovery inside Lookup so one -// slow entry node cannot stall the scatter. +// peersTimeout bounds peer discovery so one slow entry node cannot stall the caller. const peersTimeout = 5 * time.Second // NodeInfo is one node's operational state from GET /v1/info. @@ -69,12 +68,6 @@ func (c *Client) Sandboxes(ctx context.Context) ([]SandboxSummary, error) { return reply.Sandboxes, nil } -// peers fetches the cluster's node addresses, best-effort (nil on failure). -func (c *Client) peers(ctx context.Context) []string { - addrs, _ := c.peersOrErr(ctx) - return addrs -} - // peersOrErr fetches the node addresses, surfacing a discovery failure. It reads // /v1/peers (tenant-accessible), so it works under a tenant token. func (c *Client) peersOrErr(ctx context.Context) ([]string, error) { diff --git a/sdk/go/lsp.go b/sdk/go/lsp.go index aa9ccf1a..46d748e4 100644 --- a/sdk/go/lsp.go +++ b/sdk/go/lsp.go @@ -33,12 +33,7 @@ func (l *Lsp) Stop(ctx context.Context) error { // (rooted at root), returning a handle. On the base image, which ships no // language servers, this fails with silkd's typed not_found. func (s *Sandbox) StartLsp(ctx context.Context, language, root string) (*Lsp, error) { - conn, done, err := s.call(ctx, &wire.LspStart{Language: language, Root: root}) - if err != nil { - return nil, err - } - defer done() - started, err := expect[wire.LspStarted](ctx, conn) + started, err := oneShotRPC[wire.LspStarted](ctx, s, &wire.LspStart{Language: language, Root: root}) if err != nil { return nil, err } diff --git a/sdk/go/pools.go b/sdk/go/pools.go index ad6c9ad6..3bec953d 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "net/http" + "slices" "sync" ) @@ -44,15 +45,7 @@ func (c *Client) SetPools(ctx context.Context, pools []PoolSpec) (*NodeInfo, err // was reached — an incomplete apply to retry, not a single-node cluster (nil). func (c *Client) SetPoolsCluster(ctx context.Context, pools []PoolSpec) ([]PoolResult, error) { peers, peersErr := c.peersOrErr(ctx) - seen := map[string]struct{}{} - var addrs []string - for _, a := range append([]string{c.addr}, peers...) { - if _, dup := seen[a]; dup { - continue - } - seen[a] = struct{}{} - addrs = append(addrs, a) - } + addrs := slices.Compact(slices.Sorted(slices.Values(append([]string{c.addr}, peers...)))) results := make([]PoolResult, len(addrs)) var wg sync.WaitGroup for i, addr := range addrs { diff --git a/sdk/go/proc.go b/sdk/go/proc.go index 28610d51..433ac01f 100644 --- a/sdk/go/proc.go +++ b/sdk/go/proc.go @@ -1,7 +1,6 @@ package sandbox import ( - "cmp" "context" "fmt" "io" @@ -63,38 +62,11 @@ func (s *Sandbox) Attach(ctx context.Context, pid uint32, stdout, stderr io.Writ return s.drainProc(ctx, &wire.Attach{PID: pid}, stdout, stderr) } -// drainProc pumps stdout/stderr frames until the terminal frame: exit -// carries the code, done means the stream ended without one. func (s *Sandbox) drainProc(ctx context.Context, req wire.Request, stdout, stderr io.Writer) (int32, bool, error) { conn, done, err := s.call(ctx, req) if err != nil { return 0, false, err } defer done() - stdout = cmp.Or(stdout, io.Discard) - stderr = cmp.Or(stderr, io.Discard) - for { - resp, err := recv(ctx, conn) - if err != nil { - return 0, false, err - } - switch resp := resp.(type) { - case *wire.Stdout: - if _, err := stdout.Write(resp.Data); err != nil { - return 0, false, err - } - case *wire.Stderr: - if _, err := stderr.Write(resp.Data); err != nil { - return 0, false, err - } - case *wire.Exit: - return resp.Code, true, nil - case *wire.Done: - return 0, false, nil - case *wire.ErrorResp: - return 0, false, resp - default: - return 0, false, unexpected(resp) - } - } + return pumpStdio(ctx, conn, stdout, stderr) } diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index ddf5418d..fa971548 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -2,7 +2,6 @@ package sandbox import ( "bytes" - "cmp" "context" "fmt" "io" @@ -117,31 +116,14 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { go pumpStdin(conn, cmd.Stdin) } - stdout := cmp.Or(cmd.Stdout, io.Discard) - stderr := cmp.Or(cmd.Stderr, io.Discard) - for { - resp, err := recv(ctx, conn) - if err != nil { - return 0, err - } - switch resp := resp.(type) { - case *wire.Started: - case *wire.Stdout: - if _, err := stdout.Write(resp.Data); err != nil { - return 0, err - } - case *wire.Stderr: - if _, err := stderr.Write(resp.Data); err != nil { - return 0, err - } - case *wire.Exit: - return int(resp.Code), nil - case *wire.ErrorResp: - return 0, resp - default: - return 0, unexpected(resp) - } + code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) + if err != nil { + return 0, err + } + if !exited { + return 0, unexpected(&wire.Done{}) } + return int(code), nil } // Fork clones the sandbox into count children — memory, disk, and guest diff --git a/sdk/go/template.go b/sdk/go/template.go index 3e8a58db..2bb2b786 100644 --- a/sdk/go/template.go +++ b/sdk/go/template.go @@ -33,20 +33,9 @@ func (t *Template) New(ctx context.Context, opts ...Option) (*Sandbox, error) { return nil, err } claim.Net, claim.Size = t.net, t.size - if len(claim.Volumes) == 0 { - claim.NoRedirect = true - body, err := encodeBody("claim", claim) - if err != nil { - return nil, err - } - cr, err := t.c.claimAt(ctx, t.addr, body) - if err != nil { - return nil, err - } - return t.c.handleFrom(t.addr, cr), nil - } + noVolumes := len(claim.Volumes) == 0 addr, cr, err := claimFollow(t.addr, "claim", func(noRedirect, requirePromoted bool) ([]byte, error) { - claim.NoRedirect, claim.RequirePromoted = noRedirect, requirePromoted + claim.NoRedirect, claim.RequirePromoted = noRedirect || noVolumes, requirePromoted return encodeBody("claim", claim) }, func(addr string, body []byte) (claimResponse, error) { return t.c.claimAt(ctx, addr, body) diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 05398617..c78d7f3a 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -7,7 +7,6 @@ import ( "net" "net/http" "strings" - "time" ) // dialAgent opens the data-plane connection to the owner node: a raw TCP dial @@ -21,9 +20,6 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Con } stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) defer stop() - if deadline, ok := ctx.Deadline(); ok { - _ = raw.SetDeadline(deadline) - } // id/token interpolate into the raw request; CR/LF would inject headers. if strings.ContainsAny(id, "\r\n\x00") || strings.ContainsAny(token, "\r\n\x00") { @@ -52,7 +48,6 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Con _ = raw.Close() return nil, err } - _ = raw.SetDeadline(time.Time{}) return &upgradedConn{Conn: raw, r: br}, nil } diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 00b943d0..42a60545 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -1,6 +1,7 @@ package sandbox import ( + "cmp" "context" "fmt" "io" @@ -52,6 +53,37 @@ func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink func([ return drainData(ctx, conn, sink) } +// pumpStdio copies stdout/stderr frames to the writers until the terminal frame: exit carries the code, done means the stream ended without one. +func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) (code int32, exited bool, err error) { + stdout = cmp.Or(stdout, io.Discard) + stderr = cmp.Or(stderr, io.Discard) + for { + resp, err := recv(ctx, conn) + if err != nil { + return 0, false, err + } + switch resp := resp.(type) { + case *wire.Started: + case *wire.Stdout: + if _, err := stdout.Write(resp.Data); err != nil { + return 0, false, err + } + case *wire.Stderr: + if _, err := stderr.Write(resp.Data); err != nil { + return 0, false, err + } + case *wire.Exit: + return resp.Code, true, nil + case *wire.Done: + return 0, false, nil + case *wire.ErrorResp: + return 0, false, resp + default: + return 0, false, unexpected(resp) + } + } +} + // oneShotRPC sends req and returns its single typed reply frame. func oneShotRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) (*T, error) { conn, done, err := s.call(ctx, req) From 8f4c4a4698007fb1e7b070eb951d03a0ea65a3ea Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:55:59 +0900 Subject: [PATCH 03/14] =?UTF-8?q?review:=20sandboxd/mcp=20=E2=80=94=20one?= =?UTF-8?q?=20sweep-flag=20derivation,=20inlined=20accessors,=20dead=20gua?= =?UTF-8?q?rd?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The idle/archive sweep switches were derived three times (NewManager from the config specs, adoptPersistedPools from pools.json, SetPools from the live pool set); recomputeSweepFlags now derives them once from m.pools, after adoption and after every SetPools. poolWarmup's locked read moves into its only caller, claimsSnapshot forwarded to store.mark at both call sites, retryRemovals' empty-batch return duplicated what runBounded(0) already does, Info iterates a sorted maps.Values instead of building a filtered slice first, and mcp's str dropped a description guard no call site can reach (all 16 pass a literal). --- mcp/tools.go | 6 +----- sandboxd/pool/hibernate.go | 2 +- sandboxd/pool/pool.go | 34 +++++++++++++++------------------- sandboxd/pool/poolstore.go | 8 -------- sandboxd/pool/refill.go | 10 +++------- sandboxd/pool/remove.go | 3 --- sandboxd/pool/setpools.go | 12 +----------- 7 files changed, 21 insertions(+), 54 deletions(-) diff --git a/mcp/tools.go b/mcp/tools.go index 9bc168fd..8666d272 100644 --- a/mcp/tools.go +++ b/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 { diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index cae92656..1a256fb6 100644 --- a/sandboxd/pool/hibernate.go +++ b/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 } } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 5d45793b..1f1b654a 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "net" "os" "path/filepath" @@ -428,20 +429,12 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg } } m.idleDefault = time.Duration(cfg.IdleHibernateSeconds) * time.Second - m.idleEnabled = m.idleDefault > 0 m.archiveAfterDefault = time.Duration(cfg.ArchiveAfterSeconds) * time.Second m.archiveDeleteDefault = time.Duration(cfg.ArchiveDeleteAfterSeconds) * time.Second - m.archiveEnabled = m.archiveAfterDefault > 0 for _, spec := range cfg.Pools { p := newPool(spec.PoolKey) p.applySpec(spec) m.pools[spec.PoolKey] = p - if spec.IdleHibernateSeconds > 0 { - m.idleEnabled = true - } - if spec.ArchiveAfterSeconds > 0 { - m.archiveEnabled = true - } if spec.Egress != nil { m.poolEgress[spec.PoolKey] = spec.Egress m.guardedEgress = true @@ -461,6 +454,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg if err := m.adoptPersistedPools(ctx); err != nil { return nil, err } + m.recomputeSweepFlags() return m, nil } @@ -510,7 +504,7 @@ func (m *Manager) Run(ctx context.Context) { // FlushClaims synchronously persists the current claim set at shutdown. func (m *Manager) FlushClaims() error { - return m.store.commit(m.claimsSnapshot()) + return m.store.commit(m.store.mark()) } // SetTemplateNotifier wires the immediate-republish hook; call it before serving starts. @@ -523,15 +517,11 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) { m.mu.Lock() defer m.mu.Unlock() now := time.Now() - live := make([]*pool, 0, len(m.pools)) - for _, p := range m.pools { - if !p.removed { - live = append(live, p) + pools := make([]PoolInfo, 0, len(m.pools)) + for _, p := range slices.SortedFunc(maps.Values(m.pools), func(a, b *pool) int { return strings.Compare(a.hash, b.hash) }) { + if p.removed { + continue } - } - slices.SortFunc(live, func(a, b *pool) int { return strings.Compare(a.hash, b.hash) }) - pools := make([]PoolInfo, 0, len(live)) - for _, p := range live { pools = append(pools, PoolInfo{ Key: p.key, Warm: len(p.warm), @@ -606,8 +596,14 @@ func (m *Manager) sweepStoreGenerations(ctx context.Context) { } } -func (m *Manager) claimsSnapshot() claimSnapshot { - return m.store.mark() +// recomputeSweepFlags derives the sweep switches from the live pool set rather than latching them: removing every idle pool turns the sweep off again. +func (m *Manager) recomputeSweepFlags() { + m.idleEnabled = m.idleDefault > 0 + m.archiveEnabled = m.archiveAfterDefault > 0 + for _, p := range m.pools { + m.idleEnabled = m.idleEnabled || p.idle > 0 + m.archiveEnabled = m.archiveEnabled || p.archiveAfter > 0 + } } func (m *Manager) untrack(set map[string]struct{}, key string) { diff --git a/sandboxd/pool/poolstore.go b/sandboxd/pool/poolstore.go index c8e642dc..40c86546 100644 --- a/sandboxd/pool/poolstore.go +++ b/sandboxd/pool/poolstore.go @@ -87,8 +87,6 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { logger.Warnf(ctx, "config.json pools differ from the API-applied set and are overridden; delete %s to return to config-owned pools", m.poolStore.path) } clear(m.pools) - m.idleEnabled = m.idleDefault > 0 - m.archiveEnabled = m.archiveAfterDefault > 0 for _, spec := range pf.Pools { spec = normalizePoolSpec(spec) if err := m.validate(spec.PoolKey); err != nil { @@ -101,12 +99,6 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { p.applySpec(spec) m.adoptGolden(p) m.pools[spec.PoolKey] = p - if spec.IdleHibernateSeconds > 0 { - m.idleEnabled = true - } - if spec.ArchiveAfterSeconds > 0 { - m.archiveEnabled = true - } } logger.Infof(ctx, "restored %d API-applied pools from pools.json", len(pf.Pools)) return nil diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index e84b0a03..c794e614 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -213,14 +213,10 @@ func (m *Manager) buildGoldenSteps(ctx context.Context, key types.PoolKey, name, return writeGoldenSidecar(final+warmupSidecarSuffix, warmupStamp(warmup)) } -func (m *Manager) poolWarmup(key types.PoolKey) []string { - m.mu.Lock() - defer m.mu.Unlock() - return m.poolWarmups[key] -} - func (m *Manager) runWarmup(ctx context.Context, key types.PoolKey, sock string) ([]string, error) { - warmup := m.poolWarmup(key) + m.mu.Lock() + warmup := m.poolWarmups[key] + m.mu.Unlock() if len(warmup) == 0 { return nil, nil } diff --git a/sandboxd/pool/remove.go b/sandboxd/pool/remove.go index 8404ea56..c3887656 100644 --- a/sandboxd/pool/remove.go +++ b/sandboxd/pool/remove.go @@ -77,9 +77,6 @@ func (m *Manager) retryRemovals(ctx context.Context) *sync.WaitGroup { } } m.mu.Unlock() - if len(batch) == 0 { - return new(sync.WaitGroup) - } names := slices.Collect(maps.Keys(batch)) return m.runBounded(ctx, len(names), func(ctx context.Context, i int) { m.retryRemoval(ctx, names[i], batch[names[i]]) diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index e327bca7..a45437be 100644 --- a/sandboxd/pool/setpools.go +++ b/sandboxd/pool/setpools.go @@ -59,17 +59,7 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { m.adoptGolden(p) m.pools[key] = p } - // recompute rather than latch: removing every idle pool turns the sweep off again - m.idleEnabled = m.idleDefault > 0 - m.archiveEnabled = m.archiveAfterDefault > 0 - for _, p := range m.pools { - if p.idle > 0 { - m.idleEnabled = true - } - if p.archiveAfter > 0 { - m.archiveEnabled = true - } - } + m.recomputeSweepFlags() m.mu.Unlock() runCtx := context.WithoutCancel(ctx) From 75c79c9f6683f24c22c4186ad60e9bd330ca8713 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:56:32 +0900 Subject: [PATCH 04/14] =?UTF-8?q?fix:=20e2e=20drivers=20=E2=80=94=20deferr?= =?UTF-8?q?ed=20checkpoint=20delete,=20required=20-secret,=20ctx-bound=20u?= =?UTF-8?q?pgrade,=20nil-safe=20branch=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit crossnode published a checkpoint to the shared store and deleted it only on the success path; every node-B failure left it behind. egresssmoke and interceptsmoke defaulted -secret to the empty string, and strings.Contains(seen, "") made the injection assertion vacuously true, so a hand run printed PASS without proving injection; both now require the flag (egresssmoke only in the guarded mode, the negative control never reaches the check). rpcbench's copy of the SDK's upgrade dial never wired ctx to the socket, so a node that accepts TCP and never answers the upgrade blocked ReadResponse past the 5-minute budget. smoke's post-checkout branch assertion read br.Current in the error branch where br is nil. --- e2e/cmd/crossnode/main.go | 1 + e2e/cmd/egresssmoke/main.go | 4 ++++ e2e/cmd/interceptsmoke/main.go | 4 ++++ e2e/cmd/rpcbench/main.go | 2 ++ e2e/cmd/smoke/main.go | 7 +++++-- 5 files changed, 16 insertions(+), 2 deletions(-) diff --git a/e2e/cmd/crossnode/main.go b/e2e/cmd/crossnode/main.go index 93131cfe..cfe4680b 100644 --- a/e2e/cmd/crossnode/main.go +++ b/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)) diff --git a/e2e/cmd/egresssmoke/main.go b/e2e/cmd/egresssmoke/main.go index 00046888..c59f41e7 100644 --- a/e2e/cmd/egresssmoke/main.go +++ b/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() diff --git a/e2e/cmd/interceptsmoke/main.go b/e2e/cmd/interceptsmoke/main.go index 5f59f934..46f4a405 100644 --- a/e2e/cmd/interceptsmoke/main.go +++ b/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() diff --git a/e2e/cmd/rpcbench/main.go b/e2e/cmd/rpcbench/main.go index a5c1d18d..ae313cc0 100644 --- a/e2e/cmd/rpcbench/main.go +++ b/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() diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index 4ff229b6..dcf06b70 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -274,8 +274,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 From f7228ffc5955f32666bf1154864397f1571316d4 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:57:00 +0900 Subject: [PATCH 05/14] review: smoke reads LSP headers with net/textproto The LSP framing is MIME-shaped (Key: Value lines terminated by a blank line), which textproto.Reader.ReadMIMEHeader already parses; the hand loop that trimmed CRLF and cut the Content-Length prefix goes. The length <= 0 guard stays so make cannot see a negative size. --- e2e/cmd/smoke/main.go | 25 ++++++++----------------- 1 file changed, 8 insertions(+), 17 deletions(-) diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index dcf06b70..9e9a309a 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -14,6 +14,7 @@ import ( "flag" "fmt" "io" + "net/textproto" "os" "slices" "strconv" @@ -709,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 { From 64bbd51dc724b767cd3e3fc63ccfbf57c913c341 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 00:59:21 +0900 Subject: [PATCH 06/14] review: hoist the silkd framing constants into protocol/wire fsChunk, readChunk and silkdChunk were three Go copies of silkd's BULK_CHUNK, and portWriteChunk was declared twice; wire already owns MaxFrame and every module imports it, so BulkChunk and PortWriteChunk live there with one comment each. The next sdk/go release cuts a wire tag alongside as the go.mod note requires. --- protocol/wire/frame.go | 4 ++++ sandboxd/engine/installca.go | 2 +- sandboxd/engine/portconn.go | 5 +---- sandboxd/engine/silkd.go | 2 -- sdk/go/port.go | 6 +----- sdk/go/silkd/silkdtest/fake.go | 5 +---- sdk/go/utils.go | 5 +---- 7 files changed, 9 insertions(+), 20 deletions(-) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 74ba6233..474235bf 100644 --- a/protocol/wire/frame.go +++ b/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" diff --git a/sandboxd/engine/installca.go b/sandboxd/engine/installca.go index e0d8876a..9749319a 100644 --- a/sandboxd/engine/installca.go +++ b/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 } diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 16176a95..5458c356 100644 --- a/sandboxd/engine/portconn.go +++ b/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 { diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index c9afd760..cbd06f77 100644 --- a/sandboxd/engine/silkd.go +++ b/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 diff --git a/sdk/go/port.go b/sdk/go/port.go index b0ffd909..81f233c5 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -14,10 +14,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// portWriteChunk keeps a data frame (payload ×4/3 base64 + envelope) well -// under wire.MaxFrame. -const portWriteChunk = 1 << 20 - var _ net.Conn = (*PortConn)(nil) // PortConn is a net.Conn to a TCP port inside the sandbox, relayed over the @@ -39,7 +35,7 @@ func (p *PortConn) Read(b []byte) (int, error) { return p.out.Read(b) } func (p *PortConn) Write(b []byte) (int, error) { sent := 0 for len(b) > 0 { - chunk := b[:min(len(b), portWriteChunk)] + chunk := b[:min(len(b), wire.PortWriteChunk)] if err := p.conn.Send(&wire.Data{Data: chunk}); err != nil { return sent, err } diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index 0560852a..9f5cf0ae 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -19,9 +19,6 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -// readChunk mirrors silkd's BULK_CHUNK so downloads exercise real framing. -const readChunk = 256 * 1024 - // Fake is a stateful silkd fake backing the fs verbs with a real directory // and tracking sessions, so an SDK write-then-read round-trips through it. // exec/info reuse the stateless handlers. It exists for host-side unit tests; @@ -474,7 +471,7 @@ func drainUpload(r *bufio.Reader) ([]byte, error) { func sendChunked(conn net.Conn, data []byte) { for len(data) > 0 { - n := min(readChunk, len(data)) + n := min(wire.BulkChunk, len(data)) send(conn, &wire.DataResp{Data: data[:n]}) data = data[n:] } diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 42a60545..84de3509 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -11,9 +11,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// fsChunk matches silkd's BULK_CHUNK. -const fsChunk = 256 * 1024 - // respPtr is a frame type's pointer form, so a non-Response T fails to compile. type respPtr[T any] interface { *T @@ -143,7 +140,7 @@ func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.R // uploadStream chunks r into Data frames terminated by DataEnd; shared by the // FsWrite payload and the FsPush tar stream. func uploadStream(conn *silkd.Conn, r io.Reader) error { - buf := make([]byte, fsChunk) + buf := make([]byte, wire.BulkChunk) for { n, readErr := r.Read(buf) if n > 0 { From 97a3d1696b76b5e1f1243122f8e73d050f32f423 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:02:48 +0900 Subject: [PATCH 07/14] fix: silkd quotes session env keys; the LSP escape test takes the env lock session_create wrote the env value POSIX-quoted but the key raw into the init script, so a key with a quote or newline broke the script before the marker printf and converse parked forever on a read with no timeout; the half-created session was in neither the table nor the reaper, so its bash leaked until silkd restarted. Quoting the key makes bash report "not a valid identifier" and continue, so the marker still arrives. lsp_start_language_name_cannot_escape reaches manifest_dir(), which reads SILKD_LSP_DIR, without holding ENV_LOCK, racing the set_var the other tests perform under it. --- silkd/src/session.rs | 2 +- silkd/tests/lsp_e2e.rs | 1 + silkd/tests/session_e2e.rs | 18 ++++++++++++++++++ 3 files changed, 20 insertions(+), 1 deletion(-) diff --git a/silkd/src/session.rs b/silkd/src/session.rs index 3b63db03..e4b1c12c 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -79,7 +79,7 @@ impl Table { } for (k, v) in env { init.push_str("export "); - init.push_str(k); + shell_quote_into(&mut init, k); init.push('='); shell_quote_into(&mut init, v); init.push('\n'); diff --git a/silkd/tests/lsp_e2e.rs b/silkd/tests/lsp_e2e.rs index 276a538b..ba4dd77b 100644 --- a/silkd/tests/lsp_e2e.rs +++ b/silkd/tests/lsp_e2e.rs @@ -59,6 +59,7 @@ async fn lsp_start_missing_manifest_is_not_found() { #[tokio::test] async fn lsp_start_language_name_cannot_escape() { + let _env_lock = ENV_LOCK.lock().await; for bad in ["../etc/passwd", "a/b", ".."] { let frames = one( &Arc::new(State::new()), diff --git a/silkd/tests/session_e2e.rs b/silkd/tests/session_e2e.rs index c3e30b2b..624f26d9 100644 --- a/silkd/tests/session_e2e.rs +++ b/silkd/tests/session_e2e.rs @@ -66,6 +66,24 @@ async fn session_applies_cwd_and_env_at_create() { assert!(dir.path().join("marker").exists(), "cwd was not applied"); } +#[tokio::test] +async fn session_create_with_hostile_env_key_still_answers() { + let state = Arc::new(State::new()); + let created = tokio::time::timeout( + Duration::from_secs(8), + one( + &state, + &json!({"op":"session_create","env":{"BAD KEY'\n": "v", "GOOD": "ok"}}).to_string(), + ), + ) + .await + .expect("session_create hung on a hostile env key"); + assert_eq!(type_of(&created[0]), "session_created", "{created:?}"); + let id = created[0]["id"].as_str().unwrap(); + let got = sh(&state, id, &["sh", "-c", "echo $GOOD"]).await; + assert_eq!(stdout_body(&got).trim(), "ok", "{got:?}"); +} + #[tokio::test] async fn session_exit_code_propagates() { let state = Arc::new(State::new()); From 9b96a9fdf8068aed441baf78fa2090982800dccf Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:11:32 +0900 Subject: [PATCH 08/14] =?UTF-8?q?review:=20silkd/boot=20=E2=80=94=20one=20?= =?UTF-8?q?kill=20guard,=20memchr=20for=20the=20exit=20line,=20comment=20d?= =?UTF-8?q?edup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit kill_group and signal_pid each carried the same guard and a byte-identical SAFETY block; send_signal holds the one unsafe call and is_valid_pid's doc carries the pid-0 fact once. converse scanned the accumulator twice for the exit-code newline with the naive contains and position; memchr covers both, as the marker search already did. FIND_MAX_FILE is public so the e2e tests pin the bound to the real constant, and its doc no longer claims a binary check it does not make. sysutil hoists its two function- local use statements and drops the NSS_LOCK comment the static's doc and the SAFETY lines already carry; boot/init's inline comments join the lowercase register and persist_network's doc stops repeating the NIC_TIMEOUT fact. --- boot/init/src/boot.rs | 6 +++--- boot/init/src/cfg.rs | 4 ++-- silkd/src/find.rs | 4 ++-- silkd/src/session.rs | 10 +++------- silkd/src/sysutil.rs | 39 ++++++++++++++++----------------------- 5 files changed, 26 insertions(+), 37 deletions(-) diff --git a/boot/init/src/boot.rs b/boot/init/src/boot.rs index 622be227..db2df6c2 100644 --- a/boot/init/src/boot.rs +++ b/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]) { 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"), diff --git a/boot/init/src/cfg.rs b/boot/init/src/cfg.rs index a5e58fee..ad466c9f 100644 --- a/boot/init/src/cfg.rs +++ b/boot/init/src/cfg.rs @@ -59,7 +59,7 @@ pub fn parse(cmdline: &str) -> Result { .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::() { cfg.timeout = Duration::from_secs(secs.min(MAX_TIMEOUT_SECS)); @@ -155,7 +155,7 @@ fn parse_ip_param(val: &str) -> Option { fn mask_to_prefix(mask: &str) -> Option { let bits = u32::from(mask.parse::().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) } diff --git a/silkd/src/find.rs b/silkd/src/find.rs index ca054fc1..104a2039 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -12,8 +12,8 @@ use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; use crate::proto::{self, ErrorKind, Response, err_frame}; -/// Size above which find skips a file as binary or huge. -const FIND_MAX_FILE: u64 = 8 * 1024 * 1024; +/// Size above which find skips a file unread. +pub const FIND_MAX_FILE: u64 = 8 * 1024 * 1024; /// Match frames in flight between the walking thread and the writer. const MATCH_QUEUE: usize = 256; diff --git a/silkd/src/session.rs b/silkd/src/session.rs index e4b1c12c..b0f2ecfc 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -89,7 +89,7 @@ impl Table { io.converse::(&init, None).await?; } - // reserve the id atomically: a concurrent create must not orphan the loser's shell. + // reserve the id under the table lock so a concurrent create with the same id sees AlreadyExists. let mut map = sysutil::lock(&self.inner); if map.contains_key(&id) { return Err(io::Error::new( @@ -219,7 +219,7 @@ impl Io { if let Some(pos) = memchr::memmem::find(&self.acc, mb) { emit(&mut out, &mut self.frame, &self.acc[..pos]).await; self.acc.drain(..pos + mb.len()); - while !self.acc.contains(&b'\n') && self.acc.len() < EXIT_TAIL_MAX { + while memchr::memchr(b'\n', &self.acc).is_none() && self.acc.len() < EXIT_TAIL_MAX { let m = self.stdout.read(&mut self.buf).await?; if m == 0 { break; @@ -227,11 +227,7 @@ impl Io { self.acc.extend_from_slice(&self.buf[..m]); } // parse only the exit-code line: a background writer can land bytes after the newline. - let end = self - .acc - .iter() - .position(|&b| b == b'\n') - .unwrap_or(self.acc.len()); + let end = memchr::memchr(b'\n', &self.acc).unwrap_or(self.acc.len()); return Ok(String::from_utf8_lossy(&self.acc[..end]) .trim() .parse() diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index 21f20878..ca42a065 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -1,7 +1,9 @@ //! Small OS helpers; the crate's unsafe work lives here, except pty.rs's pre_exec registration. +use std::ffi::{CStr, CString}; use std::io::Read; use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd}; +use std::os::unix::process::ExitStatusExt; use std::process::ExitStatus; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{LazyLock, Mutex}; @@ -46,26 +48,18 @@ pub fn rand_token() -> String { .collect() } -/// SIGKILLs the group led by `pgid`, so a session's external command dies with its shell. +/// SIGKILLs the group led by `pgid`, so a session's external command dies with its shell; a synthetic id misses with ESRCH. pub fn kill_group(pgid: u32) { - // kill(-0) would target silkd's own group; a synthetic id passes the guard and misses with ESRCH. - if !is_valid_pid(pgid) { - return; + if is_valid_pid(pgid) { + send_signal(-(pgid as libc::pid_t), libc::SIGKILL); } - // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast - // in range and away from silkd's own group. - unsafe { libc::kill(-(pgid as libc::pid_t), libc::SIGKILL) }; } /// Sends `sig` to `pid`, ignoring ESRCH against a just-exited pid. pub fn signal_pid(pid: u32, sig: i32) { - // pid 0 means silkd's whole process group to kill(2). - if !is_valid_pid(pid) { - return; + if is_valid_pid(pid) { + send_signal(pid as libc::pid_t, sig); } - // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast - // in range and away from silkd's own group. - unsafe { libc::kill(pid as libc::pid_t, sig) }; } /// The environment every exec starts from; the proxy snapshot rides only on the no-network lane. @@ -97,11 +91,8 @@ pub fn openpty(cols: u16, rows: u16) -> std::io::Result<(OwnedFd, OwnedFd)> { ws_xpixel: 0, ws_ypixel: 0, }; - // SAFETY: openpty writes two valid fds into master/slave on success; ws is a - // fully-initialized winsize (openpty only reads it). We take ownership of - // both fds immediately. A raw `*mut` pointer for winp works across libc's - // platform-varying signature (*mut on macOS, *const on Linux) without the - // &mut that clippy flags as unnecessary on Linux. + // a raw `*mut` for winp fits both libc signatures (*mut on macOS, *const on Linux). + // SAFETY: openpty writes two valid fds into master/slave on success and only reads ws; both fds are owned immediately. let rc = unsafe { libc::openpty( &mut master, @@ -185,7 +176,6 @@ pub fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { /// Maps a wait() status to the shell convention (128 + signal when killed). pub fn exit_code(status: ExitStatus) -> i32 { - use std::os::unix::process::ExitStatusExt; if let Some(code) = status.code() { return code; } @@ -212,11 +202,16 @@ fn fill_random(b: &mut [u8]) -> bool { .is_ok() } -/// Rejects pid 0 and anything that would go negative through the pid_t cast. +/// Rejects pid 0, which kill(2) reads as silkd's own process group, and anything that would go negative through the pid_t cast. fn is_valid_pid(id: u32) -> bool { id != 0 && id <= i32::MAX as u32 } +fn send_signal(target: libc::pid_t, sig: i32) { + // SAFETY: kill(2) takes no pointers; every caller vets the target with is_valid_pid first. + unsafe { libc::kill(target, sig) }; +} + fn align_proxy_env_for(cmd: &mut Command, nic: bool) { if !nic { return; @@ -258,9 +253,7 @@ fn snapshot_proxy(get: impl Fn(&str) -> Option) -> Vec<(&'static str, St } fn lookup_user(user: &str) -> Result<(u32, u32, String), String> { - use std::ffi::CString; let cname = CString::new(user).map_err(|_| format!("invalid user {user:?}"))?; - // hold NSS_LOCK across every read of the static buffer, else a concurrent lookup clobbers it. let _guard = lock(&NSS_LOCK); // SAFETY: cname is a live NUL-terminated CString for the call. let pw = unsafe { libc::getpwnam(cname.as_ptr()) }; @@ -271,7 +264,7 @@ fn lookup_user(user: &str) -> Result<(u32, u32, String), String> { // a valid passwd whose pw_dir is a NUL-terminated string it owns. let pw = unsafe { &*pw }; // SAFETY: pw_dir is NUL-terminated and stays valid while NSS_LOCK is held. - let home = unsafe { std::ffi::CStr::from_ptr(pw.pw_dir) } + let home = unsafe { CStr::from_ptr(pw.pw_dir) } .to_string_lossy() .into_owned(); Ok((pw.pw_uid, pw.pw_gid, home)) From 8cbb1fb00beea7a888eaaefa2b70a4c0f85bc582 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:11:48 +0900 Subject: [PATCH 09/14] =?UTF-8?q?review:=20silkd=20e2e=20tests=20=E2=80=94?= =?UTF-8?q?=20shared=20send/next=5Fframe,=20parsed-frame=20assertions,=20f?= =?UTF-8?q?olded=20preambles?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit common gains send, next_frame and DEADLINE, which eight sites in lsp_e2e, exec_e2e and pty_e2e re-implemented as write-then-newline and read-then-parse pairs. forward_e2e asserted on raw JSON text with a hand-rolled substring extractor; it now parses frames like every other file and shares one forwarded() handshake. exec_e2e's four poll-until- exit loops and two ps polls become wait_for_exit and wait_listed; lsp_e2e's four identical manifest-and-start preambles become started_server; session_e2e's create takes no empty json!({}) at 13 of 14 sites and the busy-command spawn is one helper. find_e2e pins its oversize fixtures to silkd::find::FIND_MAX_FILE instead of a literal, git_e2e drops a last() wrapper the other nine files never had, pty_e2e loses an assertion its own read_until predicate made tautological, function-local use statements move to the top, and tree_e2e sorts in place instead of cloning. --- silkd/tests/common/mod.rs | 25 ++++- silkd/tests/exec_e2e.rs | 154 ++++++++++---------------- silkd/tests/find_e2e.rs | 37 ++++--- silkd/tests/forward_e2e.rs | 111 +++++++------------ silkd/tests/fs_e2e.rs | 7 +- silkd/tests/git_e2e.rs | 16 ++- silkd/tests/lsp_e2e.rs | 215 +++++++++---------------------------- silkd/tests/pty_e2e.rs | 39 +++---- silkd/tests/session_e2e.rs | 67 ++++++------ silkd/tests/tree_e2e.rs | 11 +- 10 files changed, 250 insertions(+), 432 deletions(-) diff --git a/silkd/tests/common/mod.rs b/silkd/tests/common/mod.rs index 28901d11..895e1c19 100644 --- a/silkd/tests/common/mod.rs +++ b/silkd/tests/common/mod.rs @@ -1,17 +1,20 @@ -//! Shared harness: drives silkd's server over an in-memory duplex as a relayed host connection would. -//! Each test binary compiles it separately, so #![allow(dead_code)] covers the helpers one binary skips. +//! Shared harness: drives silkd's server over an in-memory duplex as a relayed host connection would; each test binary compiles it separately, hence allow(dead_code). #![allow(clippy::unwrap_used, clippy::expect_used)] #![allow(dead_code)] use std::sync::Arc; +use std::time::Duration; use base64::Engine; -use serde_json::Value; +use serde_json::{Value, json}; use silkd::server::State; use tokio::io::{ AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, Lines, ReadHalf, WriteHalf, }; use tokio::task::JoinHandle; +use tokio::time::timeout; + +pub const DEADLINE: Duration = Duration::from_secs(30); pub type FrameWriter = WriteHalf; pub type FrameLines = Lines>>; @@ -25,6 +28,20 @@ pub fn connect(state: &Arc) -> (FrameWriter, FrameLines, JoinHandle Value { + let line = timeout(DEADLINE, lines.next_line()) + .await + .expect("deadline") + .unwrap() + .expect("stream closed"); + serde_json::from_str(&line).unwrap() +} + pub async fn roundtrip(request_line: &str) -> Vec { request_on(&Arc::new(State::new()), &[request_line.to_string()]).await } @@ -82,7 +99,7 @@ pub fn stdout_body(frames: &[Value]) -> String { pub fn data_frames(bytes: &[u8]) -> Vec { let mut lines: Vec = bytes .chunks(16 * 1024) - .map(|c| serde_json::json!({"op":"data","data":b64(c)}).to_string()) + .map(|c| json!({"op":"data","data":b64(c)}).to_string()) .collect(); lines.push(r#"{"op":"data_end"}"#.to_string()); lines diff --git a/silkd/tests/exec_e2e.rs b/silkd/tests/exec_e2e.rs index 89c04f59..d902f9ad 100644 --- a/silkd/tests/exec_e2e.rs +++ b/silkd/tests/exec_e2e.rs @@ -6,21 +6,51 @@ mod common; use std::sync::Arc; use std::time::Duration; +use serde_json::{Value, json}; use silkd::server::State; -use common::{decode, exchange, one, roundtrip, stdout_body, type_of}; +use common::{ + b64, connect, decode, exchange, next_frame, one, roundtrip, send, stdout_body, type_of, +}; async fn detached_pid(state: &Arc, script: &str) -> u64 { - let argv = serde_json::json!(["/bin/sh", "-c", script]); + let argv = json!(["/bin/sh", "-c", script]); let started = one( state, - &serde_json::json!({"op":"exec","argv":argv,"detach":true}).to_string(), + &json!({"op":"exec","argv":argv,"detach":true}).to_string(), ) .await; assert_eq!(type_of(&started[0]), "started"); started[0]["pid"].as_u64().unwrap() } +async fn wait_for_exit(state: &Arc, pid: u64) -> Vec { + for _ in 0..250 { + let logs = one(state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; + if logs.iter().any(|f| type_of(f) == "exit") { + return logs; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("pid {pid} never exited"); +} + +async fn wait_listed(state: &Arc, pid: u64) -> Value { + for _ in 0..250 { + let ps = one(state, r#"{"op":"ps"}"#).await; + if let Some(p) = ps[0]["procs"] + .as_array() + .unwrap() + .iter() + .find(|p| p["pid"].as_u64() == Some(pid)) + { + return p.clone(); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("pid {pid} never appeared in ps"); +} + #[tokio::test] async fn exec_streams_stdout_then_exit() { let frames = roundtrip(r#"{"op":"exec","argv":["/bin/echo","-n","hello"]}"#).await; @@ -133,13 +163,7 @@ async fn attach_streams_live_output_then_exit() { async fn attach_to_exited_process_returns_exit_immediately() { let state = Arc::new(State::new()); let pid = detached_pid(&state, "echo done-fast").await; - for _ in 0..100 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + wait_for_exit(&state, pid).await; let frames = tokio::time::timeout( Duration::from_secs(3), one(&state, &format!(r#"{{"op":"attach","pid":{pid}}}"#)), @@ -161,30 +185,10 @@ async fn attach_unknown_pid_is_not_found() { async fn kill_actually_terminates_a_live_process() { let state = Arc::new(State::new()); let pid = detached_pid(&state, "sleep 30").await; - for _ in 0..50 { - let ps = one(&state, r#"{"op":"ps"}"#).await; - if ps[0]["procs"] - .as_array() - .unwrap() - .iter() - .any(|p| p["pid"].as_u64() == Some(pid)) - { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + wait_listed(&state, pid).await; let killed = one(&state, &format!(r#"{{"op":"kill","pid":{pid}}}"#)).await; assert_eq!(type_of(&killed[0]), "done"); - let mut exited = false; - for _ in 0..150 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - exited = true; - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - assert!(exited, "killed process never reached exited state"); + wait_for_exit(&state, pid).await; } #[tokio::test] @@ -207,7 +211,7 @@ async fn exec_missing_binary_is_an_internal_error() { async fn exec_stdin_is_piped_to_the_child() { let frames = exchange(&[ r#"{"op":"exec","argv":["/bin/cat"]}"#.to_string(), - serde_json::json!({"op":"stdin","data":common::b64(b"piped-in\n")}).to_string(), + json!({"op":"stdin","data":b64(b"piped-in\n")}).to_string(), r#"{"op":"stdin_close"}"#.to_string(), ]) .await; @@ -218,19 +222,8 @@ async fn exec_stdin_is_piped_to_the_child() { #[tokio::test] async fn kill_of_an_exited_process_is_a_noop_success() { let state = Arc::new(State::new()); - let started = one( - &state, - r#"{"op":"exec","argv":["/bin/echo","x"],"detach":true}"#, - ) - .await; - let pid = started[0]["pid"].as_u64().unwrap(); - for _ in 0..100 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + let pid = detached_pid(&state, "echo x").await; + wait_for_exit(&state, pid).await; let killed = one(&state, &format!(r#"{{"op":"kill","pid":{pid}}}"#)).await; assert_eq!( type_of(&killed[0]), @@ -242,68 +235,33 @@ async fn kill_of_an_exited_process_is_a_noop_success() { #[tokio::test] async fn detached_exec_is_listed_then_logs_replay_output_and_exit() { let state = Arc::new(State::new()); - let started = one( - &state, - r#"{"op":"exec","argv":["/bin/echo","detached-hi"],"detach":true}"#, - ) - .await; - assert_eq!(type_of(&started[0]), "started"); - let pid = started[0]["pid"].as_u64().unwrap(); - - let mut listed = false; - for _ in 0..250 { - let ps = one(&state, r#"{"op":"ps"}"#).await; - if ps[0]["procs"] - .as_array() - .unwrap() - .iter() - .any(|p| p["pid"].as_u64() == Some(pid) && p["detached"] == true) - { - listed = true; - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - assert!(listed, "detached pid {pid} never appeared in ps"); - - for _ in 0..250 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - assert_eq!(stdout_body(&logs), "detached-hi\n"); - return; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - panic!("logs never reported an exit for pid {pid}"); + let pid = detached_pid(&state, "echo detached-hi").await; + let listed = wait_listed(&state, pid).await; + assert_eq!(listed["detached"], true, "{listed:?}"); + let logs = wait_for_exit(&state, pid).await; + assert_eq!(stdout_body(&logs), "detached-hi\n"); } #[tokio::test] async fn disconnect_during_drain_publishes_real_exit_code() { - use tokio::io::AsyncWriteExt; - let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - let req = serde_json::json!({ - "op": "exec", - "argv": ["/bin/sh", "-c", "{ sleep 0.5; echo late; sleep 30; } & exit 7"] - }) - .to_string(); - cw.write_all(req.as_bytes()).await.unwrap(); - cw.write_all(b"\n").await.unwrap(); + let (mut cw, mut lines, _) = connect(&state); + send( + &mut cw, + json!({ + "op": "exec", + "argv": ["/bin/sh", "-c", "{ sleep 0.5; echo late; sleep 30; } & exit 7"] + }), + ) + .await; - let started: serde_json::Value = - serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + let started = next_frame(&mut lines).await; assert_eq!(type_of(&started), "started"); let pid = started["pid"].as_u64().unwrap(); let st = Arc::clone(&state); - let attach = tokio::spawn(async move { - one( - &st, - &serde_json::json!({"op":"attach","pid":pid}).to_string(), - ) - .await - }); + let attach = + tokio::spawn(async move { one(&st, &json!({"op":"attach","pid":pid}).to_string()).await }); tokio::time::sleep(Duration::from_millis(200)).await; drop(lines); drop(cw); diff --git a/silkd/tests/find_e2e.rs b/silkd/tests/find_e2e.rs index 45f1a99a..6684d253 100644 --- a/silkd/tests/find_e2e.rs +++ b/silkd/tests/find_e2e.rs @@ -3,20 +3,17 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::os::unix::fs::PermissionsExt; use std::sync::Arc; use serde_json::{Value, json}; use silkd::server::State; use tokio::io::AsyncWriteExt; -use common::{exchange, type_of}; - -async fn write_file(dir: &std::path::Path, name: &str, body: &str) { - tokio::fs::write(dir.join(name), body).await.unwrap(); -} +use common::{connect, exchange, type_of}; async fn find_frames(req: Value) -> Vec { - let (mut cw, mut out, handle) = common::connect(&Arc::new(State::new())); + let (mut cw, mut out, handle) = connect(&Arc::new(State::new())); cw.write_all(format!("{req}\n").as_bytes()).await.unwrap(); let mut frames = Vec::new(); while let Some(l) = out.next_line().await.unwrap() { @@ -35,10 +32,19 @@ async fn find_frames(req: Value) -> Vec { #[tokio::test] async fn find_streams_line_matches() { let dir = tempfile::tempdir().unwrap(); - write_file(dir.path(), "a.rs", "fn main() {\n // TODO: fix\n}\n").await; - write_file(dir.path(), "b.txt", "TODO ignore me (wrong glob)\n").await; + tokio::fs::write( + dir.path().join("a.rs"), + "fn main() {\n // TODO: fix\n}\n", + ) + .await + .unwrap(); + tokio::fs::write(dir.path().join("b.txt"), "TODO ignore me (wrong glob)\n") + .await + .unwrap(); tokio::fs::create_dir(dir.path().join("sub")).await.unwrap(); - write_file(&dir.path().join("sub"), "c.rs", "// TODO nested\n").await; + tokio::fs::write(dir.path().join("sub/c.rs"), "// TODO nested\n") + .await + .unwrap(); let frames = find_frames(json!({ "op": "fs_find", @@ -63,8 +69,12 @@ async fn find_streams_line_matches() { #[tokio::test] async fn find_glob_is_a_real_glob_not_a_substring() { let dir = tempfile::tempdir().unwrap(); - write_file(dir.path(), "a.rs", "TODO\n").await; - write_file(dir.path(), "a.rs.bak", "TODO\n").await; + tokio::fs::write(dir.path().join("a.rs"), "TODO\n") + .await + .unwrap(); + tokio::fs::write(dir.path().join("a.rs.bak"), "TODO\n") + .await + .unwrap(); let frames = find_frames(json!({ "op": "fs_find", @@ -144,7 +154,7 @@ async fn replace_rewrites_and_counts() { async fn replace_skips_a_file_over_the_size_bound() { let dir = tempfile::tempdir().unwrap(); let big = dir.path().join("big.log"); - let body = format!("foo\n{}", "x".repeat(8 * 1024 * 1024)); + let body = format!("foo\n{}", "x".repeat(silkd::find::FIND_MAX_FILE as usize)); tokio::fs::write(&big, &body).await.unwrap(); let frames = exchange(&[json!({ @@ -172,7 +182,7 @@ async fn find_skips_a_file_over_the_size_bound() { let dir = tempfile::tempdir().unwrap(); tokio::fs::write( dir.path().join("big.log"), - format!("foo\n{}", "x".repeat(8 * 1024 * 1024)), + format!("foo\n{}", "x".repeat(silkd::find::FIND_MAX_FILE as usize)), ) .await .unwrap(); @@ -190,7 +200,6 @@ async fn find_skips_a_file_over_the_size_bound() { #[tokio::test] async fn replace_preserves_exec_bit() { - use std::os::unix::fs::PermissionsExt; let dir = tempfile::tempdir().unwrap(); let script = dir.path().join("run.sh"); tokio::fs::write(&script, "#!/bin/sh\nfoo\n").await.unwrap(); diff --git a/silkd/tests/forward_e2e.rs b/silkd/tests/forward_e2e.rs index 24316d7c..9226a9b7 100644 --- a/silkd/tests/forward_e2e.rs +++ b/silkd/tests/forward_e2e.rs @@ -4,16 +4,14 @@ mod common; use std::sync::Arc; -use std::time::Duration; -use base64::Engine; -use base64::engine::general_purpose::STANDARD; +use serde_json::json; use silkd::server::State; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::time::timeout; -const DEADLINE: Duration = Duration::from_secs(30); +use common::{DEADLINE, FrameLines, FrameWriter, b64, connect, decode, next_frame, send, type_of}; async fn echo_listener() -> u16 { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); @@ -38,10 +36,12 @@ async fn echo_listener() -> u16 { port } -fn data_payload(line: &str) -> Option<&str> { - let start = line.find("\"data\":\"")? + "\"data\":\"".len(); - let rest = &line[start..]; - Some(&rest[..rest.find('"')?]) +async fn forwarded(state: &Arc, port: u16) -> (FrameWriter, FrameLines) { + let (mut cw, mut lines, _) = connect(state); + send(&mut cw, json!({"v":1,"op":"port_forward","port":port})).await; + let ready = next_frame(&mut lines).await; + assert_eq!(type_of(&ready), "ready", "got {ready}"); + (cw, lines) } #[tokio::test] @@ -49,28 +49,16 @@ async fn forward_round_trips_and_done_on_server_close() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); - - cw.write_all(b"{\"v\":1,\"op\":\"data\",\"data\":\"aGk=\"}\n") - .await - .expect("send data"); - let echoed = lines.next_line().await.expect("read").expect("data"); - assert!( - echoed.contains("\"data\"") && echoed.contains("aGk="), - "got {echoed}" - ); - - cw.write_all(b"{\"v\":1,\"op\":\"data_end\"}\n") - .await - .expect("send data_end"); - let done = lines.next_line().await.expect("read").expect("done"); - assert!(done.contains("\"done\""), "got {done}"); + let (mut cw, mut lines) = forwarded(&state, port).await; + + send(&mut cw, json!({"v":1,"op":"data","data":b64(b"hi")})).await; + let echoed = next_frame(&mut lines).await; + assert_eq!(type_of(&echoed), "data", "got {echoed}"); + assert_eq!(decode(&echoed), b"hi"); + + send(&mut cw, json!({"v":1,"op":"data_end"})).await; + let done = next_frame(&mut lines).await; + assert_eq!(type_of(&done), "done", "got {done}"); }) .await .expect("test deadline"); @@ -84,34 +72,21 @@ async fn forward_bidirectional_bulk_no_deadlock() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); + let (mut cw, mut lines) = forwarded(&state, port).await; - let payload = STANDARD.encode(vec![b'x'; CHUNK]); - let frame = format!("{{\"v\":1,\"op\":\"data\",\"data\":\"{payload}\"}}\n"); + let frame = json!({"v":1,"op":"data","data":b64(&[b'x'; CHUNK])}).to_string() + "\n"; let writer = tokio::spawn(async move { for _ in 0..FRAMES { cw.write_all(frame.as_bytes()).await.expect("write frame"); } - cw.write_all(b"{\"v\":1,\"op\":\"data_end\"}\n") - .await - .expect("data_end"); + send(&mut cw, json!({"v":1,"op":"data_end"})).await; }); let mut got = 0usize; while got < CHUNK * FRAMES { - let line = lines - .next_line() - .await - .expect("read") - .expect("stream ended before all bytes echoed"); - if let Some(b64) = data_payload(&line) { - got += STANDARD.decode(b64).expect("decode").len(); + let frame = next_frame(&mut lines).await; + if type_of(&frame) == "data" { + got += decode(&frame).len(); } } writer.await.expect("writer"); @@ -125,16 +100,12 @@ async fn forward_bidirectional_bulk_no_deadlock() { async fn forward_refused_port_is_not_found() { timeout(DEADLINE, async { let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(b"{\"v\":1,\"op\":\"port_forward\",\"port\":1}\n") - .await - .expect("send request"); - let err = lines.next_line().await.expect("read").expect("error"); - assert!( - err.contains("\"error\"") && err.contains("not_found"), - "got {err}" - ); + let (mut cw, mut lines, _) = connect(&state); + + send(&mut cw, json!({"v":1,"op":"port_forward","port":1})).await; + let err = next_frame(&mut lines).await; + assert_eq!(type_of(&err), "error", "got {err}"); + assert_eq!(err["kind"], "not_found", "got {err}"); }) .await .expect("test deadline"); @@ -145,22 +116,12 @@ async fn forward_stray_frame_is_bad_request() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); - - cw.write_all(b"{\"v\":1,\"op\":\"ps\"}\n") - .await - .expect("send stray"); - let err = lines.next_line().await.expect("read").expect("error"); - assert!( - err.contains("\"error\"") && err.contains("bad_request"), - "got {err}" - ); + let (mut cw, mut lines) = forwarded(&state, port).await; + + send(&mut cw, json!({"v":1,"op":"ps"})).await; + let err = next_frame(&mut lines).await; + assert_eq!(type_of(&err), "error", "got {err}"); + assert_eq!(err["kind"], "bad_request", "got {err}"); }) .await .expect("test deadline"); diff --git a/silkd/tests/fs_e2e.rs b/silkd/tests/fs_e2e.rs index 917c5960..2f92c712 100644 --- a/silkd/tests/fs_e2e.rs +++ b/silkd/tests/fs_e2e.rs @@ -3,9 +3,11 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::os::unix::fs::PermissionsExt; + use serde_json::json; -use common::{b64, exchange, type_of}; +use common::{b64, exchange, payload, type_of}; #[tokio::test] async fn write_then_read_roundtrips_bytes() { @@ -26,7 +28,7 @@ async fn write_then_read_roundtrips_bytes() { ); let read = exchange(&[json!({"op":"fs_read","path":path}).to_string()]).await; - assert_eq!(common::payload(&read, "data"), b"silk file body"); + assert_eq!(payload(&read, "data"), b"silk file body"); assert_eq!(type_of(read.last().unwrap()), "done"); } @@ -172,7 +174,6 @@ async fn truncated_write_leaves_no_file_and_reports_error() { #[tokio::test] async fn overwrite_preserves_destination_mode() { - use std::os::unix::fs::PermissionsExt; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("script.sh"); std::fs::write(&path, b"old").unwrap(); diff --git a/silkd/tests/git_e2e.rs b/silkd/tests/git_e2e.rs index 3d63ce6d..4851573f 100644 --- a/silkd/tests/git_e2e.rs +++ b/silkd/tests/git_e2e.rs @@ -3,7 +3,7 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; -use serde_json::{Value, json}; +use serde_json::json; use common::{exchange, type_of}; @@ -29,10 +29,6 @@ fn init_repo() -> tempfile::TempDir { dir } -fn last(frames: &[Value]) -> &Value { - frames.last().unwrap() -} - #[tokio::test] async fn add_commit_status_reports_structured_state() { let repo = init_repo(); @@ -50,7 +46,7 @@ async fn add_commit_status_reports_structured_state() { ); let add = exchange(&[json!({"op":"git_add","path":p,"files":["a.txt"]}).to_string()]).await; - assert_eq!(type_of(last(&add)), "done"); + assert_eq!(type_of(add.last().unwrap()), "done"); let commit = exchange(&[json!({ "op":"git_commit","path":p,"message":"first","author":"Dev " }) @@ -78,7 +74,7 @@ async fn branch_create_list_checkout() { json!({"op":"git_branch","path":p,"action":"create","name":"feature"}).to_string(), ]) .await; - assert_eq!(type_of(last(&created)), "done"); + assert_eq!(type_of(created.last().unwrap()), "done"); let list = exchange(&[json!({"op":"git_branch","path":p,"action":"list"}).to_string()]).await; assert_eq!(type_of(&list[0]), "git_branches"); @@ -95,7 +91,7 @@ async fn branch_create_list_checkout() { json!({"op":"git_branch","path":p,"action":"checkout","name":"feature"}).to_string(), ]) .await; - assert_eq!(type_of(last(&co)), "done"); + assert_eq!(type_of(co.last().unwrap()), "done"); } #[tokio::test] @@ -135,7 +131,7 @@ async fn clone_lane_behavior() { .to_string()]) .await; silkd::net::override_egress_for_tests(None); - assert_eq!(type_of(last(&ok)), "done", "clone failed: {ok:?}"); + assert_eq!(type_of(ok.last().unwrap()), "done", "clone failed: {ok:?}"); assert!(target.join("r.txt").exists()); } @@ -183,7 +179,7 @@ async fn commit_works_without_preconfigured_identity() { std::fs::write(dir.path().join("f"), "x").unwrap(); let p = dir.path().to_str().unwrap(); let add = exchange(&[json!({"op":"git_add","path":p,"files":["f"]}).to_string()]).await; - assert_eq!(type_of(last(&add)), "done"); + assert_eq!(type_of(add.last().unwrap()), "done"); let commit = exchange(&[json!({ "op":"git_commit","path":p,"message":"m","author":"Dev " }) diff --git a/silkd/tests/lsp_e2e.rs b/silkd/tests/lsp_e2e.rs index ba4dd77b..e45184e4 100644 --- a/silkd/tests/lsp_e2e.rs +++ b/silkd/tests/lsp_e2e.rs @@ -4,8 +4,8 @@ mod common; use std::io::Write; +use std::os::unix::fs::PermissionsExt; use std::sync::Arc; -use std::time::Duration; use serde_json::json; use silkd::server::State; @@ -13,33 +13,56 @@ use tempfile::TempDir; use tokio::io::AsyncWriteExt; use tokio::time::timeout; -use common::{b64, connect, decode, one, type_of}; - -const DEADLINE: Duration = Duration::from_secs(30); - -static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +use common::{DEADLINE, b64, connect, decode, next_frame, one, send, type_of}; const FAKE_SERVER: &str = r#"#!/bin/sh IFS= read -r line printf 'reply:%s\n' "$line" "#; +static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + fn manifest_env(server_body: &str) -> TempDir { let dir = tempfile::tempdir().unwrap(); let bin = dir.path().join("fake-lsp"); let mut f = std::fs::File::create(&bin).unwrap(); f.write_all(server_body.as_bytes()).unwrap(); - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); - } + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); // SAFETY: ENV_LOCK, held by the caller for its whole test, serializes every // writer and reader of SILKD_LSP_DIR in this binary. unsafe { std::env::set_var("SILKD_LSP_DIR", dir.path()) }; dir } +async fn started_server(lang: &str, server_body: &str) -> (TempDir, Arc, String) { + let env = manifest_env(server_body); + std::fs::write( + env.path().join(lang), + env.path().join("fake-lsp").to_string_lossy().as_bytes(), + ) + .unwrap(); + let state = Arc::new(State::new()); + let start = one( + &state, + &json!({"op":"lsp_start","language":lang}).to_string(), + ) + .await; + let server_id = start.last().unwrap()["server_id"] + .as_str() + .unwrap() + .to_string(); + (env, state, server_id) +} + +async fn assert_stopped(state: &Arc, server_id: &str) { + let gone = one( + state, + &json!({"op":"lsp_stop","server_id":server_id}).to_string(), + ) + .await; + assert_eq!(gone.last().unwrap()["kind"], "not_found", "{gone:?}"); +} + #[tokio::test] async fn lsp_start_missing_manifest_is_not_found() { let _env_lock = ENV_LOCK.lock().await; @@ -77,118 +100,40 @@ async fn lsp_start_language_name_cannot_escape() { #[tokio::test] async fn lsp_broker_relays_to_the_server() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env(FAKE_SERVER); - let bin = env.path().join("fake-lsp"); - std::fs::write( - env.path().join("faketest"), - bin.to_string_lossy().as_bytes(), - ) - .unwrap(); - - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"faketest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("faketest", FAKE_SERVER).await; let (mut cw, mut out, handle) = connect(&state); - cw.write_all( - json!({"op":"lsp_request","server_id":server_id}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - let ready: serde_json::Value = - serde_json::from_str(&out.next_line().await.unwrap().unwrap()).unwrap(); - assert_eq!(type_of(&ready), "ready"); - cw.write_all( - json!({"op":"data","data":b64(b"ping\n")}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); + send(&mut cw, json!({"op":"lsp_request","server_id":server_id})).await; + assert_eq!(type_of(&next_frame(&mut out).await), "ready"); + send(&mut cw, json!({"op":"data","data":b64(b"ping\n")})).await; - let reply = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - let frame: serde_json::Value = serde_json::from_str(&reply).unwrap(); + let frame = next_frame(&mut out).await; assert_eq!(type_of(&frame), "data"); assert_eq!(decode(&frame), b"reply:ping\n"); - - let done = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - assert_eq!( - type_of(&serde_json::from_str::(&done).unwrap()), - "done" - ); + assert_eq!(type_of(&next_frame(&mut out).await), "done"); cw.shutdown().await.unwrap(); let _ = handle.await; - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!(gone.last().unwrap()["kind"], "not_found"); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_stop_kills_an_idle_server() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nsleep 60\n"); - std::fs::write( - env.path().join("idletest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"idletest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("idletest", "#!/bin/sh\nsleep 60\n").await; let stop = one( &state, &json!({"op":"lsp_stop","server_id":server_id}).to_string(), ) .await; assert_eq!(type_of(stop.last().unwrap()), "done"); - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!(gone.last().unwrap()["kind"], "not_found"); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_request_reaps_when_the_client_vanishes_before_ready() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nsleep 60\n"); - std::fs::write( - env.path().join("gonetest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"gonetest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("gonetest", "#!/bin/sh\nsleep 60\n").await; let (mut client, server) = tokio::io::duplex(1 << 20); let request = json!({"op":"lsp_request","server_id":server_id}).to_string(); @@ -201,79 +146,25 @@ async fn lsp_request_reaps_when_the_client_vanishes_before_ready() { .expect("deadline"); assert!(served.is_err(), "the ready write must fail: {served:?}"); - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!( - gone.last().unwrap()["kind"], - "not_found", - "a server whose client vanished before ready must be reaped: {gone:?}" - ); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_data_end_half_closes_stdin() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nprintf 'ate:%s\\n' $(cat | wc -c)\n"); - std::fs::write( - env.path().join("eoftest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"eoftest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = + started_server("eoftest", "#!/bin/sh\nprintf 'ate:%s\\n' $(cat | wc -c)\n").await; let (mut cw, mut out, handle) = connect(&state); - cw.write_all( - json!({"op":"lsp_request","server_id":server_id}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - let ready: serde_json::Value = - serde_json::from_str(&out.next_line().await.unwrap().unwrap()).unwrap(); - assert_eq!(type_of(&ready), "ready"); - - cw.write_all( - json!({"op":"data","data":b64(b"12345")}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - cw.write_all(json!({"op":"data_end"}).to_string().as_bytes()) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); + send(&mut cw, json!({"op":"lsp_request","server_id":server_id})).await; + assert_eq!(type_of(&next_frame(&mut out).await), "ready"); + send(&mut cw, json!({"op":"data","data":b64(b"12345")})).await; + send(&mut cw, json!({"op":"data_end"})).await; - let reply = timeout(DEADLINE, out.next_line()) - .await - .expect("data_end never closed stdin") - .unwrap() - .unwrap(); - let frame: serde_json::Value = serde_json::from_str(&reply).unwrap(); + let frame = next_frame(&mut out).await; assert_eq!(type_of(&frame), "data"); assert_eq!(decode(&frame), b"ate:5\n"); - - let done = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - assert_eq!( - type_of(&serde_json::from_str::(&done).unwrap()), - "done" - ); + assert_eq!(type_of(&next_frame(&mut out).await), "done"); cw.shutdown().await.unwrap(); let _ = handle.await; } diff --git a/silkd/tests/pty_e2e.rs b/silkd/tests/pty_e2e.rs index 3b329276..15b3f7f6 100644 --- a/silkd/tests/pty_e2e.rs +++ b/silkd/tests/pty_e2e.rs @@ -8,14 +8,8 @@ use std::time::Duration; use serde_json::{Value, json}; use silkd::server::State; -use tokio::io::AsyncWriteExt; -use common::{FrameLines, FrameWriter}; - -async fn send(cw: &mut FrameWriter, frame: Value) { - cw.write_all(frame.to_string().as_bytes()).await.unwrap(); - cw.write_all(b"\n").await.unwrap(); -} +use common::{FrameLines, b64, connect, decode, exchange, one, send, type_of}; async fn read_until(lines: &mut FrameLines, pred: impl Fn(&Value) -> bool) -> Value { tokio::time::timeout(Duration::from_secs(5), async { @@ -34,7 +28,7 @@ async fn read_until(lines: &mut FrameLines, pred: impl Fn(&Value) -> bool) -> Va #[tokio::test] async fn pty_runs_a_shell_and_echoes() { let state = Arc::new(State::new()); - let (mut cw, mut lines, handle) = common::connect(&state); + let (mut cw, mut lines, handle) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let started = read_until(&mut lines, |v| v["type"] == "started").await; @@ -43,17 +37,15 @@ async fn pty_runs_a_shell_and_echoes() { send( &mut cw, - json!({"op":"stdin","data":common::b64(b"echo silk-pty-marker\n")}), + json!({"op":"stdin","data":b64(b"echo silk-pty-marker\n")}), ) .await; - let out = read_until(&mut lines, |v| { - v["type"] == "stdout" - && String::from_utf8_lossy(&common::decode(v)).contains("silk-pty-marker") + read_until(&mut lines, |v| { + v["type"] == "stdout" && String::from_utf8_lossy(&decode(v)).contains("silk-pty-marker") }) .await; - assert_eq!(out["type"], "stdout"); - send(&mut cw, json!({"op":"stdin","data":common::b64(b"exit\n")})).await; + send(&mut cw, json!({"op":"stdin","data":b64(b"exit\n")})).await; let exit = read_until(&mut lines, |v| v["type"] == "exit").await; assert!(exit["code"].is_number()); @@ -64,13 +56,13 @@ async fn pty_runs_a_shell_and_echoes() { #[tokio::test] async fn pty_appears_in_ps_and_resizes() { let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); + let (mut cw, mut lines, _) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let pid = read_until(&mut lines, |v| v["type"] == "started").await["pid"] .as_u64() .unwrap(); - let ps = common::one(&state, r#"{"op":"ps"}"#).await; + let ps = one(&state, r#"{"op":"ps"}"#).await; assert!( ps[0]["procs"] .as_array() @@ -79,12 +71,12 @@ async fn pty_appears_in_ps_and_resizes() { .any(|p| p["pid"].as_u64() == Some(pid)), "pty not in ps: {ps:?}" ); - let resized = common::one( + let resized = one( &state, &json!({"op":"pty_resize","pid":pid,"cols":120,"rows":40}).to_string(), ) .await; - assert_eq!(common::type_of(&resized[0]), "done"); + assert_eq!(type_of(&resized[0]), "done"); drop(cw); } @@ -92,18 +84,15 @@ async fn pty_appears_in_ps_and_resizes() { #[tokio::test] async fn resize_unknown_pid_is_not_found() { let frames = - common::exchange( - &[json!({"op":"pty_resize","pid":999999,"cols":80,"rows":24}).to_string()], - ) - .await; - assert_eq!(common::type_of(&frames[0]), "error"); + exchange(&[json!({"op":"pty_resize","pid":999999,"cols":80,"rows":24}).to_string()]).await; + assert_eq!(type_of(&frames[0]), "error"); assert_eq!(frames[0]["kind"], "not_found"); } #[tokio::test] async fn pty_disconnect_tears_down_without_spin() { let state = Arc::new(State::new()); - let (mut cw, mut lines, handle) = common::connect(&state); + let (mut cw, mut lines, handle) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let pid = read_until(&mut lines, |v| v["type"] == "started").await["pid"] .as_u64() @@ -116,7 +105,7 @@ async fn pty_disconnect_tears_down_without_spin() { let _ = tokio::time::timeout(Duration::from_secs(5), handle) .await .expect("pty did not tear down on disconnect"); - let ps = common::one(&state, r#"{"op":"ps"}"#).await; + let ps = one(&state, r#"{"op":"ps"}"#).await; assert!( !ps[0]["procs"] .as_array() diff --git a/silkd/tests/session_e2e.rs b/silkd/tests/session_e2e.rs index 624f26d9..5ec58fe1 100644 --- a/silkd/tests/session_e2e.rs +++ b/silkd/tests/session_e2e.rs @@ -8,16 +8,15 @@ use std::time::Duration; use serde_json::{Value, json}; use silkd::server::State; +use tokio::task::JoinHandle; use common::{one, stdout_body, type_of}; -async fn create(state: &Arc, extra: Value) -> String { - let mut req = json!({"op":"session_create"}); - if let Value::Object(map) = extra { - for (k, v) in map { - req[k] = v; - } - } +async fn create(state: &Arc) -> String { + create_with(state, json!({"op":"session_create"})).await +} + +async fn create_with(state: &Arc, req: Value) -> String { let f = one(state, &req.to_string()).await; assert_eq!(type_of(&f[0]), "session_created", "create failed: {f:?}"); f[0]["id"].as_str().unwrap().to_string() @@ -32,10 +31,16 @@ async fn sh(state: &Arc, id: &str, cmd: &[&str]) -> Vec { .await } +fn spawn_sh(state: &Arc, id: &str, cmd: &'static [&'static str]) -> JoinHandle> { + let s = Arc::clone(state); + let sid = id.to_string(); + tokio::spawn(async move { sh(&s, &sid, cmd).await }) +} + #[tokio::test] async fn session_persists_env_across_calls() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let set = sh(&state, &id, &["export", "MARKER=silk123"]).await; assert_eq!(type_of(set.last().unwrap()), "exit"); @@ -52,9 +57,9 @@ async fn session_persists_env_across_calls() { async fn session_applies_cwd_and_env_at_create() { let dir = tempfile::tempdir().unwrap(); let state = Arc::new(State::new()); - let id = create( + let id = create_with( &state, - json!({"cwd": dir.path().to_str().unwrap(), "env": {"GREETING": "hello-silk"}}), + json!({"op":"session_create","cwd": dir.path().to_str().unwrap(), "env": {"GREETING": "hello-silk"}}), ) .await; @@ -87,7 +92,7 @@ async fn session_create_with_hostile_env_key_still_answers() { #[tokio::test] async fn session_exit_code_propagates() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let ok = sh(&state, &id, &["true"]).await; assert_eq!(ok.last().unwrap()["code"], 0); @@ -98,7 +103,7 @@ async fn session_exit_code_propagates() { #[tokio::test] async fn session_list_and_rm() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let listed = one(&state, r#"{"op":"session_list"}"#).await; assert!( @@ -120,8 +125,8 @@ async fn session_list_and_rm() { #[tokio::test] async fn stdin_reading_command_does_not_wedge_the_session() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let r = tokio::time::timeout(std::time::Duration::from_secs(8), sh(&state, &id, &["cat"])) + let id = create(&state).await; + let r = tokio::time::timeout(Duration::from_secs(8), sh(&state, &id, &["cat"])) .await .expect("session wedged on a stdin-reading command"); assert_eq!(type_of(r.last().unwrap()), "exit"); @@ -132,7 +137,7 @@ async fn stdin_reading_command_does_not_wedge_the_session() { #[tokio::test] async fn shell_killing_command_removes_the_session() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let _ = sh(&state, &id, &["exit"]).await; let again = sh(&state, &id, &["echo", "hi"]).await; assert_eq!(type_of(&again[0]), "error"); @@ -142,7 +147,7 @@ async fn shell_killing_command_removes_the_session() { #[tokio::test] async fn forged_sentinel_in_output_does_not_desync() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let r = sh( &state, &id, @@ -164,7 +169,7 @@ async fn forged_sentinel_in_output_does_not_desync() { #[tokio::test] async fn reap_idle_removes_idle_sessions() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let reaped = state.sessions.reap_idle(Duration::ZERO); assert_eq!(reaped, 1); let gone = sh(&state, &id, &["echo", "hi"]).await; @@ -178,12 +183,8 @@ async fn reap_idle_removes_idle_sessions() { #[tokio::test] async fn reap_skips_a_session_running_a_command() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let busy = { - let s = Arc::clone(&state); - let sid = id.clone(); - tokio::spawn(async move { sh(&s, &sid, &["sleep", "1"]).await }) - }; + let id = create(&state).await; + let busy = spawn_sh(&state, &id, &["sleep", "1"]); tokio::time::sleep(Duration::from_millis(150)).await; let reaped = state.sessions.reap_idle(Duration::ZERO); assert_eq!(reaped, 0, "a running session must not be reaped"); @@ -195,7 +196,7 @@ async fn reap_skips_a_session_running_a_command() { #[tokio::test] async fn session_argv_with_metacharacters_is_one_literal_token() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let evil = "a b; echo INJECTED $(id) `whoami` 'q'"; let r = sh(&state, &id, &["printf", "%s", evil]).await; assert_eq!( @@ -212,16 +213,12 @@ async fn session_argv_with_metacharacters_is_one_literal_token() { #[tokio::test] async fn session_rm_unwedges_a_running_external_command() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let busy = { - let s = Arc::clone(&state); - let sid = id.clone(); - tokio::spawn(async move { sh(&s, &sid, &["sleep", "3600"]).await }) - }; - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let id = create(&state).await; + let busy = spawn_sh(&state, &id, &["sleep", "3600"]); + tokio::time::sleep(Duration::from_millis(200)).await; let rm = one(&state, &json!({"op":"session_rm","id":id}).to_string()).await; assert_eq!(type_of(&rm[0]), "done"); - let done = tokio::time::timeout(std::time::Duration::from_secs(5), busy) + let done = tokio::time::timeout(Duration::from_secs(5), busy) .await .expect("session_rm did not unwedge the external command"); let frames = done.unwrap(); @@ -231,7 +228,7 @@ async fn session_rm_unwedges_a_running_external_command() { #[tokio::test] async fn empty_argv_in_session_is_a_bad_request() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let f = one( &state, &json!({"op":"exec","argv":[],"session":id}).to_string(), @@ -244,7 +241,7 @@ async fn empty_argv_in_session_is_a_bad_request() { #[tokio::test] async fn detach_with_session_is_a_bad_request() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let f = one( &state, &json!({"op":"exec","argv":["true"],"session":id,"detach":true}).to_string(), @@ -265,7 +262,7 @@ async fn exec_in_unknown_session_is_not_found() { #[tokio::test] async fn session_merges_stderr_into_stdout() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let frames = sh(&state, &id, &["sh", "-c", "echo oops >&2"]).await; assert_eq!( diff --git a/silkd/tests/tree_e2e.rs b/silkd/tests/tree_e2e.rs index 234cf7b4..9aa42eb4 100644 --- a/silkd/tests/tree_e2e.rs +++ b/silkd/tests/tree_e2e.rs @@ -3,6 +3,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::io::Write; +use std::os::unix::fs::symlink; use std::path::Path; use std::process::Command; @@ -23,7 +25,6 @@ fn sys_tar_create(dir: &Path) -> Vec { } fn sys_tar_extract(archive: &[u8], into: &Path) { - use std::io::Write; let mut child = Command::new("tar") .arg("-x") .arg("-C") @@ -160,7 +161,6 @@ async fn pull_missing_parent_errors_not_found() { #[tokio::test] async fn pull_dangling_symlink_archives_the_link() { - use std::os::unix::fs::symlink; let src = tempfile::tempdir().unwrap(); let link = src.path().join("dangling"); symlink("/no/such/target", &link).unwrap(); @@ -202,13 +202,12 @@ async fn push_failure_leaves_dest_untouched() { .await; assert_eq!(type_of(frames.last().unwrap()), "error"); - let names: Vec = std::fs::read_dir(dest.path()) + let mut names: Vec = std::fs::read_dir(dest.path()) .unwrap() .map(|e| e.unwrap().file_name().to_string_lossy().into_owned()) .collect(); - let mut sorted = names.clone(); - sorted.sort(); - assert_eq!(sorted, ["keep.txt", "sub"], "dest mutated: {names:?}"); + names.sort(); + assert_eq!(names, ["keep.txt", "sub"], "dest mutated: {names:?}"); assert_eq!( std::fs::read(dest.path().join("keep.txt")).unwrap(), b"KEEP" From a38a3798b1928314b1fbf8a37558bf0a0d5a39a3 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:12:19 +0900 Subject: [PATCH 10/14] =?UTF-8?q?fix:=20python=20=E2=80=94=20running()=20c?= =?UTF-8?q?atches=20every=20SDK=20error,=20the=20mcp=20driver=20kills=20a?= =?UTF-8?q?=20wedged=20child,=20the=20exec=20description=20states=20the=20?= =?UTF-8?q?inactivity=20bound?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CocoonSandboxSession.running caught APIError, SilkdError and OSError, but the commonest not-running path is a failed dial, which the SDK raises as ProtocolError; it and StreamTimeout escaped a predicate whose job is to answer the question. SandboxError is the base of all four. mcp/e2e.py's finally-close could raise TimeoutExpired over the AssertionError naming the failed step and left the child running; McpClient is a context manager that kills on timeout. The langchain exec description promised a 5-minute cut-off, but CALL_TIMEOUT is a socket inactivity bound. --- mcp/e2e.py | 50 +++++++++---------- .../cocoonsandbox_langchain/toolkit.py | 7 +-- sdk/openai/cocoonsandbox_openai/adapter.py | 16 +++--- sdk/openai/tests/test_adapter.py | 18 ++++++- 4 files changed, 50 insertions(+), 41 deletions(-) diff --git a/mcp/e2e.py b/mcp/e2e.py index 8f576374..5bebb731 100644 --- a/mcp/e2e.py +++ b/mcp/e2e.py @@ -5,6 +5,8 @@ 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 @@ -12,13 +14,17 @@ 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 diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index 5d36a113..bf9d1e81 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -8,12 +8,13 @@ import asyncio import json import threading +from collections.abc import Callable from cocoonsandbox import Client, Sandbox from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field -# Mirrors the MCP exec contract the tool description states. +# a socket inactivity bound, not a wall clock; the tool description states it CALL_TIMEOUT = 300.0 @@ -72,7 +73,7 @@ def get_tools(self) -> list[StructuredTool]: "Returns stdout; a non-empty stderr is appended as a 'stderr:' " "line and a non-zero status as an 'exit code: N' line; a " "command that prints nothing and exits 0 returns '(no output)'. " - "The call is cut off after 5 minutes. " + "The call is cut off after 5 minutes without output. " "Files and installed packages persist across calls; environment " "variables and the working directory do not.", ExecInput, @@ -126,7 +127,7 @@ def _claim(self) -> Sandbox: return self._client.checkpoint(self._from_checkpoint).new(ttl_seconds=self._ttl) return self._client.new(self._template, net=self._net, ttl_seconds=self._ttl) - def _tool(self, name: str, description: str, schema: type[BaseModel], func) -> StructuredTool: + def _tool(self, name: str, description: str, schema: type[BaseModel], func: Callable[..., str]) -> StructuredTool: async def arun(**kwargs): return await asyncio.to_thread(func, **kwargs) diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index 9ea611b3..989aa467 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -19,7 +19,7 @@ from agents.sandbox.session.sandbox_session_state import SandboxSessionState from agents.sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User -from cocoonsandbox import APIError, Client, Sandbox, SilkdError +from cocoonsandbox import Client, Sandbox, SandboxError, SilkdError class CocoonSandboxClientOptions(BaseSandboxClientOptions): @@ -59,12 +59,10 @@ def from_state(cls, state: CocoonSandboxSessionState) -> CocoonSandboxSession: async def _prepare_backend_workspace(self) -> None: sb = self._sandbox() - await asyncio.to_thread(sb.mkdir, str(self.state.manifest.root), True) + await asyncio.to_thread(sb.mkdir, str(self.state.manifest.root), parents=True) async def _exec_internal(self, *command: str | Path, timeout: float | None = None) -> ExecResult: - # Bound the blocking SDK call by the socket timeout too, so a - # wait_for cancellation is matched by the worker thread actually - # unblocking (recv wakes) instead of lingering. + # the socket timeout bounds the blocking SDK call too, so a wait_for cancellation unblocks the worker thread sb = self._sandbox(timeout=timeout) argv = [str(part) for part in command] stdout, stderr = bytearray(), bytearray() @@ -95,7 +93,7 @@ async def write(self, path: Path, data: io.IOBase, *, user: str | User | None = async def running(self) -> bool: try: await asyncio.to_thread(self._sandbox().stat, "/") - except (APIError, SilkdError, OSError): + except (SandboxError, OSError): return False return True @@ -115,8 +113,7 @@ async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint: return ExposedPortEndpoint(host="127.0.0.1", port=listener.getsockname()[1], tls=False) async def _shutdown_backend(self) -> None: - # The sandbox itself outlives shutdown (delete releases it); only the - # local port proxies belong to this process. + # the sandbox outlives shutdown (delete releases it); only the local port proxies belong to this process for listener in self._proxies: listener.close() self._proxies.clear() @@ -179,7 +176,6 @@ def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessio def _reject_user(user: str | User | None) -> None: - # fs_read/fs_write carry no per-call user on the wire; fail loud rather - # than silently running as the guest default. + # fs_read/fs_write carry no per-call user on the wire; fail loud rather than run as the guest default if user is not None: raise NotImplementedError("per-call user impersonation is not supported by the cocoon backend") diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index cd44f448..ccf3ca9c 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -11,7 +11,7 @@ from pathlib import Path import pytest -from cocoonsandbox import SilkdError +from cocoonsandbox import ProtocolError, SilkdError from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions, CocoonSandboxSessionState @@ -68,7 +68,6 @@ async def go(): def test_exec_maps_stdio_and_exit(node, monkeypatch): - class FakeSandbox: def __init__(self, **kw): self.id = "sb_1" @@ -106,6 +105,21 @@ async def go(): asyncio.run(go()) +def test_running_is_false_when_the_dial_fails(node, monkeypatch): + class FakeSandbox: + def stat(self, path): + raise ProtocolError("dial 127.0.0.1:1: connection refused") + + async def go(): + client = CocoonSandboxClient() + session = await client.create(options=CocoonSandboxClientOptions(addr=node)) + inner = session._inner + monkeypatch.setattr(inner, "_sandbox", lambda timeout=None: FakeSandbox()) + assert await inner.running() is False + + asyncio.run(go()) + + def test_write_and_persist_use_tree_verbs(node, monkeypatch): calls = {} From c4335a310b91d2a61fe981b33d9b5e31e512a7e6 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:12:32 +0900 Subject: [PATCH 11/14] =?UTF-8?q?review:=20python=20=E2=80=94=20one=20clai?= =?UTF-8?q?m=20recorder,=20inlined=20stubs,=20helpers=20below=20tests,=20a?= =?UTF-8?q?nnotated=20helpers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_client wrote the same recording claim handler seven times; recording_claim returns the seen list. test_wire_binding's three one-line stub wrappers fold into the CASES they serve, and the fixture helpers in test_wire_binding, test_proc and test_toolkit move below the tests they support. test_proc built a FakeConn it immediately replaced. client's three hardest helpers (_try_each, _redirect_fallback, _scatter) gain full annotations, the default retry policy becomes a named function, and their docstrings keep the policy while dropping the sentence that restated the name; sandbox annotates the stdio callbacks the public methods already type. --- sdk/langchain/tests/test_toolkit.py | 66 ++++++++-------- sdk/python/cocoonsandbox/client.py | 52 +++++++------ sdk/python/cocoonsandbox/sandbox.py | 12 ++- sdk/python/tests/test_client.py | 94 +++++++---------------- sdk/python/tests/test_proc.py | 90 +++++++++++----------- sdk/python/tests/test_wire_binding.py | 106 ++++++++++++-------------- 6 files changed, 191 insertions(+), 229 deletions(-) diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 08e8c1f8..ab604229 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -8,39 +8,6 @@ from cocoonsandbox_langchain import CocoonToolkit -class FakeSandbox: - def __init__(self): - self.closed = 0 - self.files = {} - - def run(self, argv, cwd="", on_stdout=None, on_stderr=None, **_): - assert argv[:2] == ["sh", "-c"] - if argv[2] == "boom": - on_stderr(b"kaboom\n") - return 3 - on_stdout(f"ran: {argv[2]}\n".encode()) - return 0 - - def write_file(self, path, data): - self.files[path] = data - - def read_file(self, path): - return self.files[path] - - def list_dir(self, path): - return [{"name": "a.txt", "kind": "file", "size": 3}] - - def close(self): - self.closed += 1 - - -def hooked(monkeypatch): - kit = CocoonToolkit("127.0.0.1:1") - fake = FakeSandbox() - monkeypatch.setattr(kit, "_claim", lambda: fake) - return kit, fake - - def test_tools_shape(monkeypatch): kit, _ = hooked(monkeypatch) tools = kit.get_tools() @@ -86,3 +53,36 @@ def test_use_after_close_raises(monkeypatch): kit.close() with pytest.raises(RuntimeError): kit.sandbox() + + +class FakeSandbox: + def __init__(self): + self.closed = 0 + self.files = {} + + def run(self, argv, cwd="", on_stdout=None, on_stderr=None, **_): + assert argv[:2] == ["sh", "-c"] + if argv[2] == "boom": + on_stderr(b"kaboom\n") + return 3 + on_stdout(f"ran: {argv[2]}\n".encode()) + return 0 + + def write_file(self, path, data): + self.files[path] = data + + def read_file(self, path): + return self.files[path] + + def list_dir(self, path): + return [{"name": "a.txt", "kind": "file", "size": 3}] + + def close(self): + self.closed += 1 + + +def hooked(monkeypatch): + kit = CocoonToolkit("127.0.0.1:1") + fake = FakeSandbox() + monkeypatch.setattr(kit, "_claim", lambda: fake) + return kit, fake diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 44d8e8cc..943f5d51 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -11,12 +11,15 @@ import urllib.error import urllib.parse import urllib.request -from collections.abc import Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence +from typing import TypeVar from .checkpoint import Checkpoint from .errors import APIError from .sandbox import Sandbox +T = TypeVar("T") + _PEERS_TIMEOUT = 5.0 @@ -192,7 +195,7 @@ def _request( except urllib.error.URLError as exc: raise APIError(verb, 0, str(exc.reason)) from None except (OSError, http.client.HTTPException) as exc: - # A reset or truncated body read is neither HTTPError nor URLError. + # a reset or truncated body read is neither HTTPError nor URLError raise APIError(verb, 0, str(exc)) from None if not raw: return {} @@ -266,11 +269,16 @@ def _error_message(raw: bytes) -> str: return raw.decode(errors="replace").strip() -def _try_each(candidates, call, retry=lambda exc: exc.status in (404, 0)): - """Calls call against each candidate in turn, returning the first - success. An APIError for which retry(exc) is true moves on to the next +def _retry_miss(exc: APIError) -> bool: + """Default _try_each policy: only a miss (404) or a dead peer (status 0) moves on.""" + return exc.status in (404, 0) + + +def _try_each( + candidates: Iterable[str], call: Callable[[str], T], retry: Callable[[APIError], bool] = _retry_miss +) -> T: + """An APIError for which retry(exc) is true moves on to the next candidate and the last such error propagates; any other raises at once. - The default retries only a miss (404 or a dead peer, status 0); candidates must be non-empty.""" last_error = None for addr in candidates: @@ -293,17 +301,18 @@ def _retry_transient(exc: APIError) -> bool: return exc.status in (0, 401, 404, 429, 503, 500, 502, 504) -def _redirect_fallback(origin: str, candidates: list, post, verb: str): - """Walks candidates via post(addr) -> raw reply dict, retrying broadly - (any candidate failure moves to the next) so one wrong candidate doesn't - cost a candidate that would still succeed. If every candidate is - exhausted and the last failure was transient (_retry_transient), gives - the origin one more no_redirect attempt -- the node that issued the - redirect provisions or heals locally instead of leaving the claim stuck - on stale gossip. A definitive last failure skips the fallback: the origin - would fail the same way. A second-level redirect (a compliant server - never sends one once no_redirect is set) fails the candidate rather than - being followed. Returns (addr, reply).""" +def _redirect_fallback( + origin: str, candidates: Sequence[str], post: Callable[[str], dict], verb: str +) -> tuple[str, dict]: + """Retries broadly across candidates (any failure moves to the next) so + one wrong candidate doesn't cost one that would still succeed. If every + candidate is exhausted and the last failure was transient + (_retry_transient), gives the origin one more no_redirect attempt -- the + node that issued the redirect provisions or heals locally instead of + leaving the claim stuck on stale gossip. A definitive last failure skips + the fallback: the origin would fail the same way. A second-level redirect + (a compliant server never sends one once no_redirect is set) fails the + candidate rather than being followed.""" def attempt(addr): reply = post(addr) @@ -324,11 +333,10 @@ def attempt(addr): raise APIError(verb, origin_exc.status, combined) from origin_exc -def _scatter(addrs, probe): - """Probes every addr concurrently and returns the first success; when - all probes fail the last error propagates. Loser threads are daemons - whose requests die with _request's own timeout; the queue is bounded by - len(addrs) so they never block. addrs must be non-empty.""" +def _scatter(addrs: Sequence[str], probe: Callable[[str], T]) -> T: + """When all probes fail the last error propagates. Loser threads are + daemons whose requests die with _request's own timeout; the queue is + bounded by len(addrs) so they never block. addrs must be non-empty.""" results = queue.Queue(maxsize=len(addrs)) def run(addr): diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 7470c983..7d57855a 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -406,7 +406,13 @@ def _call(self, op: str, expect: str, **fields) -> dict: def _done_rpc(self, op: str, **fields) -> None: self._call(op, "done", **fields) - def _drain_proc(self, op: str, pid: int, on_stdout, on_stderr) -> int | None: + def _drain_proc( + self, + op: str, + pid: int, + on_stdout: Callable[[bytes], object] | None, + on_stderr: Callable[[bytes], object] | None, + ) -> int | None: with self._dial() as conn: conn.send(op, pid=pid) return _pump_stdio(conn, on_stdout, on_stderr) @@ -541,7 +547,9 @@ def _feed_stdin(conn: Conn, stdin: bytes) -> None: conn.send("stdin_close") -def _pump_stdio(conn: Conn, on_stdout, on_stderr) -> int | None: +def _pump_stdio( + conn: Conn, on_stdout: Callable[[bytes], object] | None, on_stderr: Callable[[bytes], object] | None +) -> int | None: """Streams stdout/stderr frames into the callbacks until the terminal frame: the exit code, or None when the stream ends with done.""" for frame in conn.recv_until("exit", "done"): diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 08130f74..44020ee9 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -64,11 +64,8 @@ def test_claim_happy_path(node): def test_claim_sends_volumes(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { + seen = recording_claim( + { "id": "sb_1", "token": "tok", "volumes": [ @@ -76,8 +73,7 @@ def claim(body, path): {"name": "weights-llama", "mount": "/models"}, ], } - - FakeNode.routes[("POST", "/v1/claim")] = claim + ) sb = Client(node).new("rt:24.04", volumes=["imagenet", {"name": "weights-llama", "mount": "/models"}]) assert seen == [ { @@ -108,19 +104,9 @@ def test_claim_rejects_invalid_volumes(node, volumes, match): def test_claim_sends_volume_mode_rw(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_1", - "token": "tok", - "volumes": [ - {"name": "scratch", "mount": "/data", "mode": "rw"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_1", "token": "tok", "volumes": [{"name": "scratch", "mount": "/data", "mode": "rw"}]} + ) sb = Client(node).new("rt:24.04", volumes=[{"name": "scratch", "mount": "/data", "mode": "rw"}]) assert seen == [ { @@ -134,33 +120,16 @@ def claim(body, path): def test_claim_omits_volume_mode_ro(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_1", "token": "tok"} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_1", "token": "tok"}) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": "ro"}]) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": ""}]) assert seen[0]["volumes"] == seen[1]["volumes"] == [{"name": "imagenet"}], seen def test_claim_attaches_volumes_without_mounting(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_1", - "token": "tok", - "volumes": [ - {"name": "imagenet"}, - {"name": "scratch", "mode": "rw"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_1", "token": "tok", "volumes": [{"name": "imagenet"}, {"name": "scratch", "mode": "rw"}]} + ) sb = Client(node).new("rt:24.04", volumes=["imagenet", {"name": "scratch", "mode": "rw"}], mount=False) assert seen == [ { @@ -173,13 +142,7 @@ def claim(body, path): def test_template_claim_attaches_volumes_without_mounting(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet"}]} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet"}]}) sb = Template(Client(node), node, "task:v1", "none", "small").new(volumes=["imagenet"], mount=False) assert seen[0]["volumes_attach_only"] is True assert seen[0]["volumes"] == [{"name": "imagenet"}] @@ -196,31 +159,15 @@ def test_claim_rejects_mount_without_mounting(node): def test_claim_keeps_mounting_by_default(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_1", "token": "tok"} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_1", "token": "tok"}) Client(node).new("rt:24.04", volumes=["imagenet"]) assert "volumes_attach_only" not in seen[0] def test_template_claim_sends_volumes(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_2", - "token": "tok", - "volumes": [ - {"name": "imagenet", "mount": "/datasets/imagenet"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet", "mount": "/datasets/imagenet"}]} + ) sb = Template(Client(node), node, "task:v1", "none", "small").new( volumes=[{"name": "imagenet", "mount": "/datasets/imagenet"}] ) @@ -357,3 +304,14 @@ def test_checkpoint_listing_binds_handles(node): ) branch = ckpts[0].new() assert branch.id == "sb_branch" + + +def recording_claim(reply): + seen = [] + + def claim(body, path): + seen.append(body) + return 200, reply + + FakeNode.routes[("POST", "/v1/claim")] = claim + return seen diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py index 2f54b452..8cb4bd64 100644 --- a/sdk/python/tests/test_proc.py +++ b/sdk/python/tests/test_proc.py @@ -7,6 +7,51 @@ from cocoonsandbox.frames import FS_CHUNK +def test_spawn_returns_pid(monkeypatch): + sb, conn = fake_sandbox(monkeypatch, [{"type": "started", "pid": 41}]) + assert sb.spawn("sleep", "30") == 41 + op, fields = conn.sent[0] + assert op == "exec" and fields["detach"] is True and fields["argv"] == ["sleep", "30"] + + +def test_ps_lists_procs(monkeypatch): + procs = [{"pid": 41, "argv": ["sleep"], "detached": True, "state": "running", "started_at_epoch_secs": 1}] + sb, _ = fake_sandbox(monkeypatch, [{"type": "procs", "procs": procs}]) + assert sb.ps() == procs + + +def test_kill_sends_signal(monkeypatch): + sb, conn = fake_sandbox(monkeypatch, [{"type": "done"}]) + sb.kill(41, signal=15) + assert conn.sent[0] == ("kill", {"pid": 41, "signal": 15}) + + +def test_logs_running_proc_has_no_code(monkeypatch): + frames = [{"type": "stdout", "data": b"hello"}, {"type": "stderr", "data": b"oops"}, {"type": "done"}] + sb, _ = fake_sandbox(monkeypatch, frames) + out, errs = [], [] + assert sb.logs(41, on_stdout=out.append, on_stderr=errs.append) is None + assert out == [b"hello"] and errs == [b"oops"] + + +def test_attach_returns_exit_code(monkeypatch): + frames = [{"type": "stdout", "data": b"late"}, {"type": "exit", "code": 7}] + sb, _ = fake_sandbox(monkeypatch, frames) + out = [] + assert sb.attach(41, on_stdout=out.append) == 7 + assert out == [b"late"] + + +def test_run_pumps_stdin_while_reading_output(monkeypatch): + sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") + blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) + monkeypatch.setattr(sb, "_dial", lambda: blocking) + + assert sb.run(["cat"], stdin=b"x" * (FS_CHUNK * 3)) == 0 + assert [op for op, _ in blocking.sent].count("stdin") == 3 + assert blocking.sent[-1][0] == "stdin_close" + + class FakeConn: """Answers one RPC with a canned frame list.""" @@ -62,48 +107,3 @@ def fake_sandbox(monkeypatch, frames): conn = FakeConn(frames) monkeypatch.setattr(sb, "_dial", lambda: conn) return sb, conn - - -def test_spawn_returns_pid(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "started", "pid": 41}]) - assert sb.spawn("sleep", "30") == 41 - op, fields = conn.sent[0] - assert op == "exec" and fields["detach"] is True and fields["argv"] == ["sleep", "30"] - - -def test_ps_lists_procs(monkeypatch): - procs = [{"pid": 41, "argv": ["sleep"], "detached": True, "state": "running", "started_at_epoch_secs": 1}] - sb, _ = fake_sandbox(monkeypatch, [{"type": "procs", "procs": procs}]) - assert sb.ps() == procs - - -def test_kill_sends_signal(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "done"}]) - sb.kill(41, signal=15) - assert conn.sent[0] == ("kill", {"pid": 41, "signal": 15}) - - -def test_logs_running_proc_has_no_code(monkeypatch): - frames = [{"type": "stdout", "data": b"hello"}, {"type": "stderr", "data": b"oops"}, {"type": "done"}] - sb, _ = fake_sandbox(monkeypatch, frames) - out, errs = [], [] - assert sb.logs(41, on_stdout=out.append, on_stderr=errs.append) is None - assert out == [b"hello"] and errs == [b"oops"] - - -def test_attach_returns_exit_code(monkeypatch): - frames = [{"type": "stdout", "data": b"late"}, {"type": "exit", "code": 7}] - sb, _ = fake_sandbox(monkeypatch, frames) - out = [] - assert sb.attach(41, on_stdout=out.append) == 7 - assert out == [b"late"] - - -def test_run_pumps_stdin_while_reading_output(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "exit", "code": 0}]) - blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) - monkeypatch.setattr(sb, "_dial", lambda: blocking) - - assert sb.run(["cat"], stdin=b"x" * (FS_CHUNK * 3)) == 0 - assert [op for op, _ in blocking.sent].count("stdin") == 3 - assert blocking.sent[-1][0] == "stdin_close" diff --git a/sdk/python/tests/test_wire_binding.py b/sdk/python/tests/test_wire_binding.py index f7e1de60..0622c826 100644 --- a/sdk/python/tests/test_wire_binding.py +++ b/sdk/python/tests/test_wire_binding.py @@ -60,7 +60,7 @@ ), ("req_git_push", [{"type": "done"}], lambda sb, f: sb.git_push(f["path"], auth=f["auth"])), ("req_git_pull", [{"type": "done"}], lambda sb, f: sb.git_pull(f["path"], auth=f["auth"])), - ("req_pty_resize", [{"type": "done"}], lambda sb, f: _pty_stub(sb, f["pid"]).resize(f["cols"], f["rows"])), + ("req_pty_resize", [{"type": "done"}], lambda sb, f: Pty(sb, None, f["pid"]).resize(f["cols"], f["rows"])), ("req_exec_detach", [{"type": "started", "pid": 7}], lambda sb, f: sb.spawn(*f["argv"])), ("req_ps", [{"type": "procs", "procs": []}], lambda sb, f: sb.ps()), ("req_kill", [{"type": "done"}], lambda sb, f: sb.kill(f["pid"], signal=f["signal"])), @@ -68,14 +68,14 @@ ("req_attach", [{"type": "exit", "code": 0}], lambda sb, f: sb.attach(f["pid"])), ("req_fs_watch", [{"type": "ready"}], lambda sb, f: sb.watch(f["path"], recursive=f["recursive"]).close()), ("req_git_branch", [{"type": "done"}], lambda sb, f: sb.git_create_branch(f["path"], f["name"])), - ("req_session_rm", [{"type": "done"}], lambda sb, f: _session_stub(sb, f["id"]).close()), + ("req_session_rm", [{"type": "done"}], lambda sb, f: Session(sb, f["id"]).close()), ( "req_lsp_start", [{"type": "lsp_started", "server_id": "lsp-1"}], lambda sb, f: sb.start_lsp(f["language"], root=f["root"]), ), - ("req_lsp_request", [{"type": "ready"}], lambda sb, f: _lsp_stub(sb, f["server_id"]).request().close()), - ("req_lsp_stop", [{"type": "done"}], lambda sb, f: _lsp_stub(sb, f["server_id"]).stop()), + ("req_lsp_request", [{"type": "ready"}], lambda sb, f: Lsp(sb, f["server_id"]).request().close()), + ("req_lsp_stop", [{"type": "done"}], lambda sb, f: Lsp(sb, f["server_id"]).stop()), ("req_port_forward", [{"type": "ready"}], lambda sb, f: sb.dial_port(f["port"]).close()), ( "req_pty_open", @@ -88,6 +88,48 @@ UNSENT = {"req_session_create": {"id"}} +@pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") +def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): + fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) + sb, sent, thread = fake_sandbox(monkeypatch, replies) + invoke(sb, fixture) + thread.join(timeout=5) + + assert sent, "no frame reached the guest" + frame = sent[0] + assert frame["op"] == fixture["op"] + assert frame["v"] == PROTO_VERSION + for key, value in frame.items(): + assert key in fixture, f"field {key!r} not in the golden corpus (typo?)" + if key != "data": + assert value == fixture[key], f"field {key!r}: {value!r} != {fixture[key]!r}" + for key in fixture: + if key in UNSENT.get(stem, ()): + continue + assert key in frame, f"fixture field {key!r} was never emitted (dropped field?)" + + +def test_enum_value_sets_match_corpus(): + enums = json.loads((FIXTURES / "enums.json").read_text()) + assert set(enums["error_kind"]) == {"bad_request", "not_found", "unimplemented", "internal"} + assert set(enums["event_kind"]) == {"created", "modified", "deleted", "renamed"} + assert set(enums["file_kind"]) == {"file", "dir", "symlink", "other"} + assert set(enums["git_branch_action"]) == {"list", "create", "delete", "checkout"} + + +def test_git_branch_actions_come_from_the_corpus(monkeypatch): + enums = json.loads((FIXTURES / "enums.json").read_text()) + sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") + sent = [] + monkeypatch.setattr(sb, "_dial", lambda: BranchActionConn(sent)) + + sb.git_branches("/w") + sb.git_create_branch("/w", "b") + sb.git_delete_branch("/w", "b") + sb.git_checkout("/w", "b") + assert set(sent) == set(enums["git_branch_action"]) + + class BranchActionConn: """Records the action each git_branch verb puts on the wire.""" @@ -114,19 +156,7 @@ def recv_until(self, *terminal): yield self.recv() -def _pty_stub(sb, pid): - return Pty(sb, None, pid) - - -def _session_stub(sb, id): - return Session(sb, id) - - -def _lsp_stub(sb, server_id): - return Lsp(sb, server_id) - - -def _fake_sandbox(monkeypatch, replies): +def fake_sandbox(monkeypatch, replies): """A Sandbox whose _dial yields a real Conn over a socketpair; a guest thread records inbound frames and answers the scripted replies.""" sent = [] @@ -154,45 +184,3 @@ def guest(): sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") monkeypatch.setattr(sb, "_dial", lambda: Conn(client_sock, client_sock.makefile("rb"))) return sb, sent, thread - - -@pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") -def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): - fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) - sb, sent, thread = _fake_sandbox(monkeypatch, replies) - invoke(sb, fixture) - thread.join(timeout=5) - - assert sent, "no frame reached the guest" - frame = sent[0] - assert frame["op"] == fixture["op"] - assert frame["v"] == PROTO_VERSION - for key, value in frame.items(): - assert key in fixture, f"field {key!r} not in the golden corpus (typo?)" - if key != "data": - assert value == fixture[key], f"field {key!r}: {value!r} != {fixture[key]!r}" - for key in fixture: - if key in UNSENT.get(stem, ()): - continue - assert key in frame, f"fixture field {key!r} was never emitted (dropped field?)" - - -def test_enum_value_sets_match_corpus(): - enums = json.loads((FIXTURES / "enums.json").read_text()) - assert set(enums["error_kind"]) == {"bad_request", "not_found", "unimplemented", "internal"} - assert set(enums["event_kind"]) == {"created", "modified", "deleted", "renamed"} - assert set(enums["file_kind"]) == {"file", "dir", "symlink", "other"} - assert set(enums["git_branch_action"]) == {"list", "create", "delete", "checkout"} - - -def test_git_branch_actions_come_from_the_corpus(monkeypatch): - enums = json.loads((FIXTURES / "enums.json").read_text()) - sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") - sent = [] - monkeypatch.setattr(sb, "_dial", lambda: BranchActionConn(sent)) - - sb.git_branches("/w") - sb.git_create_branch("/w", "b") - sb.git_delete_branch("/w", "b") - sb.git_checkout("/w", "b") - assert set(sent) == set(enums["git_branch_action"]) From 583ddcdfd95606dc65f22c7c025a98ee21546b89 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:13:48 +0900 Subject: [PATCH 12/14] =?UTF-8?q?build:=20python=20gate=20=E2=80=94=20form?= =?UTF-8?q?at=20check,=20pinned=20ruff=20and=20pytest,=20mcp/=20governed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI ran ruff check only, with a floating ruff; the /code-py gate is format --check + check + pytest at pinned versions. A root ruff.toml gives mcp/e2.py the same 120-column, same-rule configuration the three sdk packages carry in their pyproject files, and the workflow now fires on mcp/**/*.py. --- .github/workflows/python.yml | 8 ++++++-- ruff.toml | 5 +++++ 2 files changed, 11 insertions(+), 2 deletions(-) create mode 100644 ruff.toml diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 4b82aaa7..484f416c 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -6,6 +6,8 @@ on: - "sdk/python/**" - "sdk/openai/**" - "sdk/langchain/**" + - "mcp/**/*.py" + - "ruff.toml" - "protocol/**" - ".github/workflows/python.yml" pull_request: @@ -13,6 +15,8 @@ on: - "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 diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 00000000..66f33c4f --- /dev/null +++ b/ruff.toml @@ -0,0 +1,5 @@ +line-length = 120 +target-version = "py310" + +[lint] +select = ["E", "F", "W", "I", "UP", "B", "SIM"] From c82965b1464abcadc42a888f98f3948945312427 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:14:07 +0900 Subject: [PATCH 13/14] docs: the langchain exec tool's cut-off is an inactivity bound --- docs/langchain.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/langchain.md b/docs/langchain.md index aafdb362..112a1cbd 100644 --- a/docs/langchain.md +++ b/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 | From efa769a650fec82399e3431350af2228fcf93971 Mon Sep 17 00:00:00 2001 From: CMGS Date: Sat, 12 Sep 2026 01:17:29 +0900 Subject: [PATCH 14/14] =?UTF-8?q?review:=20sdk/go=20=E2=80=94=20Run=20assi?= =?UTF-8?q?gns=20the=20stdin-close=20error=20to=20the=20outer=20err?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With the stdio pump returning through the function-scope err, the inner err in the StdinClose branch shadowed it (govet shadow). --- sdk/go/sandbox.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index fa971548..f786b1c0 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -109,7 +109,7 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { defer done() if cmd.Stdin == nil { - if err := conn.Send(wire.StdinClose{}); err != nil { + if err = conn.Send(wire.StdinClose{}); err != nil { return 0, fmt.Errorf("close stdin: %w", err) } } else {