git.status caps its frame; find streams to the caller

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 <dev@example.com>")
 sb.git_push("/work/repo", auth=token)     # egress lane only

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 <dev@example.com>")
 err  = sb.GitPush(ctx, "/work/repo", token)   // egress lane only

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/<language>` (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 |

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}

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" }

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)
 	}
 }
 

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) {

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(), "/", "(", "")

sdk/go/utils.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"iter"
 
 	"github.com/cocoonstack/sandbox/protocol/wire"
 	"github.com/cocoonstack/sandbox/sdk/go/silkd"
@@ -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
 		}
 	}
 }

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:

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: AsyncWrite + Unpin>(
     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();

silkd/src/proto.rs

@@ -278,6 +278,8 @@ pub enum Response {
         ahead: u32,
         behind: u32,
         files: Vec<GitFileStatus>,
+        #[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"
         );
     }