feat: emit cloud events for dependency cooldown and host observations (#243)

* feat: emit cloud events for dependency cooldown and host observations (#237)

Wire cooldown blocks and proxy host observations through the cloud sync
pipeline so they appear as telemetry in Control Tower.

- Cooldown blocks emit PACKAGE_DECISION with COOLDOWN_BLOCKED action and
  PmgDependencyCooldown context (publish date, cooldown days, days since
  publish, days remaining)
- Proxy host observations emit HOST_OBSERVATION with PmgHostObservation
  (hostname, method)
- Session summary now includes cooldown_blocked_count
- Updated buf API dependency for new proto schema

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* format file

* fix: add explicit eventlog mapping for EventTypeDependencyCooldown

Follow the existing pattern where every audit event type has an explicit
case in mapEventType and a corresponding constant in the eventlog package.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Apply suggestion from @devin-ai-integration[bot]

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Signed-off-by: Abhisek Datta <abhisek.datta@gmail.com>

---------

Signed-off-by: Abhisek Datta <abhisek.datta@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Abhisek Datta <abhisek.datta@gmail.com>
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

go.mod

@@ -3,8 +3,8 @@ module github.com/safedep/pmg
 go 1.25.1
 
 require (
-	buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260409081445-73994c4e35a3.1
+	buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260507092425-ac47f9a19339.1
-	buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260409081445-73994c4e35a3.1
+	buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260507092425-ac47f9a19339.1
 	github.com/Masterminds/semver v1.5.0
 	github.com/elazarl/goproxy v1.8.1
 	github.com/fatih/color v1.18.0

go.sum

@@ -4,8 +4,12 @@ buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-202405082006
 buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20240508200655-46a4cf4ba109.1/go.mod h1:tvtbpgaVXZX4g6Pn+AnzFycuRK3MOz5HJfEGeEllXYM=
 buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260409081445-73994c4e35a3.1 h1:OKKSMXf1k7NZ1qcp5gfpTw0F3QGVM/X6ghkRs9GtrIA=
 buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260409081445-73994c4e35a3.1/go.mod h1:k1tkPvr2SWI5hWfY/fF9XgIBh1sZOMPMPl/bUH3L0wE=
+buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260507092425-ac47f9a19339.1 h1:TgL0Xu+EFQhr68C/eGrflL+TaFtEUWkfWXQ+f+vWGdg=
+buf.build/gen/go/safedep/api/grpc/go v1.6.1-20260507092425-ac47f9a19339.1/go.mod h1:AU7tshd3hSyemWEYn6mofmWPUJaWmGExFHAQSjp6N4o=
 buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260409081445-73994c4e35a3.1 h1:x5a6h/YeT2cqgxqmBGOCKlifMEMb4VRJ3eOgQkNvJrs=
 buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260409081445-73994c4e35a3.1/go.mod h1:I8E+sZXJNqzWBtSlRGCoiEorLSRiix50h2R/66aBzME=
+buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260507092425-ac47f9a19339.1 h1:JRzHMhoJg1Mlae+PR+ZZ1I1aaqAJdyF4WY3CidJD/us=
+buf.build/gen/go/safedep/api/protocolbuffers/go v1.36.11-20260507092425-ac47f9a19339.1/go.mod h1:I8E+sZXJNqzWBtSlRGCoiEorLSRiix50h2R/66aBzME=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg=

internal/audit/audit.go

@@ -199,6 +199,23 @@ func LogProxyHostObserved(hostname, method, reason string, details map[string]in
 	})
 }
 
+// LogDependencyCooldown records that a package was blocked by the dependency cooldown policy.
+func LogDependencyCooldown(pv *packagev1.PackageVersion, publishDate time.Time, cooldownDays, daysAgo, daysLeft int) {
+	logEvent(AuditEvent{
+		Type:           EventTypeDependencyCooldown,
+		Message:        fmt.Sprintf("Package blocked by cooldown policy: %s@%s (published %d days ago, %d days remaining)", pkgName(pv), pkgVersion(pv), daysAgo, daysLeft),
+		PackageVersion: pv,
+		PublishDate:    publishDate,
+		CooldownDays:   cooldownDays,
+		DaysAgo:        daysAgo,
+		DaysLeft:       daysLeft,
+	})
+
+	if global != nil {
+		global.recordCooldownBlocked()
+	}
+}
+
 // LogSandboxOverride records that runtime sandbox policy overrides were applied.
 func LogSandboxOverride(sandboxProfile string, overrides []map[string]string) {
 	logEvent(AuditEvent{
@@ -259,6 +276,7 @@ func LogSessionComplete(outcome Outcome, flowType FlowType) {
 			ConfirmedCount:       s.confirmedCount,
 			TrustedSkipped:       s.trustedSkipped,
 			InsecureBypassed:     s.insecureBypassed,
+			CooldownBlockedCount: s.cooldownBlockedCount,
 			Duration:             time.Since(s.startTime),
 			SandboxEnabled:       cfg.Config.Sandbox.Enabled,
 			ParanoidMode:         cfg.Config.Paranoid,

internal/audit/auditor.go

@@ -19,6 +19,7 @@ type session struct {
 	confirmedCount       uint32
 	trustedSkipped       uint32
 	insecureBypassed     uint32
+	cooldownBlockedCount uint32
 }
 
 type auditor struct {
@@ -125,3 +126,14 @@ func (a *auditor) recordInsecureBypassed() {
 	s.insecureBypassed++
 	s.totalAnalyzed++
 }
+
+func (a *auditor) recordCooldownBlocked() {
+	s := a.getSession()
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.cooldownBlockedCount++
+	s.totalAnalyzed++
+}

internal/audit/cloud_translate.go

@@ -5,6 +5,7 @@ import (
 
 	controltowerv1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/controltower/v1"
 	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func (s *cloudSink) translateToPmgEvents(event AuditEvent) []*controltowerv1.PmgEvent {
@@ -18,6 +19,10 @@ func (s *cloudSink) translateToPmgEvents(event AuditEvent) []*controltowerv1.Pmg
 		// not a per-package event. It is emitted as part of EventTypeSessionComplete when
 		// the session's insecureBypassed counter is > 0.
 		return nil
+	case EventTypeDependencyCooldown:
+		return []*controltowerv1.PmgEvent{newCooldownBlockedEvent(event)}
+	case EventTypeProxyHostObserved:
+		return []*controltowerv1.PmgEvent{newHostObservationEvent(event)}
 	case EventTypeSandboxOverride:
 		return []*controltowerv1.PmgEvent{newSandboxOverrideEvent(event)}
 	case EventTypeError:
@@ -84,6 +89,37 @@ func newErrorEvent(event AuditEvent) *controltowerv1.PmgEvent {
 	return e
 }
 
+func newCooldownBlockedEvent(event AuditEvent) *controltowerv1.PmgEvent {
+	decision := &controltowerv1.PmgPackageDecision{}
+	decision.SetPackageVersion(event.PackageVersion)
+	decision.SetAction(controltowerv1.PmgPackageAction_PMG_PACKAGE_ACTION_COOLDOWN_BLOCKED)
+
+	cooldown := &controltowerv1.PmgDependencyCooldown{}
+	if !event.PublishDate.IsZero() {
+		cooldown.SetPublishDate(timestamppb.New(event.PublishDate))
+	}
+	cooldown.SetCooldownDays(uint32(event.CooldownDays))
+	cooldown.SetDaysSincePublish(uint32(event.DaysAgo))
+	cooldown.SetDaysRemaining(uint32(event.DaysLeft))
+	decision.SetCooldown(cooldown)
+
+	e := &controltowerv1.PmgEvent{}
+	e.SetEventType(controltowerv1.PmgEventType_PMG_EVENT_TYPE_PACKAGE_DECISION)
+	e.SetPackageDecision(decision)
+	return e
+}
+
+func newHostObservationEvent(event AuditEvent) *controltowerv1.PmgEvent {
+	obs := &controltowerv1.PmgHostObservation{}
+	obs.SetHostname(event.Hostname)
+	obs.SetMethod(event.Method)
+
+	e := &controltowerv1.PmgEvent{}
+	e.SetEventType(controltowerv1.PmgEventType_PMG_EVENT_TYPE_HOST_OBSERVATION)
+	e.SetHostObservation(obs)
+	return e
+}
+
 func newSessionSummaryEvent(data *SessionData) *controltowerv1.PmgEvent {
 	summary := &controltowerv1.PmgSessionSummary{}
 	summary.SetPackageManager(mapPackageManager(data.PackageManager))
@@ -93,6 +129,7 @@ func newSessionSummaryEvent(data *SessionData) *controltowerv1.PmgEvent {
 	summary.SetBlockedCount(data.BlockedCount)
 	summary.SetConfirmedCount(data.ConfirmedCount)
 	summary.SetTrustedSkipped(data.TrustedSkipped)
+	summary.SetCooldownBlockedCount(data.CooldownBlockedCount)
 	summary.SetDuration(durationpb.New(data.Duration))
 	summary.SetSandboxEnabled(data.SandboxEnabled)
 	summary.SetParanoidMode(data.ParanoidMode)

internal/audit/cloud_translate_test.go

@@ -127,9 +127,57 @@ func TestTranslateErrorNilError(t *testing.T) {
 	assert.Equal(t, "unknown issue", pmgErr.GetMessage())
 }
 
+func TestTranslateCooldownBlocked(t *testing.T) {
+	publishDate := time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC)
+	event := AuditEvent{
+		Type:           EventTypeDependencyCooldown,
+		PackageVersion: testPackageVersion("new-pkg", "1.0.0", "npm"),
+		PublishDate:    publishDate,
+		CooldownDays:   30,
+		DaysAgo:        6,
+		DaysLeft:       24,
+	}
+
+	results := testSink.translateToPmgEvents(event)
+	require.Len(t, results, 1)
+	result := results[0]
+
+	assert.Equal(t, controltowerv1.PmgEventType_PMG_EVENT_TYPE_PACKAGE_DECISION, result.GetEventType())
+	require.True(t, result.HasPackageDecision())
+
+	decision := result.GetPackageDecision()
+	assert.Equal(t, controltowerv1.PmgPackageAction_PMG_PACKAGE_ACTION_COOLDOWN_BLOCKED, decision.GetAction())
+	assert.NotNil(t, decision.GetPackageVersion())
+
+	require.True(t, decision.HasCooldown())
+	cooldown := decision.GetCooldown()
+	assert.Equal(t, publishDate.Unix(), cooldown.GetPublishDate().AsTime().Unix())
+	assert.Equal(t, uint32(30), cooldown.GetCooldownDays())
+	assert.Equal(t, uint32(6), cooldown.GetDaysSincePublish())
+	assert.Equal(t, uint32(24), cooldown.GetDaysRemaining())
+}
+
+func TestTranslateHostObservation(t *testing.T) {
+	event := AuditEvent{
+		Type:     EventTypeProxyHostObserved,
+		Hostname: "evil.example.com",
+		Method:   "CONNECT",
+	}
+
+	results := testSink.translateToPmgEvents(event)
+	require.Len(t, results, 1)
+	result := results[0]
+
+	assert.Equal(t, controltowerv1.PmgEventType_PMG_EVENT_TYPE_HOST_OBSERVATION, result.GetEventType())
+	require.True(t, result.HasHostObservation())
+
+	obs := result.GetHostObservation()
+	assert.Equal(t, "evil.example.com", obs.GetHostname())
+	assert.Equal(t, "CONNECT", obs.GetMethod())
+}
+
 func TestTranslateUnsupportedEventReturnsEmpty(t *testing.T) {
 	unsupported := []EventType{
-		EventTypeProxyHostObserved,
 		EventTypeDependencyResolved,
 		EventTypeInstallStarted,
 		EventTypeInstallAllowed,
@@ -185,6 +233,7 @@ func TestTranslateSessionComplete(t *testing.T) {
 			ConfirmedCount:       1,
 			TrustedSkipped:       2,
 			InsecureBypassed:     0,
+			CooldownBlockedCount: 3,
 			Duration:             5 * time.Second,
 			SandboxEnabled:       true,
 			ParanoidMode:         false,
@@ -204,6 +253,7 @@ func TestTranslateSessionComplete(t *testing.T) {
 	assert.Equal(t, uint32(1), summary.GetBlockedCount())
 	assert.Equal(t, uint32(1), summary.GetConfirmedCount())
 	assert.Equal(t, uint32(2), summary.GetTrustedSkipped())
+	assert.Equal(t, uint32(3), summary.GetCooldownBlockedCount())
 	assert.True(t, summary.GetSandboxEnabled())
 	assert.False(t, summary.GetParanoidMode())
 	assert.True(t, summary.GetTransitiveEnabled())

internal/audit/event.go

@@ -17,6 +17,7 @@ type SessionData struct {
 	ConfirmedCount       uint32
 	TrustedSkipped       uint32
 	InsecureBypassed     uint32
+	CooldownBlockedCount uint32
 	Duration             time.Duration
 	SandboxEnabled       bool
 	ParanoidMode         bool
@@ -54,6 +55,7 @@ const (
 	EventTypeDependencyResolved    EventType = "dependency_resolved"
 	EventTypeInstallInsecureBypass EventType = "install_insecure_bypass"
 	EventTypeProxyHostObserved     EventType = "proxy_host_observed"
+	EventTypeDependencyCooldown    EventType = "dependency_cooldown"
 	EventTypeSandboxOverride       EventType = "sandbox_override"
 	EventTypeError                 EventType = "error"
 	EventTypeSessionComplete       EventType = "session_complete"
@@ -91,6 +93,12 @@ type AuditEvent struct {
 	Method   string
 	Reason   string
 
+	// Cooldown context
+	PublishDate  time.Time
+	CooldownDays int
+	DaysAgo      int
+	DaysLeft     int
+
 	// Error context
 	Error error
 

internal/audit/eventlog_sink.go

@@ -41,6 +41,7 @@ func sessionDataToDetails(sd *SessionData) map[string]interface{} {
 		"confirmed_count":        sd.ConfirmedCount,
 		"trusted_skipped":        sd.TrustedSkipped,
 		"insecure_bypassed":      sd.InsecureBypassed,
+		"cooldown_blocked_count": sd.CooldownBlockedCount,
 	}
 }
 
@@ -66,6 +67,8 @@ func mapEventType(t EventType) eventlog.EventType {
 		return eventlog.EventTypeInstallInsecureBypass
 	case EventTypeProxyHostObserved:
 		return eventlog.EventTypeProxyHostObserved
+	case EventTypeDependencyCooldown:
+		return eventlog.EventTypeDependencyCooldown
 	case EventTypeSandboxOverride:
 		return eventlog.EventTypeSandboxOverride
 	case EventTypeError:

internal/eventlog/eventlog.go

@@ -25,6 +25,7 @@ const (
 	EventTypeDependencyResolved    EventType = "dependency_resolved"
 	EventTypeInstallInsecureBypass EventType = "install_insecure_bypass"
 	EventTypeProxyHostObserved     EventType = "proxy_host_observed"
+	EventTypeDependencyCooldown    EventType = "dependency_cooldown"
 	EventTypeSandboxOverride       EventType = "sandbox_override"
 	EventTypeError                 EventType = "error"
 )

proxy/interceptors/cooldown.go

@@ -1,6 +1,11 @@
 package interceptors
 
-import "time"
+import (
+	"time"
+
+	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
+	"github.com/safedep/pmg/internal/audit"
+)
 
 // cooldownIsWithinWindow reports whether a version published at publishDate is still
 // within the cooldown window of cooldownDays. Returns withinCooldown, daysSincePublish,
@@ -34,21 +39,32 @@ func cooldownOldestVersion(dates map[string]time.Time) (string, time.Time) {
 // recordCooldownStats records a cooldown block event. When all versions are blocked
 // (remaining == 0), it reports the oldest version (closest to exiting cooldown).
 // Otherwise, if a pinned version was stripped, it reports that specific version.
-func recordCooldownStats(statsCollector *AnalysisStatsCollector, packageName string, pinnedVersion string, dates map[string]time.Time, remaining int, cooldownDays int) {
+func recordCooldownStats(statsCollector *AnalysisStatsCollector, ecosystem packagev1.Ecosystem, packageName string, pinnedVersion string, dates map[string]time.Time, remaining int, cooldownDays int) {
 	if statsCollector == nil {
 		return
 	}
 
+	logCooldown := func(version string, publishDate time.Time, daysAgo, daysLeft int) {
+		statsCollector.RecordCooldownBlocked(packageName, version, publishDate, daysAgo, daysLeft, cooldownDays)
+
+		pv := &packagev1.PackageVersion{}
+		pv.SetPackage(&packagev1.Package{})
+		pv.GetPackage().SetName(packageName)
+		pv.GetPackage().SetEcosystem(ecosystem)
+		pv.SetVersion(version)
+		audit.LogDependencyCooldown(pv, publishDate, cooldownDays, daysAgo, daysLeft)
+	}
+
 	if remaining == 0 {
 		oldestVer, oldestDate := cooldownOldestVersion(dates)
 		if oldestVer != "" {
 			_, daysAgo, daysLeft := cooldownIsWithinWindow(oldestDate, cooldownDays)
-			statsCollector.RecordCooldownBlocked(packageName, oldestVer, oldestDate, daysAgo, daysLeft, cooldownDays)
+			logCooldown(oldestVer, oldestDate, daysAgo, daysLeft)
 		}
 	} else if pinnedVersion != "" {
 		if pinnedDate, ok := dates[pinnedVersion]; ok {
 			if withinCooldown, daysAgo, daysLeft := cooldownIsWithinWindow(pinnedDate, cooldownDays); withinCooldown {
-				statsCollector.RecordCooldownBlocked(packageName, pinnedVersion, pinnedDate, daysAgo, daysLeft, cooldownDays)
+				logCooldown(pinnedVersion, pinnedDate, daysAgo, daysLeft)
 			}
 		}
 	}

proxy/interceptors/npm_cooldown.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"time"
 
+	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
 	"github.com/safedep/dry/log"
 	"github.com/safedep/pmg/proxy"
 )
@@ -66,7 +67,7 @@ func (h *npmCooldownHandler) HandleMetadataRequest(ctx *proxy.RequestContext, pa
 			log.Infof("[%s] Cooldown: stripped %d version(s) from %s metadata (%d days, %d eligible remain)",
 				ctx.RequestID, stripped, packageName, cooldownDays, remaining)
 
-			recordCooldownStats(h.statsCollector, packageName, pinnedVersion, dates, remaining, cooldownDays)
+			recordCooldownStats(h.statsCollector, packagev1.Ecosystem_ECOSYSTEM_NPM, packageName, pinnedVersion, dates, remaining, cooldownDays)
 
 			// Prevent npm from caching the modified response. Without this,
 			// npm would serve the stripped metadata from cache even after the

proxy/interceptors/pypi_cooldown.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"time"
 
+	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
 	"github.com/safedep/dry/log"
 	"github.com/safedep/pmg/proxy"
 )
@@ -53,7 +54,7 @@ func (h *pypiCooldownHandler) HandleMetadataRequest(ctx *proxy.RequestContext, p
 			log.Infof("[%s] Cooldown: stripped %d version(s) from %s metadata (%d days, %d eligible remain)",
 				ctx.RequestID, stripped, packageName, cooldownDays, remaining)
 
-			recordCooldownStats(h.statsCollector, packageName, pinnedVersion, dates, remaining, cooldownDays)
+			recordCooldownStats(h.statsCollector, packagev1.Ecosystem_ECOSYSTEM_PYPI, packageName, pinnedVersion, dates, remaining, cooldownDays)
 
 			headers.Set("Cache-Control", "no-store")
 			return statusCode, headers, strippedBody, nil