From 8c21419ed288c4cf0ec96cb32a24f7e5ef450303 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 13:08:07 +0800 Subject: [PATCH 1/2] git.status caps its frame; find streams to the caller A status of a large tree rendered one frame past the cap and dropped the connection with no terminal frame. The entry list now stops at an eighth of the frame cap in estimated bytes and the result carries truncated: true, so the head of the list still arrives. The Go SDK gains FindSeq and the Python SDK find_iter: matches stream to the caller, and leaving the loop closes the connection, which ends the walk in the guest, so the caller bounds the result instead of accumulating it. Find keeps its slice API on top of the same stream. --- docs/sdk-python.md | 3 +- docs/sdk.md | 7 ++- docs/silkd.md | 2 +- .../v1/resp_git_status_result_truncated.json | 1 + protocol/wire/frame.go | 9 ++-- protocol/wire/frame_test.go | 4 +- sdk/go/find.go | 7 +++ sdk/go/find_test.go | 31 +++++++++++ sdk/go/utils.go | 53 +++++++++++++------ sdk/python/cocoonsandbox/sandbox.py | 9 +++- silkd/src/git.rs | 31 +++++++++++ silkd/src/proto.rs | 4 +- 12 files changed, 133 insertions(+), 28 deletions(-) create mode 100644 protocol/wire/fixtures/v1/resp_git_status_result_truncated.json diff --git a/docs/sdk-python.md b/docs/sdk-python.md index 99e3c75e..c944f136 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -367,6 +367,7 @@ path on the no-network lane. ```python matches = sb.find("/work", r"TODO|FIXME", glob="*.py") +for m in sb.find_iter("/work", r"TODO"): # streamed; leaving the loop ends the walk in the guest # [{"file", "line", "content"}]; glob is anchored *? wildcards on the file name results = sb.replace(["/work/main.py"], r"foo", "bar") @@ -394,7 +395,7 @@ when the relay drops, which `w.error` tells apart from a clean close (`None`). ```python sb.git_clone(url, "/work/repo", branch="main", depth=1, auth=token) # egress lane only -st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files"} +st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files", "truncated"?} sb.git_add("/work/repo", ["a.txt"]) sha = sb.git_commit("/work/repo", "message", "Dev ") sb.git_push("/work/repo", auth=token) # egress lane only diff --git a/docs/sdk.md b/docs/sdk.md index 577aa31d..720bf103 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -441,6 +441,11 @@ err = sb.Pull(ctx, "/work", tarWriter) // stream /work back as a tar ```go matches, err := sb.Find(ctx, "/work", `TODO|FIXME`, "*.go") // []wire.Match{File, Line, Content}; glob is anchored *? wildcards on the file name +for m, err := range sb.FindSeq(ctx, "/work", `TODO`, "") { // streamed; a break ends the walk in the guest + if err != nil || m.Line > 100 { + break + } +} results, err := sb.Replace(ctx, []string{"/work/main.go"}, `foo`, "bar") // []wire.Replaced{File, Replacements}; per-file atomic @@ -468,7 +473,7 @@ overflow instead of the stream silently dropping events. ```go err = sb.GitClone(ctx, url, "/work/repo", "main", 0, token) // egress lane only; depth > 0 = shallow -st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[] +st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[]; Truncated when the list was cut at ~1 MiB err = sb.GitAdd(ctx, "/work/repo", "a.txt") hash, err := sb.GitCommit(ctx, "/work/repo", "message", "Dev ") err = sb.GitPush(ctx, "/work/repo", token) // egress lane only diff --git a/docs/silkd.md b/docs/silkd.md index 4f096a58..20e2d0e2 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -38,7 +38,7 @@ a frame only one side can parse fails CI. | search | `fs_find {path, pattern, glob?}` → `match{file, line, content}`… → `done` / `fs_replace {files, pattern, replacement}` → `replaced{file, replacements}`… → `done` | regex as data, no shell quoting; `glob` is anchored `*`/`?` wildcards over file names; find skips binary and >8 MiB files, and replace skips the same 8 MiB bound with a zero count | | watch | `fs_watch {path, recursive?}` | `ready` once armed (events after it are guaranteed captured), then `event{kind, path}` until the client disconnects; watcher or delivery-queue overflow errors arrive as a terminal `error` instead of silently losing events | | pty | `pty_open {cols, rows, cwd?, env?, user?}` / `pty_resize {pid, cols, rows}` | a shell under a pseudo-terminal; output as `stdout` frames, input as `stdin` frames, exit terminal. PTYs register in the proc table like any exec | -| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). `auth` is injected as an in-memory header, never written to guest disk | +| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | | port | `port_forward {port}` | relays guest TCP 127.0.0.1:port over this connection: `ready` once connected, then `data` both ways (`data_end` half-closes the guest socket); the server closing ends the stream with `done`. Works on both lanes — the no-network lane's only way in | | lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the flavor) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | | misc | `info` | `{version, proto, uptime_secs, procs, sessions}` — the in-guest readiness probe (sandboxd consumes it; no SDK surface). Distinct from the control plane's `GET /v1/info`, which reports node pools and claims | diff --git a/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json new file mode 100644 index 00000000..be64dcb7 --- /dev/null +++ b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json @@ -0,0 +1 @@ +{"type":"git_status_result","branch":"main","ahead":0,"behind":0,"files":[{"path":"a.txt","staged":"?","unstaged":"?"}],"truncated":true} diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 24dcd9e7..499809f2 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -594,10 +594,11 @@ func (Event) RespType() string { return "event" } // GitStatusResult answers GitStatus. type GitStatusResult struct { - Branch string `json:"branch"` - Ahead uint32 `json:"ahead"` - Behind uint32 `json:"behind"` - Files []GitFileStatus `json:"files"` + Branch string `json:"branch"` + Ahead uint32 `json:"ahead"` + Behind uint32 `json:"behind"` + Files []GitFileStatus `json:"files"` + Truncated bool `json:"truncated,omitempty"` } func (GitStatusResult) RespType() string { return "git_status_result" } diff --git a/protocol/wire/frame_test.go b/protocol/wire/frame_test.go index 50eefd1e..4efef372 100644 --- a/protocol/wire/frame_test.go +++ b/protocol/wire/frame_test.go @@ -58,8 +58,8 @@ func TestFixtureCorpusRoundTrips(t *testing.T) { seen++ } - if seen != 60 { - t.Fatalf("fixture corpus: %d frames, want exactly 60", seen) + if seen != 61 { + t.Fatalf("fixture corpus: %d frames, want exactly 61", seen) } } diff --git a/sdk/go/find.go b/sdk/go/find.go index 9fd2c4bf..69b0c5b3 100644 --- a/sdk/go/find.go +++ b/sdk/go/find.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" ) @@ -13,6 +14,12 @@ func (s *Sandbox) Find(ctx context.Context, path, pattern, glob string) ([]wire. return collectRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) } +// FindSeq streams the matches Find would collect; breaking out of the loop +// ends the walk in the guest, so the caller bounds the result. +func (s *Sandbox) FindSeq(ctx context.Context, path, pattern, glob string) iter.Seq2[wire.Match, error] { + return streamRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) +} + // Replace rewrites pattern (a regular expression) to replacement in each of // files, returning one result per file with its replacement count. func (s *Sandbox) Replace(ctx context.Context, files []string, pattern, replacement string) ([]wire.Replaced, error) { diff --git a/sdk/go/find_test.go b/sdk/go/find_test.go index 2b5861a0..bf681b5f 100644 --- a/sdk/go/find_test.go +++ b/sdk/go/find_test.go @@ -33,6 +33,37 @@ func TestFindStreamsMatches(t *testing.T) { } } +func TestFindSeqStopsAfterTheCallerBreaks(t *testing.T) { + sb := fakeSandbox(t) + ctx := t.Context() + if err := sb.Mkdir(ctx, "/work", true); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := sb.WriteFile(ctx, "/work/many.txt", []byte(strings.Repeat("TODO\n", 500)), nil); err != nil { + t.Fatalf("write: %v", err) + } + + var seen []wire.Match + for m, err := range sb.FindSeq(ctx, "/work", "TODO", "") { + if err != nil { + t.Fatalf("FindSeq: %v", err) + } + if seen = append(seen, m); len(seen) == 3 { + break + } + } + if len(seen) != 3 || seen[2].Line != 3 { + t.Fatalf("seen %+v, want the first three lines", seen) + } + all, err := sb.Find(ctx, "/work", "TODO", "") + if err != nil { + t.Fatalf("Find after an abandoned stream: %v", err) + } + if len(all) != 500 { + t.Errorf("Find returned %d matches, want 500", len(all)) + } +} + func TestFindBadPatternIsTypedError(t *testing.T) { sb := fakeSandbox(t) _, err := sb.Find(t.Context(), "/", "(", "") diff --git a/sdk/go/utils.go b/sdk/go/utils.go index c334e78d..1c8bf948 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "iter" "fmt" "io" @@ -63,28 +64,46 @@ func oneShotRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire. // collectRPC sends req and gathers every streamed frame of type T until Done. func collectRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) ([]T, error) { - conn, done, err := s.call(ctx, req) - if err != nil { - return nil, err - } - defer done() var out []T - for { - resp, err := recv(ctx, conn) + for v, err := range streamRPC[T, PT](ctx, s, req) { if err != nil { return nil, err } - if v, ok := resp.(PT); ok { - out = append(out, *v) - continue + out = append(out, v) + } + return out, nil +} + +// streamRPC sends req and yields each streamed frame of type T until Done; breaking out closes the connection, which ends the guest-side producer. +func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) iter.Seq2[T, error] { + return func(yield func(T, error) bool) { + var zero T + conn, done, err := s.call(ctx, req) + if err != nil { + yield(zero, err) + return } - switch r := resp.(type) { - case *wire.Done: - return out, nil - case *wire.ErrorResp: - return nil, r - default: - return nil, unexpected(resp) + defer done() + for { + resp, err := recv(ctx, conn) + if err != nil { + yield(zero, err) + return + } + if v, ok := resp.(PT); ok { + if !yield(*v, nil) { + return + } + continue + } + switch r := resp.(type) { + case *wire.Done: + case *wire.ErrorResp: + yield(zero, r) + default: + yield(zero, unexpected(resp)) + } + return } } } diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index d536dde4..d0629673 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -157,9 +157,16 @@ def pull(self, path: str) -> bytes: return _drain_data(conn) def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: + return list(self.find_iter(path, pattern, glob)) + + def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: + """Yields matches as they stream; leaving the loop early closes the + connection, which ends the walk in the guest.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) - return [f for f in conn.recv_until("done") if f["type"] == "match"] + for f in conn.recv_until("done"): + if f["type"] == "match": + yield f def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict]: with self._dial() as conn: diff --git a/silkd/src/git.rs b/silkd/src/git.rs index effd09a8..a0e32250 100644 --- a/silkd/src/git.rs +++ b/silkd/src/git.rs @@ -12,6 +12,10 @@ use tokio::process::Command; use crate::proto::{self, ErrorKind, GitBranchOp, GitFileStatus, Response}; use crate::sysutil; +/// Estimated JSON bytes of file entries one status frame carries before it truncates; an eighth of the frame cap leaves room for escaping. +const STATUS_FILES_BYTES: usize = proto::MAX_FRAME / 8; +const STATUS_ENTRY_OVERHEAD: usize = 40; + /// Clones `url` into `path` (optionally a branch, shallow depth), then Done. pub async fn clone( w: &mut W, @@ -220,12 +224,19 @@ fn parse_status(text: &str) -> Response { let mut branch = String::new(); let (mut ahead, mut behind) = (0, 0); let mut files = Vec::new(); + let mut bytes = 0; + let mut truncated = false; for line in text.lines() { if let Some(rest) = line.strip_prefix("# branch.head ") { branch = rest.to_string(); } else if let Some(rest) = line.strip_prefix("# branch.ab ") { (ahead, behind) = parse_ahead_behind(rest); } else if let Some(f) = parse_file_line(line) { + bytes += f.path.len() + STATUS_ENTRY_OVERHEAD; + if bytes > STATUS_FILES_BYTES { + truncated = true; + break; + } files.push(f); } } @@ -234,6 +245,7 @@ fn parse_status(text: &str) -> Response { ahead, behind, files, + truncated, } } @@ -309,6 +321,25 @@ fn split_author(author: &str) -> (&str, &str) { mod tests { use super::*; + #[test] + fn status_truncates_past_the_frame_budget() { + let text: String = (0..200_000).map(|i| format!("? f{i}\n")).collect(); + let Response::GitStatusResult { + files, truncated, .. + } = parse_status(&text) + else { + panic!("status frame") + }; + assert!(truncated); + assert!(!files.is_empty() && files.len() < 200_000); + assert!(serde_json::to_vec(&parse_status(&text)).unwrap().len() < proto::MAX_FRAME); + + let Response::GitStatusResult { truncated, .. } = parse_status("? one\n") else { + panic!("status frame") + }; + assert!(!truncated); + } + #[test] fn parses_ordinary_rename_untracked_and_conflict() { let ordinary = parse_file_line("1 .M N... 100644 100644 100644 h1 h2 src/main.rs").unwrap(); diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index c3051513..9df8ea94 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -278,6 +278,8 @@ pub enum Response { ahead: u32, behind: u32, files: Vec, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + truncated: bool, }, GitCommitResult { hash: String, @@ -817,7 +819,7 @@ mod tests { seen += 1; } assert_eq!( - seen, 60, + seen, 61, "fixture corpus: adding a verb means adding its fixture" ); } From 1809217967ea42d292acd5d44c08c8fd23da79ed Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 13:14:20 +0800 Subject: [PATCH 2/2] sdk/go: sort the iter import --- sdk/go/utils.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 1c8bf948..00b943d0 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -2,9 +2,9 @@ package sandbox import ( "context" - "iter" "fmt" "io" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" "github.com/cocoonstack/sandbox/sdk/go/silkd"