feat: chunked file transfer protocol

.github/workflows/go.yml

@@ -79,6 +79,50 @@ jobs:
         with:
           version: v2.6.1
 
+  # Protobuf checks run once per workflow execution, not per commit.
+  #
+  # buf breaking is deliberately not run: the wire format is still changing while
+  # the project is pre-1.0, and agent and client are released together. Add it
+  # once the protocol is stable.
+  protobuf:
+    name: Protobuf
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Go
+        uses: actions/setup-go@v7
+        with:
+          go-version-file: 'go.mod'
+
+      - uses: bufbuild/buf-setup-action@v1
+        with:
+          version: 1.47.2
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Lint
+        working-directory: protobuf
+        run: buf lint
+
+      - name: Format
+        working-directory: protobuf
+        run: buf format --diff --exit-code
+
+      # The generated Go code is committed, so it has to match the .proto files.
+      # Plugin versions come from go.mod rather than being pinned again here, so
+      # a dependency bump cannot silently desynchronise the two.
+      - name: Install codegen plugins
+        run: |
+          go install google.golang.org/protobuf/cmd/protoc-gen-go@"$(go list -m -f '{{.Version}}' google.golang.org/protobuf)"
+          go install connectrpc.com/connect/cmd/protoc-gen-connect-go@"$(go list -m -f '{{.Version}}' connectrpc.com/connect)"
+
+      - name: Generated code is up to date
+        working-directory: protobuf
+        run: |
+          buf generate
+          git diff --exit-code -- gen \
+            || { echo "::error::protobuf/gen is stale - run 'buf generate' in protobuf/ and commit the result"; exit 1; }
+
   # Build and test each commit for both architectures
   # Uses a matrix strategy to create separate job runs for each combination of commit and architecture
   per_commit_build_test:

cmds/dutagent/states.go

@@ -249,6 +249,16 @@ func executeModules(ctx context.Context, args runCmdArgs) (runCmdArgs, fsm.State
 		}
 
 		l.Info("all modules finished successfully")
+
+		// A module may start a file transfer that completes asynchronously (a
+		// download handed to SendFile is streamed by the broker workers after the
+		// module's Run returns). Signal graceful shutdown — which stops forwarding
+		// further module output but keeps the transfer flowing — and wait for every
+		// in-flight transfer to finish before cancelling the workers. abortTransfers
+		// is the backstop if the workers exit first (e.g. the client disconnects).
+		broker.Shutdown()
+		broker.WaitForTransfersToComplete()
+
 		modCtxCancel()
 		close(args.moduleErrCh)
 	}()

cmds/dutagent/states_test.go

@@ -700,3 +700,65 @@ func TestWaitModules(t *testing.T) {
 		})
 	}
 }
+
+// ctxCapturingModule records the context Run was handed.
+type ctxCapturingModule struct {
+	ctx context.Context //nolint:containedctx // captured for assertion, never used to call
+}
+
+func (m *ctxCapturingModule) Help() string                   { return "capture" }
+func (m *ctxCapturingModule) Init(_ context.Context) error   { return nil }
+func (m *ctxCapturingModule) Deinit(_ context.Context) error { return nil }
+
+func (m *ctxCapturingModule) Run(ctx context.Context, _ module.Session, _ ...string) error {
+	m.ctx = ctx
+
+	return nil
+}
+
+// TestModuleContextOutlivesBrokerCancel pins where the module's context comes
+// from. File transfers are bounded by it and outlive Run, so it must derive from
+// the RPC context, not the broker's. Deriving it from modCtx looks like a
+// tidy-up and passes every other test: downloads would be truncated at
+// modCtxCancel and the run would still report success.
+func TestModuleContextOutlivesBrokerCancel(t *testing.T) {
+	mod := &ctxCapturingModule{}
+
+	wrap := dut.Module{}
+	wrap.Config.Name = "captureMod"
+	wrap.Config.Passthrough = true
+	wrap.Module = mod
+
+	moduleErrCh := make(chan error, 1)
+
+	args := runCmdArgs{
+		stream:      &fakes.FakeStream{},
+		cmdMsg:      &pb.Command{Device: "devX", Command: "cmdY"},
+		cmd:         dut.Command{Modules: []dut.Module{wrap}},
+		moduleErrCh: moduleErrCh,
+	}
+
+	_, _, err := executeModules(context.Background(), args)
+	if err != nil {
+		t.Fatalf("executeModules: %v", err)
+	}
+
+	// Closed channel means the run goroutine finished, so modCtxCancel has run.
+	select {
+	case _, ok := <-moduleErrCh:
+		if ok {
+			t.Fatal("module reported an error")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("module execution did not finish")
+	}
+
+	if mod.ctx == nil {
+		t.Fatal("module was never run")
+	}
+
+	if err := mod.ctx.Err(); err != nil {
+		t.Errorf("module context cancelled by broker teardown: %v — "+
+			"a transfer still streaming after Run returned would be aborted", err)
+	}
+}