Repository: open-telemetry/opamp-go
Pull request #76 (2022-05-17 17:14:50)

Implement package syncing
- This implements "Packages" spec section https://github.com/open-telemetry/opamp-spec/blob/main/specification.md#packages-1
- TODO: increase test coverage to handle many more edge cases.
- TODO: add package usage to the example Agent/Server.

codecov[bot]: # [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) Report
> Merging [#76](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (6946be7) into [main](https://codecov.io/gh/open-telemetry/opamp-go/commit/888d3adc29d859534099b0807b7d9cd8bafd772e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (888d3ad) will **decrease** coverage by `0.60%`.
> The diff coverage is `72.59%`.

```diff
@@            Coverage Diff             @@
##             main      #76      +/-   ##
==========================================
- Coverage   78.17%   77.57%   -0.61%     
==========================================
  Files          20       22       +2     
  Lines        1306     1717     +411     
==========================================
+ Hits         1021     1332     +311     
- Misses        215      289      +74     
- Partials       70       96      +26     
```


| [Impacted Files](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | Coverage Δ | |
|---|---|---|
| [client/internal/packagessyncer.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL3BhY2thZ2Vzc3luY2VyLmdv) | `59.25% <59.25%> (ø)` | |
| [client/internal/clientcommon.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2NsaWVudGNvbW1vbi5nbw==) | `82.66% <75.00%> (-2.19%)` | :arrow_down: |
| [client/internal/mockserver.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL21vY2tzZXJ2ZXIuZ28=) | `80.64% <75.75%> (-0.87%)` | :arrow_down: |
| [client/internal/inmempackagestore.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2lubWVtcGFja2FnZXN0b3JlLmdv) | `76.47% <76.47%> (ø)` | |
| [client/internal/clientstate.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2NsaWVudHN0YXRlLmdv) | `81.57% <92.10%> (+10.52%)` | :arrow_up: |
| [client/httpclient.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2h0dHBjbGllbnQuZ28=) | `93.75% <100.00%> (+0.89%)` | :arrow_up: |
| [client/internal/httpsender.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2h0dHBzZW5kZXIuZ28=) | `81.20% <100.00%> (+5.01%)` | :arrow_up: |
| [client/internal/receivedprocessor.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL3JlY2VpdmVkcHJvY2Vzc29yLmdv) | `86.40% <100.00%> (+2.49%)` | :arrow_up: |
| [client/internal/wsreceiver.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL3dzcmVjZWl2ZXIuZ28=) | `91.42% <100.00%> (ø)` | |
| [client/wsclient.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L3dzY2xpZW50Lmdv) | `84.07% <100.00%> (+0.74%)` | :arrow_up: |
| ... and [5 more](https://codecov.io/gh/open-telemetry/opamp-go/pull/76/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Last update [888d3ad...6946be7](https://codecov.io/gh/open-telemetry/opamp-go/pull/76?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).

andykellr: Aside from the small issue I mentioned, overall this looks good. However, unless I missed something, it appears to me that it is possible to have multiple sync operations running at the same time which could be confusing. 
tigrannajaryan: >However, unless I missed something, it appears to me that it is possible to have multiple sync operations running at the same time which could be confusing.

Good point. I will create an issue to fix this in a future PR. 

Patch:
diff --git a/client/httpclient.go b/client/httpclient.go
index b32f691..a74b0f9 100644
--- a/client/httpclient.go
+++ b/client/httpclient.go
@@ -76,5 +76,11 @@ func (c *httpClient) runUntilStopped(ctx context.Context) {
 	// Start the HTTP sender. This will make request/responses with retries for
 	// failures and will wait with configured polling interval if there is nothing
 	// to send.
-	c.sender.Run(ctx, c.opAMPServerURL, c.common.Callbacks, &c.common.ClientSyncedState)
+	c.sender.Run(
+		ctx,
+		c.opAMPServerURL,
+		c.common.Callbacks,
+		&c.common.ClientSyncedState,
+		c.common.PackagesStateProvider,
+	)
 }
diff --git a/client/internal/clientcommon.go b/client/internal/clientcommon.go
index bf7e719..aa3d22b 100644
--- a/client/internal/clientcommon.go
+++ b/client/internal/clientcommon.go
@@ -10,14 +10,15 @@ import (
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/protobufs"
+	"google.golang.org/protobuf/proto"
 )
 
 var (
 	ErrAgentDescriptionMissing      = errors.New("AgentDescription is nil")
 	ErrAgentDescriptionNoAttributes = errors.New("AgentDescription has no attributes defined")
-	errRemoteConfigStatusMissing    = errors.New("RemoteConfigStatus is not set")
-	errAlreadyStarted               = errors.New("already started")
-	errCannotStopNotStarted         = errors.New("cannot stop because not started")
+
+	errAlreadyStarted       = errors.New("already started")
+	errCannotStopNotStarted = errors.New("cannot stop because not started")
 )
 
 // ClientCommon contains the OpAMP logic that is common between WebSocket and
@@ -29,6 +30,9 @@ type ClientCommon struct {
 	// Client state storage. This is needed if the Server asks to report the state.
 	ClientSyncedState ClientSyncedState
 
+	// PackagesStateProvider provides access to the local state of packages.
+	PackagesStateProvider types.PackagesStateProvider
+
 	// The transport-specific sender.
 	sender Sender
 
@@ -59,6 +63,7 @@ func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSetti
 		return ErrAgentDescriptionMissing
 	}
 
+	// Prepare remote config status.
 	if settings.RemoteConfigStatus == nil {
 		// RemoteConfigStatus is not provided. Start with empty.
 		settings.RemoteConfigStatus = &protobufs.RemoteConfigStatus{
@@ -70,6 +75,27 @@ func (c *ClientCommon) PrepareStart(_ context.Context, settings types.StartSetti
 		return err
 	}
 
+	// Prepare package statuses.
+	c.PackagesStateProvider = settings.PackagesStateProvider
+	var packageStatuses *protobufs.PackageStatuses
+	if settings.PackagesStateProvider != nil {
+		// Set package status from the value previously saved in the PackagesStateProvider.
+		var err error
+		packageStatuses, err = settings.PackagesStateProvider.LastReportedStatuses()
+		if err != nil {
+			return err
+		}
+	}
+
+	if packageStatuses == nil {
+		// PackageStatuses is not provided. Start with empty.
+		packageStatuses = &protobufs.PackageStatuses{}
+	}
+	if err := c.ClientSyncedState.SetPackageStatuses(packageStatuses); err != nil {
+		return err
+	}
+
+	// Prepare callbacks.
 	c.Callbacks = settings.Callbacks
 	if c.Callbacks == nil {
 		// Make sure it is always safe to call Callbacks.
@@ -154,24 +180,24 @@ func (c *ClientCommon) PrepareFirstMessage(ctx context.Context) error {
 				msg.StatusReport = &protobufs.StatusReport{}
 			}
 			msg.StatusReport.AgentDescription = c.ClientSyncedState.AgentDescription()
-
 			msg.StatusReport.EffectiveConfig = cfg
-
 			msg.StatusReport.RemoteConfigStatus = c.ClientSyncedState.RemoteConfigStatus()
+			msg.PackageStatuses = c.ClientSyncedState.PackageStatuses()
 
-			if msg.PackageStatuses == nil {
-				msg.PackageStatuses = &protobufs.PackageStatuses{}
+			if c.PackagesStateProvider != nil {
+				// We have a state provider, so package related capabilities can work.
+				msg.StatusReport.Capabilities |= protobufs.AgentCapabilities_AcceptsPackages
+				msg.StatusReport.Capabilities |= protobufs.AgentCapabilities_ReportsPackageStatuses
 			}
-
-			// TODO: set PackageStatuses.ServerProvidedAllPackagesHash field and
-			// handle the Hashes for PackageStatuses properly.
 		},
 	)
 	return nil
 }
 
+// AgentDescription returns the current state of the AgentDescription.
 func (c *ClientCommon) AgentDescription() *protobufs.AgentDescription {
-	return c.ClientSyncedState.AgentDescription()
+	// Return a cloned copy to allow caller to do whatever they want with the result.
+	return proto.Clone(c.ClientSyncedState.AgentDescription()).(*protobufs.AgentDescription)
 }
 
 // SetAgentDescription sends a status update to the Server with the new AgentDescription
@@ -183,7 +209,7 @@ func (c *ClientCommon) SetAgentDescription(descr *protobufs.AgentDescription) er
 		return err
 	}
 	c.sender.NextMessage().UpdateStatus(func(statusReport *protobufs.StatusReport) {
-		statusReport.AgentDescription = descr
+		statusReport.AgentDescription = c.ClientSyncedState.AgentDescription()
 	})
 	c.sender.ScheduleSend()
 	return nil
diff --git a/client/internal/clientstate.go b/client/internal/clientstate.go
index b6c718f..be74d56 100644
--- a/client/internal/clientstate.go
+++ b/client/internal/clientstate.go
@@ -2,7 +2,9 @@ package internal
 
 import (
 	"crypto/sha256"
+	"errors"
 	"hash"
+	"sort"
 	"sync"
 
 	"github.com/open-telemetry/opamp-go/protobufs"
@@ -10,6 +12,11 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+var (
+	errRemoteConfigStatusMissing = errors.New("RemoteConfigStatus is not set")
+	errPackageStatusesMissing    = errors.New("PackageStatuses is not set")
+)
+
 // ClientSyncedState stores the state of the Agent messages that the OpAMP Client needs to
 // have access to synchronize to the Server. 3 messages can be stored in this store:
 // AgentDescription, RemoteConfigStatus and PackageStatuses.
@@ -135,3 +142,49 @@ func (s *ClientSyncedState) SetRemoteConfigStatus(status *protobufs.RemoteConfig
 
 	return nil
 }
+
+func calcHashPackageStatuses(status *protobufs.PackageStatuses) {
+	h := sha256.New()
+
+	// Convert package names to slice to sort it and make sure hash calculation is
+	// deterministic.
+
+	var names []string
+	for name := range status.Packages {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	for _, name := range names {
+		pkg := status.Packages[name]
+		h.Write([]byte(name))
+		h.Write([]byte(pkg.AgentHasVersion))
+		h.Write(pkg.AgentHasHash)
+		h.Write([]byte(pkg.ServerOfferedVersion))
+		h.Write(pkg.ServerOfferedHash)
+		h.Write([]byte(pkg.Status.String()))
+		h.Write([]byte(pkg.ErrorMessage))
+	}
+
+	h.Write(status.ServerProvidedAllPackagesHash)
+
+	status.Hash = h.Sum(nil)
+}
+
+// SetPackageStatuses sets the PackageStatuses in the state.
+// Will calculate the Hash from the content of the other fields.
+func (s *ClientSyncedState) SetPackageStatuses(status *protobufs.PackageStatuses) error {
+	if status == nil {
+		return errPackageStatusesMissing
+	}
+
+	calcHashPackageStatuses(status)
+
+	clone := proto.Clone(status).(*protobufs.PackageStatuses)
+
+	defer s.mutex.Unlock()
+	s.mutex.Lock()
+	s.packageStatuses = clone
+
+	return nil
+}
diff --git a/client/internal/httpsender.go b/client/internal/httpsender.go
index ec4ad68..454d054 100644
--- a/client/internal/httpsender.go
+++ b/client/internal/httpsender.go
@@ -59,10 +59,16 @@ func NewHTTPSender(logger types.Logger) *HTTPSender {
 // Should not be called concurrently with itself. Can be called concurrently with
 // modifying NextMessage().
 // Run continues until ctx is cancelled.
-func (h *HTTPSender) Run(ctx context.Context, url string, callbacks types.Callbacks, clientSyncedState *ClientSyncedState) {
+func (h *HTTPSender) Run(
+	ctx context.Context,
+	url string,
+	callbacks types.Callbacks,
+	clientSyncedState *ClientSyncedState,
+	packagesStateProvider types.PackagesStateProvider,
+) {
 	h.url = url
 	h.callbacks = callbacks
-	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState)
+	h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider)
 
 	for {
 		pollingTimer := time.NewTimer(time.Millisecond * time.Duration(atomic.LoadInt64(&h.pollingIntervalMs)))
@@ -101,6 +107,10 @@ func (h *HTTPSender) makeOneRequestRoundtrip(ctx context.Context) {
 	if err != nil {
 		return
 	}
+	if resp == nil {
+		// No request was sent and nothing to receive.
+		return
+	}
 	h.receiveResponse(ctx, resp)
 }
 
@@ -114,6 +124,10 @@ func (h *HTTPSender) sendRequestWithRetries(ctx context.Context) (*http.Response
 		}
 		return nil, err
 	}
+	if req == nil {
+		// Nothing to send.
+		return nil, nil
+	}
 
 	// Repeatedly try requests with a backoff strategy.
 	infiniteBackoff := backoff.NewExponentialBackOff()
@@ -174,7 +188,6 @@ func recalculateInterval(interval time.Duration, resp *http.Response) time.Durat
 
 func (h *HTTPSender) prepareRequest(ctx context.Context) (*http.Request, error) {
 	msgToSend := h.nextMessage.PopPending()
-
 	if msgToSend == nil || proto.Equal(msgToSend, &protobufs.AgentToServer{}) {
 		// There is no pending message or the message is empty.
 		// Nothing to send.
diff --git a/client/internal/inmempackagestore.go b/client/internal/inmempackagestore.go
new file mode 100644
index 0000000..ae995ca
--- /dev/null
+++ b/client/internal/inmempackagestore.go
@@ -0,0 +1,97 @@
+package internal
+
+import (
+	"context"
+	"io"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+// InMemPackagesStore is a package store used for testing. Keeps the packages in memory.
+type InMemPackagesStore struct {
+	allPackagesHash      []byte
+	pkgState             map[string]types.PackageState
+	fileContents         map[string][]byte
+	fileHashes           map[string][]byte
+	lastReportedStatuses *protobufs.PackageStatuses
+}
+
+var _ types.PackagesStateProvider = (*InMemPackagesStore)(nil)
+
+func NewInMemPackagesStore() *InMemPackagesStore {
+	return &InMemPackagesStore{
+		fileContents: map[string][]byte{},
+		fileHashes:   map[string][]byte{},
+		pkgState:     map[string]types.PackageState{},
+	}
+}
+
+func (l *InMemPackagesStore) AllPackagesHash() ([]byte, error) {
+	return l.allPackagesHash, nil
+}
+
+func (l *InMemPackagesStore) Packages() ([]string, error) {
+	var names []string
+	for k := range l.pkgState {
+		names = append(names, k)
+	}
+	return names, nil
+}
+
+func (l *InMemPackagesStore) PackageState(packageName string) (state types.PackageState, err error) {
+	if pkg, ok := l.pkgState[packageName]; ok {
+		return pkg, nil
+	}
+	return types.PackageState{Exists: false}, nil
+}
+
+func (l *InMemPackagesStore) CreatePackage(packageName string, typ protobufs.PackageAvailable_PackageType) error {
+	l.pkgState[packageName] = types.PackageState{
+		Exists: true,
+		Type:   typ,
+	}
+	return nil
+}
+
+func (l *InMemPackagesStore) FileContentHash(packageName string) ([]byte, error) {
+	return l.fileHashes[packageName], nil
+}
+
+func (l *InMemPackagesStore) UpdateContent(_ context.Context, packageName string, data io.Reader, contentHash []byte) error {
+	b, err := io.ReadAll(data)
+	if err != nil {
+		return err
+	}
+	l.fileContents[packageName] = b
+	l.fileHashes[packageName] = contentHash
+	return nil
+}
+
+func (l *InMemPackagesStore) SetPackageState(packageName string, state types.PackageState) error {
+	l.pkgState[packageName] = state
+	return nil
+}
+
+func (l *InMemPackagesStore) DeletePackage(packageName string) error {
+	delete(l.pkgState, packageName)
+	return nil
+}
+
+func (l *InMemPackagesStore) SetAllPackagesHash(hash []byte) error {
+	l.allPackagesHash = hash
+	return nil
+}
+
+func (l *InMemPackagesStore) GetContent() map[string][]byte {
+	return l.fileContents
+}
+
+func (l *InMemPackagesStore) LastReportedStatuses() (*protobufs.PackageStatuses, error) {
+	return l.lastReportedStatuses, nil
+}
+
+func (l *InMemPackagesStore) SetLastReportedStatuses(statuses *protobufs.PackageStatuses) error {
+	l.lastReportedStatuses = statuses
+	return nil
+}
diff --git a/client/internal/mockserver.go b/client/internal/mockserver.go
index 6997651..274cea1 100644
--- a/client/internal/mockserver.go
+++ b/client/internal/mockserver.go
@@ -29,6 +29,7 @@ type MockServer struct {
 	srv         *httptest.Server
 
 	expectedHandlers chan receivedMessageHandler
+	expectedComplete chan struct{}
 	isExpectMode     bool
 }
 
@@ -41,6 +42,7 @@ func StartMockServer(t *testing.T) *MockServer {
 	srv := &MockServer{
 		t:                t,
 		expectedHandlers: make(chan receivedMessageHandler, 0),
+		expectedComplete: make(chan struct{}, 0),
 	}
 
 	m := http.NewServeMux()
@@ -144,6 +146,7 @@ func (m *MockServer) handleReceivedBytes(msgBytes []byte, alwaysRespond bool) []
 		t := time.NewTimer(5 * time.Second)
 		select {
 		case h := <-m.expectedHandlers:
+			defer func() { m.expectedComplete <- struct{}{} }()
 			response = h(&request)
 		case <-t.C:
 			m.t.Error("Time out waiting for Expect() to handle the received message")
@@ -182,11 +185,58 @@ func (m *MockServer) Expect(handler receivedMessageHandler) {
 		// push the handler to the channel.
 		// the handler will be fetched and called by handleReceivedBytes() when
 		// message is received.
+		<-m.expectedComplete
+
 	case <-t.C:
 		m.t.Error("Time out waiting to receive a message from the client")
 	}
 }
 
+// EventuallyExpect expects to receive a message and calls the handler for every
+// received message until eventually the handler returns true for the second
+// element of the return tuple.
+// Typically used when we know we expect to receive a particular message but 0 or more
+// other messages may be received before that.
+func (m *MockServer) EventuallyExpect(
+	msg string,
+	handler func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool),
+) {
+	t := time.NewTimer(5 * time.Second)
+
+	conditionCh := make(chan bool)
+	wrappedHandler := func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+		response, condition := handler(msg)
+		conditionCh <- condition
+		return response
+	}
+
+	for {
+		select {
+		case m.expectedHandlers <- wrappedHandler:
+			// push the handler to the channel.
+			// the handler will be fetched and called by handleReceivedBytes() when
+			// message is received.
+
+			select {
+			case condition := <-conditionCh:
+				<-m.expectedComplete
+				if condition {
+					return
+				}
+			case <-t.C:
+				m.t.Errorf("Time out expecting a message from the client: %v", msg)
+				<-conditionCh
+				<-m.expectedComplete
+				return
+			}
+
+		case <-t.C:
+			m.t.Errorf("Time out expecting a message from the client: %v", msg)
+			return
+		}
+	}
+}
+
 func (m *MockServer) Close() {
 	close(m.expectedHandlers)
 	m.srv.Close()
diff --git a/client/internal/packagessyncer.go b/client/internal/packagessyncer.go
new file mode 100644
index 0000000..422c217
--- /dev/null
+++ b/client/internal/packagessyncer.go
@@ -0,0 +1,327 @@
+package internal
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+// packagesSyncer performs the package syncing process.
+type packagesSyncer struct {
+	logger            types.Logger
+	available         *protobufs.PackagesAvailable
+	clientSyncedState *ClientSyncedState
+	localState        types.PackagesStateProvider
+	sender            Sender
+
+	statuses *protobufs.PackageStatuses
+	doneCh   chan struct{}
+}
+
+func NewPackagesSyncer(
+	logger types.Logger,
+	available *protobufs.PackagesAvailable,
+	sender Sender,
+	clientSyncedState *ClientSyncedState,
+	packagesStateProvider types.PackagesStateProvider,
+) *packagesSyncer {
+	return &packagesSyncer{
+		logger:            logger,
+		available:         available,
+		sender:            sender,
+		clientSyncedState: clientSyncedState,
+		localState:        packagesStateProvider,
+		doneCh:            make(chan struct{}),
+	}
+}
+
+func (s *packagesSyncer) Sync(ctx context.Context) error {
+
+	defer func() {
+		close(s.doneCh)
+	}()
+
+	// Prepare package statuses.
+	if err := s.initStatuses(); err != nil {
+		return err
+	}
+
+	if err := s.clientSyncedState.SetPackageStatuses(s.statuses); err != nil {
+		return err
+	}
+
+	// Now do the actual syncing in the background.
+	go s.doSync(ctx)
+
+	return nil
+}
+
+func (s *packagesSyncer) Done() <-chan struct{} {
+	return s.doneCh
+}
+
+// initStatuses initializes the "statuses" field from the current "localState".
+// The "statuses" will be updated as the syncing progresses and will be
+// periodically reported to the Server.
+func (s *packagesSyncer) initStatuses() error {
+	if s.localState == nil {
+		return errors.New("cannot sync packages because PackagesStateProvider is not provided")
+	}
+
+	// Restore statuses that were previously stored in the local state.
+	var err error
+	s.statuses, err = s.localState.LastReportedStatuses()
+	if err != nil {
+		return err
+	}
+	if s.statuses == nil {
+		// No statuses are stored locally (maybe first time), just start with empty.
+		s.statuses = &protobufs.PackageStatuses{}
+	}
+
+	if s.statuses.Packages == nil {
+		s.statuses.Packages = map[string]*protobufs.PackageStatus{}
+	}
+
+	// Report to the Server the "all" hash that we received from the Server so that
+	// the Server knows we are processing the right offer.
+	s.statuses.ServerProvidedAllPackagesHash = s.available.AllPackagesHash
+
+	return nil
+}
+
+func (s *packagesSyncer) doSync(ctx context.Context) {
+	hash, err := s.localState.AllPackagesHash()
+	if err != nil {
+		s.logger.Errorf("Package syncing failed: %V", err)
+		return
+	}
+	if bytes.Compare(hash, s.available.AllPackagesHash) == 0 {
+		s.logger.Debugf("All packages are already up to date.")
+		return
+	}
+
+	failed := false
+	if err := s.deleteUnneededLocalPackages(); err != nil {
+		s.logger.Errorf("Cannot delete unneeded packages: %v", err)
+		failed = true
+	}
+
+	// Iterate through offered packages and sync them all from server.
+	for name, pkg := range s.available.Packages {
+		err := s.syncPackage(ctx, name, pkg)
+		if err != nil {
+			s.logger.Errorf("Cannot sync package %s: %v", name, err)
+			failed = true
+		}
+	}
+
+	if !failed {
+		// Update the "all" hash on success, so that next time Sync() does not thing,
+		// unless a new hash is received from the Server.
+		if err := s.localState.SetAllPackagesHash(s.available.AllPackagesHash); err != nil {
+			s.logger.Errorf("SetAllPackagesHash failed: %v", err)
+		} else {
+			s.logger.Debugf("All packages are synced and up to date.")
+		}
+	} else {
+		s.logger.Errorf("Package syncing was not successful.")
+	}
+
+	_ = s.reportStatuses(true)
+}
+
+func (s *packagesSyncer) syncPackage(
+	ctx context.Context,
+	pkgName string,
+	pkgAvail *protobufs.PackageAvailable,
+) error {
+
+	status := s.statuses.Packages[pkgName]
+	if status == nil {
+		// This package has no status. Create one.
+		status = &protobufs.PackageStatus{
+			Name:                 pkgName,
+			ServerOfferedVersion: pkgAvail.Version,
+			ServerOfferedHash:    pkgAvail.Hash,
+		}
+		s.statuses.Packages[pkgName] = status
+	}
+
+	pkgLocal, err := s.localState.PackageState(pkgName)
+	if err != nil {
+		return err
+	}
+
+	mustCreate := !pkgLocal.Exists
+	if pkgLocal.Exists {
+		if bytes.Equal(pkgLocal.Hash, pkgAvail.Hash) {
+			s.logger.Debugf("Package %s hash is unchanged, skipping", pkgName)
+			return nil
+		}
+		if pkgLocal.Type != pkgAvail.Type {
+			// Package is of wrong type. Need to re-create it. So, delete it.
+			if err := s.localState.DeletePackage(pkgName); err != nil {
+				err = fmt.Errorf("cannot delete existing version of package %s: %v", pkgName, err)
+				status.Status = protobufs.PackageStatus_InstallFailed
+				status.ErrorMessage = err.Error()
+				return err
+			}
+			// And mark that it needs to be created.
+			mustCreate = true
+		}
+	}
+
+	// Report that we are beginning to install it.
+	status.Status = protobufs.PackageStatus_Installing
+	_ = s.reportStatuses(true)
+
+	if mustCreate {
+		// Make sure the package exists.
+		err = s.localState.CreatePackage(pkgName, pkgAvail.Type)
+		if err != nil {
+			err = fmt.Errorf("cannot create package %s: %v", pkgName, err)
+			status.Status = protobufs.PackageStatus_InstallFailed
+			status.ErrorMessage = err.Error()
+			return err
+		}
+	}
+
+	// Sync package file: ensure it exists or download it.
+	err = s.syncPackageFile(ctx, pkgName, pkgAvail.File)
+	if err == nil {
+		// Only save the state on success, so that next sync does not retry this package.
+		pkgLocal.Hash = pkgAvail.Hash
+		pkgLocal.Version = pkgAvail.Version
+		if err := s.localState.SetPackageState(pkgName, pkgLocal); err == nil {
+			status.Status = protobufs.PackageStatus_Installed
+			status.AgentHasHash = pkgAvail.Hash
+			status.AgentHasVersion = pkgAvail.Version
+		}
+	}
+
+	if err != nil {
+		status.Status = protobufs.PackageStatus_InstallFailed
+		status.ErrorMessage = err.Error()
+	}
+	_ = s.reportStatuses(true)
+
+	return err
+}
+
+func (s *packagesSyncer) syncPackageFile(
+	ctx context.Context, pkgName string, file *protobufs.DownloadableFile,
+) error {
+	shouldDownload, err := s.shouldDownloadFile(pkgName, file)
+	if err == nil && shouldDownload {
+		err = s.downloadFile(ctx, pkgName, file)
+	}
+
+	return err
+}
+
+func (s *packagesSyncer) shouldDownloadFile(
+	packageName string,
+	file *protobufs.DownloadableFile,
+) (bool, error) {
+	fileContentHash, err := s.localState.FileContentHash(packageName)
+
+	if err != nil {
+		err := fmt.Errorf("cannot calculate checksum of %s: %v", packageName, err)
+		s.logger.Errorf(err.Error())
+		return true, nil
+	} else {
+		// Compare the checksum of the file we have with what
+		// we are offered by the server.
+		if bytes.Compare(fileContentHash, file.ContentHash) != 0 {
+			s.logger.Debugf("Package %s: file hash mismatch, will download.", packageName)
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func (s *packagesSyncer) downloadFile(ctx context.Context, pkgName string, file *protobufs.DownloadableFile) error {
+	s.logger.Debugf("Downloading package %s file from %s", pkgName, file.DownloadUrl)
+
+	req, err := http.NewRequestWithContext(ctx, "GET", file.DownloadUrl, nil)
+	if err != nil {
+		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("cannot download file from %s, HTTP response=%v", file.DownloadUrl, resp.StatusCode)
+	}
+
+	// TODO: either add a callback to verify file.Signature or pass the Signature
+	// as a parameter to UpdateContent.
+
+	err = s.localState.UpdateContent(ctx, pkgName, resp.Body, file.ContentHash)
+	if err != nil {
+		return fmt.Errorf("cannot download file from %s: %v", file.DownloadUrl, err)
+	}
+	return nil
+}
+
+func (s *packagesSyncer) deleteUnneededLocalPackages() error {
+	// Read the list of packages we have locally.
+	localPackages, err := s.localState.Packages()
+	if err != nil {
+		return err
+	}
+
+	var lastErr error
+	for _, localPkg := range localPackages {
+		// Do we have a package that is not offered?
+		if _, offered := s.available.Packages[localPkg]; !offered {
+			s.logger.Debugf("Package %s is no longer needed, deleting.", localPkg)
+			err := s.localState.DeletePackage(localPkg)
+			if err != nil {
+				lastErr = err
+			}
+		}
+	}
+
+	// Also remove packages that were not offered from the statuses.
+	for name := range s.statuses.Packages {
+		if _, offered := s.available.Packages[name]; !offered {
+			delete(s.statuses.Packages, name)
+		}
+	}
+
+	return lastErr
+}
+
+func (s *packagesSyncer) reportStatuses(sendImmediately bool) error {
+	// Save it in the user-supplied state provider.
+	if err := s.localState.SetLastReportedStatuses(s.statuses); err != nil {
+		s.logger.Errorf("Cannot save last reported statuses: %v", err)
+		return err
+	}
+
+	// Also save it in our internal state (will be needed if the Server asks for it).
+	if err := s.clientSyncedState.SetPackageStatuses(s.statuses); err != nil {
+		s.logger.Errorf("Cannot save client state: %v", err)
+		return err
+	}
+	s.sender.NextMessage().Update(
+		func(msg *protobufs.AgentToServer) {
+			msg.PackageStatuses = s.clientSyncedState.PackageStatuses()
+		})
+
+	if sendImmediately {
+		s.sender.ScheduleSend()
+	}
+	return nil
+}
diff --git a/client/internal/receivedprocessor.go b/client/internal/receivedprocessor.go
index f97a6ca..6b334f2 100644
--- a/client/internal/receivedprocessor.go
+++ b/client/internal/receivedprocessor.go
@@ -21,6 +21,8 @@ type receivedProcessor struct {
 
 	// Client state storage. This is needed if the Server asks to report the state.
 	clientSyncedState *ClientSyncedState
+
+	packagesStateProvider types.PackagesStateProvider
 }
 
 func newReceivedProcessor(
@@ -28,12 +30,14 @@ func newReceivedProcessor(
 	callbacks types.Callbacks,
 	sender Sender,
 	clientSyncedState *ClientSyncedState,
+	packagesStateProvider types.PackagesStateProvider,
 ) receivedProcessor {
 	return receivedProcessor{
-		logger:            logger,
-		callbacks:         callbacks,
-		sender:            sender,
-		clientSyncedState: clientSyncedState,
+		logger:                logger,
+		callbacks:             callbacks,
+		sender:                sender,
+		clientSyncedState:     clientSyncedState,
+		packagesStateProvider: packagesStateProvider,
 	}
 }
 
@@ -51,7 +55,7 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
 		scheduled := r.rcvRemoteConfig(ctx, msg.RemoteConfig)
 
 		r.rcvConnectionSettings(ctx, msg.ConnectionSettings)
-		r.rcvPackagesAvailable(msg.PackagesAvailable)
+		scheduled = scheduled || r.rcvPackagesAvailable(ctx, msg.PackagesAvailable)
 		r.rcvAgentIdentification(ctx, msg.AgentIdentification)
 
 		scheduled2, err := r.rcvFlags(ctx, msg.Flags)
@@ -244,8 +248,43 @@ func (r *receivedProcessor) processErrorResponse(body *protobufs.ServerErrorResp
 	r.logger.Errorf("received an error from server: %s", body.ErrorMessage)
 }
 
-func (r *receivedProcessor) rcvPackagesAvailable(packages *protobufs.PackagesAvailable) {
-	// TODO: implement this.
+func (r *receivedProcessor) rcvPackagesAvailable(ctx context.Context, available *protobufs.PackagesAvailable) bool {
+	if available == nil {
+		return false
+	}
+
+	syncer := NewPackagesSyncer(r.logger, available, r.sender, r.clientSyncedState, r.packagesStateProvider)
+
+	// The OnPackagesAvailable is expected to call syncer.Sync() which will
+	// do the syncing process in background.
+	err := r.callbacks.OnPackagesAvailable(ctx, available, syncer)
+	if err != nil {
+		// Could not even start syncing. Just report everything failed.
+		statuses := &protobufs.PackageStatuses{
+			ServerProvidedAllPackagesHash: available.AllPackagesHash,
+			Packages:                      map[string]*protobufs.PackageStatus{},
+			// TODO: we need a way to report general errors that occurred when trying
+			// install available packages, errors which are not related to a particular
+			// package.
+		}
+		// Until the TODO above is done we will set the same error message for all
+		// packages.
+		for name, pkg := range available.Packages {
+			statuses.Packages[name] = &protobufs.PackageStatus{
+				Name:                 name,
+				ServerOfferedVersion: pkg.Version,
+				ServerOfferedHash:    pkg.Hash,
+				Status:               protobufs.PackageStatus_InstallFailed,
+				ErrorMessage:         err.Error(),
+			}
+		}
+		_ = r.clientSyncedState.SetPackageStatuses(statuses)
+		r.sender.NextMessage().Update(
+			func(msg *protobufs.AgentToServer) { msg.PackageStatuses = statuses },
+		)
+	}
+
+	return true
 }
 
 func (r *receivedProcessor) rcvAgentIdentification(ctx context.Context, agentId *protobufs.AgentIdentification) {
diff --git a/client/internal/wsreceiver.go b/client/internal/wsreceiver.go
index 5103bc3..08bed68 100644
--- a/client/internal/wsreceiver.go
+++ b/client/internal/wsreceiver.go
@@ -26,13 +26,14 @@ func NewWSReceiver(
 	conn *websocket.Conn,
 	sender *WSSender,
 	clientSyncedState *ClientSyncedState,
+	packagesStateProvider types.PackagesStateProvider,
 ) *wsReceiver {
 	w := &wsReceiver{
 		conn:      conn,
 		logger:    logger,
 		sender:    sender,
 		callbacks: callbacks,
-		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState),
+		processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider),
 	}
 
 	return w
diff --git a/client/types/packagessyncer.go b/client/types/packagessyncer.go
index 04b0b0f..4a6e1b8 100644
--- a/client/types/packagessyncer.go
+++ b/client/types/packagessyncer.go
@@ -3,6 +3,8 @@ package types
 import (
 	"context"
 	"io"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
 // PackagesSyncer can be used by the Agent to initiate syncing a package from the Server.
@@ -10,27 +12,62 @@ import (
 // the particular PackageAvailable message the OnPackageAvailable callback was called for.
 type PackagesSyncer interface {
 	// Sync the available package from the Server to the Agent.
-	// The Agent must supply an PackagesStateProvider to let the Sync function
-	// know what is available locally, what data needs to be sync and how the
+	// The Agent must supply an PackagesStateProvider in StartSettings to let the Sync
+	// function know what is available locally, what data needs to be synced and how the
 	// data can be stored locally.
-	Sync(ctx context.Context, localState PackagesStateProvider) error
+	// Sync typically returns immediately and continues working in the background,
+	// downloading the packages and applying the changes to the local state.
+	// Sync should be called once only.
+	Sync(ctx context.Context) error
+
+	// Done returns a channel which is readable when the Sync is complete.
+	Done() <-chan struct{}
+}
+
+type PackageState struct {
+	// Exists indicates that the package exists locally. The rest of the fields
+	// must be ignored if this field is false.
+	Exists bool
+
+	Type    protobufs.PackageAvailable_PackageType
+	Hash    []byte
+	Version string
 }
 
+// PackagesStateProvider is an interface that is used by PackagesSyncer.Sync() to
+// query and update the Agent's local state of packages.
+// It is recommended that the local state is stored persistently so that after
+// Agent restarts full state syncing is not required.
 type PackagesStateProvider interface {
+	// AllPackagesHash returns the hash of all packages previously set via SetAllPackagesHash().
 	AllPackagesHash() ([]byte, error)
 
+	// SetAllPackagesHash must remember the AllPackagesHash. Must be returned
+	// later when AllPackagesHash is called. SetAllPackagesHash is called after all
+	// package updates complete successfully.
+	SetAllPackagesHash(hash []byte) error
+
 	// Packages returns the names of all packages that exist in the Agent's local storage.
 	Packages() ([]string, error)
 
-	// PackageHash returns the hash of a local package. packageName is one of the names
+	// PackageState returns the state of a local package. packageName is one of the names
 	// that were returned by Packages().
-	PackageHash(packageName string) ([]byte, error)
+	// Returns (PackageState{Exists:false},nil) if package does not exist locally.
+	PackageState(packageName string) (state PackageState, err error)
+
+	// SetPackageState must remember the state for the specified package. Must be returned
+	// later when PackageState is called. SetPackageState is called after UpdateContent
+	// call completes successfully.
+	// The state.Type must be equal to the current Type of the package otherwise
+	// the call may fail with an error.
+	SetPackageState(packageName string, state PackageState) error
 
 	// CreatePackage creates the package locally. If the package existed must return an error.
 	// If the package did not exist its hash should be set to nil.
-	CreatePackage(packageName string) error
+	CreatePackage(packageName string, typ protobufs.PackageAvailable_PackageType) error
 
 	// FileContentHash returns the content hash of the package file that exists locally.
+	// Returns (nil,nil) if package or package file is not found.
 	FileContentHash(packageName string) ([]byte, error)
 
 	// UpdateContent must create or update the package content file. The entire content
@@ -41,16 +78,13 @@ type PackagesStateProvider interface {
 	// The function must cancel and return an error if the context is cancelled.
 	UpdateContent(ctx context.Context, packageName string, data io.Reader, contentHash []byte) error
 
-	// SetPackageHash must remember the hash for the specified package. Must be returned
-	// later when PackageHash is called. SetPackageHash is called after UpdateContent
-	// call completes successfully.
-	SetPackageHash(packageName string, hash []byte) error
-
 	// DeletePackage deletes the package from the Agent's local storage.
 	DeletePackage(packageName string) error
 
-	// SetAllPackagesHash must remember the AllPackagesHash. Must be returned
-	// later when AllPackagesHash is called. SetAllPackagesHash is called after all
-	// package updates complete successfully.
-	SetAllPackagesHash(hash []byte) error
+	// LastReportedStatuses returns the value previously set via SetLastReportedStatuses.
+	LastReportedStatuses() (*protobufs.PackageStatuses, error)
+
+	// SetLastReportedStatuses saves the statuses in the local state. This is called
+	// periodically during syncing process to save the most recent statuses.
+	SetLastReportedStatuses(statuses *protobufs.PackageStatuses) error
 }
diff --git a/client/types/startsettings.go b/client/types/startsettings.go
index 870a9b1..a3758ec 100644
--- a/client/types/startsettings.go
+++ b/client/types/startsettings.go
@@ -37,7 +37,8 @@ type StartSettings struct {
 
 	LastConnectionSettingsHash []byte
 
-	// The hash of the last locally-saved Server-provided packages. If nil is passed
-	// it will force the Server to send packages list back.
-	LastServerProvidedAllPackagesHash []byte
+	// PackagesStateProvider provides access to the local state of packages.
+	// If nil then ReportsPackageStatuses and AcceptsPackages capabilities will be disabled,
+	// i.e. package status reporting and syncing from the Server will be disabled.
+	PackagesStateProvider PackagesStateProvider
 }
diff --git a/client/wsclient.go b/client/wsclient.go
index aa3e424..6d028b0 100644
--- a/client/wsclient.go
+++ b/client/wsclient.go
@@ -216,7 +216,14 @@ func (c *wsClient) runOneCycle(ctx context.Context) {
 	}
 
 	// First status report sent. Now loop to receive and process messages.
-	r := internal.NewWSReceiver(c.common.Logger, c.common.Callbacks, c.conn, c.sender, &c.common.ClientSyncedState)
+	r := internal.NewWSReceiver(
+		c.common.Logger,
+		c.common.Callbacks,
+		c.conn,
+		c.sender,
+		&c.common.ClientSyncedState,
+		c.common.PackagesStateProvider,
+	)
 	r.ReceiverLoop(ctx)
 
 	// Stop the background processors.


Test patch:
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
index 03db4d1..0084550 100644
--- a/client/clientimpl_test.go
+++ b/client/clientimpl_test.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"math/rand"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"sync/atomic"
 	"testing"
@@ -397,6 +398,7 @@ func TestIncludesDetailsOnReconnect(t *testing.T) {
 	eventually(t, func() bool { return atomic.LoadInt64(&receivedDetails) == 1 })
 
 	// close the Agent connection. expect it to reconnect and send details again.
+	require.NotNil(t, client.conn)
 	err := client.conn.Close()
 	assert.NoError(t, err)
 
@@ -444,7 +446,7 @@ func TestSetEffectiveConfig(t *testing.T) {
 		settings.OpAMPServerURL = "ws://" + srv.Endpoint
 		prepareClient(t, &settings, client)
 
-		assert.NoError(t, client.Start(context.Background(), settings))
+		require.NoError(t, client.Start(context.Background(), settings))
 
 		// Verify config is delivered.
 		eventually(
@@ -972,3 +974,253 @@ func TestRemoteConfigUpdate(t *testing.T) {
 		})
 	}
 }
+
+type packageTestCase struct {
+	name                string
+	errorOnCallback     bool
+	available           *protobufs.PackagesAvailable
+	expectedStatus      *protobufs.PackageStatuses
+	expectedFileContent map[string][]byte
+}
+
+const packageUpdateErrorMsg = "cannot update packages"
+
+func verifyUpdatePackages(t *testing.T, testCase packageTestCase) {
+	testClients(t, func(t *testing.T, client OpAMPClient) {
+
+		// Start a Server.
+		srv := internal.StartMockServer(t)
+		srv.EnableExpectMode()
+
+		localPackageState := internal.NewInMemPackagesStore()
+
+		var syncerDoneCh <-chan struct{}
+
+		// Prepare a callback that returns either success or failure.
+		onPackagesAvailable := func(ctx context.Context, packages *protobufs.PackagesAvailable, syncer types.PackagesSyncer) error {
+			if testCase.errorOnCallback {
+				return errors.New(packageUpdateErrorMsg)
+			} else {
+				syncerDoneCh = syncer.Done()
+				err := syncer.Sync(ctx)
+				require.NoError(t, err)
+				return nil
+			}
+		}
+
+		// Start a client.
+		settings := types.StartSettings{
+			OpAMPServerURL: "ws://" + srv.Endpoint,
+			Callbacks: types.CallbacksStruct{
+				OnPackagesAvailableFunc: onPackagesAvailable,
+			},
+			PackagesStateProvider: localPackageState,
+		}
+		prepareClient(t, &settings, client)
+
+		// Client --->
+		assert.NoError(t, client.Start(context.Background(), settings))
+
+		// ---> Server
+		srv.Expect(func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+			// Send the packages to the Agent.
+			return &protobufs.ServerToAgent{
+				InstanceUid:       msg.InstanceUid,
+				PackagesAvailable: testCase.available,
+			}
+		})
+
+		// The Agent will try to install the packages and will send the status
+		// report about it back to the Server.
+
+		var lastStatusHash []byte
+
+		// ---> Server
+		// Wait for the expected package statuses to be received.
+		srv.EventuallyExpect("full PackageStatuses",
+			func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool) {
+				expectedStatusReceived := false
+
+				status := msg.PackageStatuses
+				require.NotNil(t, status)
+				assert.EqualValues(t, testCase.expectedStatus.ServerProvidedAllPackagesHash, status.ServerProvidedAllPackagesHash)
+				lastStatusHash = status.Hash
+
+				// Verify individual package statuses.
+				for name, pkgExpected := range testCase.expectedStatus.Packages {
+					pkgStatus := status.Packages[name]
+					if pkgStatus == nil {
+						// Package status not yet included in the report.
+						continue
+					}
+					switch pkgStatus.Status {
+					case protobufs.PackageStatus_InstallFailed:
+						assert.Contains(t, pkgStatus.ErrorMessage, pkgExpected.ErrorMessage)
+
+					case protobufs.PackageStatus_Installed:
+						assert.EqualValues(t, pkgExpected.AgentHasHash, pkgStatus.AgentHasHash)
+						assert.EqualValues(t, pkgExpected.AgentHasVersion, pkgStatus.AgentHasVersion)
+						assert.Empty(t, pkgStatus.ErrorMessage)
+					default:
+						assert.Empty(t, pkgStatus.ErrorMessage)
+					}
+					assert.EqualValues(t, pkgExpected.ServerOfferedHash, pkgStatus.ServerOfferedHash)
+					assert.EqualValues(t, pkgExpected.ServerOfferedVersion, pkgStatus.ServerOfferedVersion)
+
+					if pkgStatus.Status == pkgExpected.Status {
+						expectedStatusReceived = true
+						assert.Len(t, status.Packages, len(testCase.available.Packages))
+					}
+				}
+				assert.NotNil(t, status.Hash)
+
+				return &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}, expectedStatusReceived
+			})
+
+		if syncerDoneCh != nil {
+			// Wait until all syncing is done.
+			<-syncerDoneCh
+
+			for pkgName, receivedContent := range localPackageState.GetContent() {
+				expectedContent := testCase.expectedFileContent[pkgName]
+				assert.EqualValues(t, expectedContent, receivedContent)
+			}
+		}
+
+		// Client --->
+		// Trigger another status report by setting AgentDescription.
+		_ = client.SetAgentDescription(client.AgentDescription())
+
+		// ---> Server
+		srv.EventuallyExpect("compressed PackageStatuses",
+			func(msg *protobufs.AgentToServer) (*protobufs.ServerToAgent, bool) {
+				// Ensure that compressed status is received.
+				status := msg.PackageStatuses
+				require.NotNil(t, status)
+				compressedReceived := status.ServerProvidedAllPackagesHash == nil
+				if compressedReceived {
+					assert.Nil(t, status.ServerProvidedAllPackagesHash)
+					assert.Nil(t, status.Packages)
+				}
+				assert.NotNil(t, status.Hash)
+				assert.Equal(t, lastStatusHash, status.Hash)
+
+				response := &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}
+
+				if compressedReceived {
+					// Ask for full report again.
+					response.Flags = protobufs.ServerToAgent_ReportPackageStatuses
+				} else {
+					// Keep triggering status report by setting AgentDescription
+					// until the compressed PackageStatuses arrives.
+					_ = client.SetAgentDescription(client.AgentDescription())
+				}
+
+				return response, compressedReceived
+			})
+
+		// Shutdown the Server.
+		srv.Close()
+
+		// Shutdown the client.
+		err := client.Stop(context.Background())
+		assert.NoError(t, err)
+	})
+}
+
+// Downloadable package file constants.
+const packageFileURL = "/validfile.pkg"
+
+var packageFileContent = []byte("Package File Content")
+
+func createDownloadSrv(t *testing.T) *httptest.Server {
+	m := http.NewServeMux()
+	m.HandleFunc(packageFileURL,
+		func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+			_, err := w.Write(packageFileContent)
+			assert.NoError(t, err)
+		},
+	)
+
+	srv := httptest.NewServer(m)
+
+	u, err := url.Parse(srv.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	endpoint := u.Host
+	testhelpers.WaitForEndpoint(endpoint)
+
+	return srv
+}
+
+func createPackageTestCase(name string, downloadSrv *httptest.Server) packageTestCase {
+	return packageTestCase{
+		name:            name,
+		errorOnCallback: false,
+		available: &protobufs.PackagesAvailable{
+			Packages: map[string]*protobufs.PackageAvailable{
+				"package1": {
+					Type:    protobufs.PackageAvailable_TopLevelPackage,
+					Version: "1.0.0",
+					File: &protobufs.DownloadableFile{
+						DownloadUrl: downloadSrv.URL + packageFileURL,
+						ContentHash: []byte{4, 5},
+					},
+					Hash: []byte{1, 2, 3},
+				},
+			},
+			AllPackagesHash: []byte{1, 2, 3, 4, 5},
+		},
+
+		expectedStatus: &protobufs.PackageStatuses{
+			Packages: map[string]*protobufs.PackageStatus{
+				"package1": {
+					Name:                 "package1",
+					AgentHasVersion:      "1.0.0",
+					AgentHasHash:         []byte{1, 2, 3},
+					ServerOfferedVersion: "1.0.0",
+					ServerOfferedHash:    []byte{1, 2, 3},
+					Status:               protobufs.PackageStatus_Installed,
+					ErrorMessage:         "",
+				},
+			},
+			ServerProvidedAllPackagesHash: []byte{1, 2, 3, 4, 5},
+		},
+
+		expectedFileContent: map[string][]byte{
+			"package1": packageFileContent,
+		},
+	}
+}
+
+func TestUpdatePackages(t *testing.T) {
+
+	downloadSrv := createDownloadSrv(t)
+	defer downloadSrv.Close()
+
+	// A success case.
+	var tests []packageTestCase
+	tests = append(tests, createPackageTestCase("success", downloadSrv))
+
+	// A case when downloading the file fails because the URL is incorrect.
+	notFound := createPackageTestCase("downloadable file not found", downloadSrv)
+	notFound.available.Packages["package1"].File.DownloadUrl = downloadSrv.URL + "/notfound"
+	notFound.expectedStatus.Packages["package1"].Status = protobufs.PackageStatus_InstallFailed
+	notFound.expectedStatus.Packages["package1"].ErrorMessage = "cannot download"
+	tests = append(tests, notFound)
+
+	// A case when OnPackagesAvailable callback returns an error.
+	errorOnCallback := createPackageTestCase("error on callback", downloadSrv)
+	errorOnCallback.expectedStatus.Packages["package1"].Status = protobufs.PackageStatus_InstallFailed
+	errorOnCallback.expectedStatus.Packages["package1"].ErrorMessage = packageUpdateErrorMsg
+	errorOnCallback.errorOnCallback = true
+	tests = append(tests, errorOnCallback)
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			verifyUpdatePackages(t, test)
+		})
+	}
+}
diff --git a/client/internal/wsreceiver_test.go b/client/internal/wsreceiver_test.go
index a78b8f3..670ce74 100644
--- a/client/internal/wsreceiver_test.go
+++ b/client/internal/wsreceiver_test.go
@@ -75,7 +75,7 @@ func TestServerToAgentCommand(t *testing.T) {
 				remoteConfigStatus: &protobufs.RemoteConfigStatus{},
 			}
 			sender := WSSender{}
-			receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, &sender, &clientSyncedState)
+			receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, &sender, &clientSyncedState, nil)
 			receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
 				Command: test.command,
 			})
@@ -99,7 +99,7 @@ func TestServerToAgentCommandExclusive(t *testing.T) {
 		},
 	}
 	clientSyncedState := ClientSyncedState{}
-	receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, nil, &clientSyncedState)
+	receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, nil, &clientSyncedState, nil)
 	receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
 		Command: &protobufs.ServerToAgentCommand{
 			Type: protobufs.ServerToAgentCommand_Restart,
@@ -122,7 +122,7 @@ func TestOnRemoteConfigReportsError(t *testing.T) {
 
 	sender := NewSender(TestLogger{t})
 	clientSyncedState := ClientSyncedState{}
-	receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, sender, &clientSyncedState)
+	receiver := NewWSReceiver(TestLogger{t}, callbacks, nil, sender, &clientSyncedState, nil)
 
 	receiver.processor.ProcessReceivedMessage(context.Background(), &protobufs.ServerToAgent{
 		RemoteConfig: &protobufs.AgentRemoteConfig{},