feat: Add Cloud Endpoint Sync Commands (#216)

* feat: Add cloud sync command

* feat: Add pmg cloud commands

* fix: Code review fixes

* fix: Code review fixes

* fix: Code review fixes

* fix: Code review fixes

* fix: Code review fixes

* fix: Code review fixes

internal/audit/audit.go

@@ -9,7 +9,6 @@ import (
 	"github.com/safedep/dry/log"
 	"github.com/safedep/pmg/config"
 	"github.com/safedep/pmg/internal/analytics"
-	"github.com/safedep/pmg/usefulerror"
 )
 
 var global *auditor
@@ -31,14 +30,10 @@ func Initialize(cfg *config.RuntimeConfig) error {
 	if cfg.Config.Cloud.Enabled && !analytics.IsDisabled() {
 		cs, err := newCloudSink(cfg)
 		if err != nil {
-			return usefulerror.Useful().
-				Wrap(err).
-				WithCode(usefulerror.ErrCodeLifecycle).
-				WithHumanError("Cloud sync is enabled but failed to initialize").
-				WithHelp("Ensure SAFEDEP_API_KEY and SAFEDEP_TENANT_ID environment variables are set").
-				WithAdditionalHelp("Disable cloud sync with 'cloud.enabled: false' in config if not needed")
+			log.Warnf("Cloud sync initialization failed: %v", err)
+		} else {
+			sinks = append(sinks, cs)
 		}
-		sinks = append(sinks, cs)
 	}
 
 	if cfg.Config.Cloud.Enabled && analytics.IsDisabled() {

internal/audit/cloud_client.go

@@ -0,0 +1,131 @@
+package audit
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/safedep/dry/cloud"
+	"github.com/safedep/dry/cloud/endpointsync"
+	"github.com/safedep/dry/log"
+	"github.com/safedep/pmg/config"
+	appVersion "github.com/safedep/pmg/internal/version"
+)
+
+// SyncClientBundle holds a SyncClient and its underlying cloud client.
+// Callers must call Close() when done.
+type SyncClientBundle struct {
+	syncClient       *endpointsync.SyncClient
+	cloudClient      *cloud.Client
+	keychainResolver cloud.CloseableCredentialResolver
+}
+
+// Sync delivers pending events from the WAL to SafeDep Cloud.
+func (b *SyncClientBundle) Sync(ctx context.Context) (int, error) {
+	return b.syncClient.Sync(ctx)
+}
+
+func (b *SyncClientBundle) Close() error {
+	var errs []error
+	if b.syncClient != nil {
+		if err := b.syncClient.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if b.cloudClient != nil {
+		if err := b.cloudClient.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if b.keychainResolver != nil {
+		if err := b.keychainResolver.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// NewSyncClientBundle creates an authenticated SyncClient connected to SafeDep Cloud.
+func NewSyncClientBundle(cfg *config.RuntimeConfig) (*SyncClientBundle, error) {
+	// Build credential resolver chain: keychain first, env fallback.
+	var resolvers []cloud.CredentialResolver
+	var keychainResolver cloud.CloseableCredentialResolver
+
+	keychainResolver, err := cloud.NewKeychainCredentialResolver(cloud.CredentialTypeAPIKey)
+	if err != nil {
+		log.Debugf("Keychain credential resolver not available, skipping: %v", err)
+	} else {
+		resolvers = append(resolvers, keychainResolver)
+	}
+
+	envResolver, err := cloud.NewEnvCredentialResolver()
+	if err != nil {
+		log.Debugf("Env credential resolver not available, skipping: %v", err)
+	} else {
+		resolvers = append(resolvers, envResolver)
+	}
+
+	if len(resolvers) == 0 {
+		if keychainResolver != nil {
+			if closeErr := keychainResolver.Close(); closeErr != nil {
+				log.Warnf("failed to close keychain resolver: %v", closeErr)
+			}
+		}
+		return nil, fmt.Errorf("no credential resolvers available")
+	}
+
+	chain := cloud.NewChainCredentialResolver(resolvers...)
+	creds, err := chain.Resolve()
+	if err != nil {
+		if keychainResolver != nil {
+			if closeErr := keychainResolver.Close(); closeErr != nil {
+				log.Warnf("failed to close keychain resolver: %v", closeErr)
+			}
+		}
+		return nil, fmt.Errorf("failed to resolve cloud credentials: %w", err)
+	}
+
+	cloudClient, err := cloud.NewDataPlaneClient("pmg", creds)
+	if err != nil {
+		if keychainResolver != nil {
+			if closeErr := keychainResolver.Close(); closeErr != nil {
+				log.Warnf("failed to close keychain resolver: %v", closeErr)
+			}
+		}
+		return nil, fmt.Errorf("failed to create data plane client: %w", err)
+	}
+
+	transport := endpointsync.NewGrpcTransport(cloudClient.Connection())
+
+	var identityOpts []endpointsync.EndpointIdentityOption
+	if cfg.Config.Cloud.EndpointID != "" {
+		identityOpts = append(identityOpts, endpointsync.WithEndpointID(cfg.Config.Cloud.EndpointID))
+	}
+
+	identity := endpointsync.NewEndpointIdentityResolver(identityOpts...)
+
+	toolVersion := appVersion.Version
+	if toolVersion == "" {
+		toolVersion = "dev"
+	}
+
+	syncClient, err := endpointsync.NewSyncClient("pmg", toolVersion, transport, identity,
+		endpointsync.WithWALPath(cfg.CloudSyncDBPath()))
+	if err != nil {
+		if closeErr := cloudClient.Close(); closeErr != nil {
+			log.Warnf("failed to close cloud client after sync client init failure: %v", closeErr)
+		}
+		if keychainResolver != nil {
+			if closeErr := keychainResolver.Close(); closeErr != nil {
+				log.Warnf("failed to close keychain resolver: %v", closeErr)
+			}
+		}
+		return nil, fmt.Errorf("failed to create sync client: %w", err)
+	}
+
+	return &SyncClientBundle{
+		syncClient:       syncClient,
+		cloudClient:      cloudClient,
+		keychainResolver: keychainResolver,
+	}, nil
+}

internal/audit/cloud_sink.go

@@ -6,79 +6,33 @@ import (
 	"fmt"
 
 	"github.com/google/uuid"
-	"github.com/safedep/dry/cloud"
 	"github.com/safedep/dry/cloud/endpointsync"
 	"github.com/safedep/dry/log"
 	"github.com/safedep/pmg/config"
-	appVersion "github.com/safedep/pmg/internal/version"
 )
 
 type cloudSink struct {
-	syncClient   *endpointsync.SyncClient
-	cloudClient  *cloud.Client
+	*SyncClientBundle
 	invocationID string
 }
 
 func newCloudSink(cfg *config.RuntimeConfig) (*cloudSink, error) {
-	resolver, err := cloud.NewEnvCredentialResolver()
+	bundle, err := NewSyncClientBundle(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create credential resolver: %w", err)
-	}
-
-	creds, err := resolver.Resolve()
-	if err != nil {
-		return nil, fmt.Errorf("failed to resolve cloud credentials: %w", err)
-	}
-
-	cloudClient, err := cloud.NewDataPlaneClient("pmg", creds)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create data plane client: %w", err)
-	}
-
-	transport := endpointsync.NewGrpcTransport(cloudClient.Connection())
-
-	sink, err := newCloudSinkWithTransport(transport, cfg.Config.Cloud.EndpointID, cfg.CloudSyncDBPath())
-	if err != nil {
-		if closeErr := cloudClient.Close(); closeErr != nil {
-			log.Warnf("failed to close cloud client after sink init failure: %v", closeErr)
-		}
 		return nil, err
 	}
 
-	sink.cloudClient = cloudClient
-	return sink, nil
-}
-
-func newCloudSinkWithTransport(transport endpointsync.EventTransport, endpointID, walPath string) (*cloudSink, error) {
-	var identityOpts []endpointsync.EndpointIdentityOption
-	if endpointID != "" {
-		identityOpts = append(identityOpts, endpointsync.WithEndpointID(endpointID))
-	}
-
-	identity := endpointsync.NewEndpointIdentityResolver(identityOpts...)
-
-	toolVersion := appVersion.Version
-	if toolVersion == "" {
-		toolVersion = "dev"
-	}
-
-	syncClient, err := endpointsync.NewSyncClient("pmg", toolVersion, transport, identity,
-		endpointsync.WithWALPath(walPath))
-	if err != nil {
-		return nil, fmt.Errorf("failed to create sync client: %w", err)
-	}
-
 	invocationID, err := uuid.NewRandom()
 	if err != nil {
-		if closeErr := syncClient.Close(); closeErr != nil {
-			log.Warnf("failed to close sync client after invocation ID generation failure: %v", closeErr)
+		if closeErr := bundle.Close(); closeErr != nil {
+			log.Warnf("failed to close sync client bundle after invocation ID failure: %v", closeErr)
 		}
 		return nil, fmt.Errorf("failed to generate invocation ID: %w", err)
 	}
 
 	return &cloudSink{
-		syncClient:   syncClient,
-		invocationID: invocationID.String(),
+		SyncClientBundle: bundle,
+		invocationID:     invocationID.String(),
 	}, nil
 }
 
@@ -109,17 +63,7 @@ func (s *cloudSink) Handle(ctx context.Context, event AuditEvent) error {
 	return nil
 }
 
+// Close delegates to the embedded SyncClientBundle.Close().
 func (s *cloudSink) Close() error {
-	var errs []error
-	if s.syncClient != nil {
-		if err := s.syncClient.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
-	if s.cloudClient != nil {
-		if err := s.cloudClient.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
-	return errors.Join(errs...)
+	return s.SyncClientBundle.Close()
 }

internal/audit/cloud_sink_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	servicev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/services/controltower/v1"
+	"github.com/safedep/dry/cloud/endpointsync"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -29,17 +30,28 @@ func (m *mockTransport) Close() error {
 	return nil
 }
 
+func newTestCloudSink(t *testing.T, transport endpointsync.EventTransport) *cloudSink {
+	t.Helper()
+	walPath := t.TempDir() + "/test-sync.db"
+	identity := endpointsync.NewEndpointIdentityResolver()
+	syncClient, err := endpointsync.NewSyncClient("pmg", "test", transport, identity,
+		endpointsync.WithWALPath(walPath))
+	require.NoError(t, err)
+	return &cloudSink{
+		SyncClientBundle: &SyncClientBundle{syncClient: syncClient},
+		invocationID:     "test-invocation",
+	}
+}
+
 func TestCloudSinkEmitsTranslatableEvents(t *testing.T) {
 	transport := &mockTransport{}
-	walPath := t.TempDir() + "/test-sync.db"
 
-	sink, err := newCloudSinkWithTransport(transport, "", walPath)
-	require.NoError(t, err)
+	sink := newTestCloudSink(t, transport)
 	defer func() {
 		require.NoError(t, sink.Close())
 	}()
 
-	err = sink.Handle(context.Background(), AuditEvent{
+	err := sink.Handle(context.Background(), AuditEvent{
 		Type:      EventTypeMalwareBlocked,
 		Timestamp: time.Now(),
 		Message:   "blocked malware package",
@@ -49,15 +61,13 @@ func TestCloudSinkEmitsTranslatableEvents(t *testing.T) {
 
 func TestCloudSinkSkipsUntranslatableEvents(t *testing.T) {
 	transport := &mockTransport{}
-	walPath := t.TempDir() + "/test-sync.db"
 
-	sink, err := newCloudSinkWithTransport(transport, "", walPath)
-	require.NoError(t, err)
+	sink := newTestCloudSink(t, transport)
 	defer func() {
 		require.NoError(t, sink.Close())
 	}()
 
-	err = sink.Handle(context.Background(), AuditEvent{
+	err := sink.Handle(context.Background(), AuditEvent{
 		Type:      EventTypeProxyHostObserved,
 		Timestamp: time.Now(),
 		Message:   "observed proxy host",
@@ -68,16 +78,14 @@ func TestCloudSinkSkipsUntranslatableEvents(t *testing.T) {
 
 func TestCloudSinkEmitAndSync(t *testing.T) {
 	transport := &mockTransport{}
-	walPath := t.TempDir() + "/test-sync.db"
 
-	sink, err := newCloudSinkWithTransport(transport, "", walPath)
-	require.NoError(t, err)
+	sink := newTestCloudSink(t, transport)
 	defer func() {
 		require.NoError(t, sink.Close())
 	}()
 
 	ctx := context.Background()
-	err = sink.Handle(ctx, AuditEvent{
+	err := sink.Handle(ctx, AuditEvent{
 		Type:      EventTypeMalwareBlocked,
 		Timestamp: time.Now(),
 		Message:   "blocked malware package",

internal/audit/cloud_translate_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 // testCloudSink returns a cloudSink with no real SyncClient, suitable for testing translation.
-var testSink = &cloudSink{invocationID: "test-invocation"}
+var testSink = &cloudSink{SyncClientBundle: &SyncClientBundle{}, invocationID: "test-invocation"}
 
 func TestTranslateMalwareBlocked(t *testing.T) {
 	event := AuditEvent{

internal/flows/proxy_flow.go

@@ -74,14 +74,19 @@ func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagema
 	startTime := time.Now()
 
 	audit.LogInstallStarted(f.pm.Name(), args)
+
+	sessionCompleted := false
 	defer func() {
+		if sessionCompleted {
+			return
+		}
+
 		// On early error returns (e.g. CA cert, analyzer init), reportData.Outcome
-		// is still the default (Success). Override to Error for these cases. Don't
-		// override when outcome was explicitly set (e.g. Blocked) — those paths may
-		// also return an error from the underlying package manager command.
+		// is still the default (Success). Override to Error for these cases.
 		if runErr != nil && reportData.Outcome == ui.OutcomeSuccess {
 			reportData.Outcome = ui.OutcomeError
 		}
+
 		audit.LogSessionComplete(audit.Outcome(reportData.Outcome.String()), audit.FlowTypeProxy)
 	}()
 
@@ -169,6 +174,7 @@ func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagema
 	}()
 
 	ui.ClearStatus()
+
 	log.Infof("Proxy server started on %s", proxyAddr)
 	log.Infof("Running %s with proxy protection enabled", f.pm.Name())
 
@@ -192,12 +198,16 @@ func (f *proxyFlow) Run(ctx context.Context, args []string, parsedCmd *packagema
 	reportData.BlockedCount = stats.BlockedCount
 	reportData.BlockedPackages = statsCollector.GetBlockedPackages()
 	reportData.ConfirmedPackages = statsCollector.GetConfirmedPackages()
-
 	reportData.CooldownBlockedPackages = statsCollector.GetCooldownBlocks()
 
 	// Set outcome based on execution result using shared inference logic
 	reportData.Outcome = inferOutcome(cfg.InsecureInstallation, cfg.DryRun, reportData.BlockedCount, stats.UserCancelledCount, executionError)
 
+	// Emit session complete before report/exit — handleExecutionResultError may call
+	// os.Exit which skips defers, so we must emit the session summary here.
+	audit.LogSessionComplete(audit.Outcome(reportData.Outcome.String()), audit.FlowTypeProxy)
+	sessionCompleted = true
+
 	// Show the report
 	ui.Report(reportData)
 

internal/ui/ui.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/safedep/pmg/analyzer"
 	"github.com/safedep/pmg/internal/models"
+	"golang.org/x/term"
 )
 
 // The UI is internal to PMG and opinionated for the CLI.
@@ -135,6 +136,51 @@ func ShowWarning(message string) {
 	fmt.Fprintf(os.Stderr, "PMG: %s\n", Colors.Red(message))
 }
 
+// Infof prints an informational message, suppressed in silent mode.
+func Infof(msg string, args ...interface{}) {
+	if verbosityLevel == VerbosityLevelSilent {
+		return
+	}
+	fmt.Println(fmt.Sprintf(msg, args...))
+}
+
+// Successf prints a green success message, suppressed in silent mode.
+func Successf(msg string, args ...interface{}) {
+	if verbosityLevel == VerbosityLevelSilent {
+		return
+	}
+	fmt.Printf("%s %s\n", Colors.Green("✓"), fmt.Sprintf(msg, args...))
+}
+
+// PromptInput prints a label and reads a line of visible input from stdin.
+func PromptInput(label string) (string, error) {
+	fmt.Printf("%s %s", Colors.Cyan("›"), Colors.Bold(label))
+	scanner := bufio.NewScanner(os.Stdin)
+	if scanner.Scan() {
+		return strings.TrimSpace(scanner.Text()), nil
+	}
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+	return "", fmt.Errorf("no input received")
+}
+
+// PromptSecret prints a label and reads input from stdin with echo disabled.
+// Returns an error if stdin is not a terminal (e.g. piped input).
+func PromptSecret(label string) (string, error) {
+	if !term.IsTerminal(int(os.Stdin.Fd())) {
+		return "", fmt.Errorf("interactive terminal required for secure input")
+	}
+
+	fmt.Printf("%s %s", Colors.Cyan("▪"), Colors.Bold(label))
+	raw, err := term.ReadPassword(int(os.Stdin.Fd()))
+	fmt.Println() // newline after hidden input
+	if err != nil {
+		return "", err
+	}
+	return strings.TrimSpace(string(raw)), nil
+}
+
 func Fatalf(msg string, args ...interface{}) {
 	ClearStatus()