feat(telemetry): export tool cache job events

CHANGELOG.md

@@ -7,6 +7,8 @@ Versioning.
 
 ### Added
 
+- Added bounded extraction of `nddev_tool_cache_event` records from verified
+  runner diagnostic artifacts, with malformed job text counted but non-blocking.
 - Added the portable scheduler-recovery decision core: exact stuck-dispatch
   detection, startup grace, cooldown, and duplicate-recovery exclusion.
 - Added checkpoint-first recovery orchestration with durable attempt identity,

cmd/gha-diagnostic-exporter/main.go

@@ -73,11 +73,27 @@ func run(args []string, stdout, stderr io.Writer) error {
 		logger.Error("diagnostic export incomplete", "error", err)
 		return err
 	}
+	for _, observed := range summary.ToolCacheEvents {
+		logger.Info(
+			"nddev tool cache event",
+			"event_type", "nddev_tool_cache_event",
+			"repository", observed.Repository,
+			"runner", observed.Runner,
+			"captured_at", observed.CapturedAt,
+			"source", observed.Event.Source,
+			"cache_result", observed.Event.CacheResult,
+			"sha256", observed.Event.SHA256,
+			"bytes", observed.Event.Bytes,
+			"duration_ms", observed.Event.DurationMS,
+		)
+	}
 	logger.Info(
 		"diagnostic export complete",
 		"source_bundles", summary.SourceBundles,
 		"exported_bundles", summary.ExportedBundles,
 		"pending_bundles", summary.PendingBundles,
+		"tool_cache_events", summary.ToolCacheEventCount,
+		"rejected_tool_cache_events", summary.RejectedToolCacheEventCount,
 	)
 	return nil
 }

internal/diagnosticexport/bundle.go

@@ -31,14 +31,16 @@ const (
 )
 
 type Bundle struct {
-	Name       string
-	Content    []byte
-	SHA256     string
-	Manifest   workerdiagnostics.Manifest
-	CapturedAt time.Time
-	ObjectKey  string
-	device     uint64
-	inode      uint64
+	Name                    string
+	Content                 []byte
+	SHA256                  string
+	Manifest                workerdiagnostics.Manifest
+	CapturedAt              time.Time
+	ObjectKey               string
+	ToolCacheEvents         []ToolCacheEvent
+	RejectedToolCacheEvents int
+	device                  uint64
+	inode                   uint64
 }
 
 type bundleScope uint8
@@ -223,15 +225,18 @@ func ReadBundle(ctx context.Context, config Config, name string) (Bundle, error)
 	if err != nil {
 		return Bundle{}, fmt.Errorf("derive diagnostic object key: %w", err)
 	}
+	events, rejectedEvents := ExtractToolCacheEvents(content)
 	return Bundle{
-		Name:       name,
-		Content:    content,
-		SHA256:     digestHex,
-		Manifest:   manifest,
-		CapturedAt: capturedAt,
-		ObjectKey:  objectKey,
-		device:     before.Dev,
-		inode:      before.Ino,
+		Name:                    name,
+		Content:                 content,
+		SHA256:                  digestHex,
+		Manifest:                manifest,
+		CapturedAt:              capturedAt,
+		ObjectKey:               objectKey,
+		ToolCacheEvents:         events,
+		RejectedToolCacheEvents: rejectedEvents,
+		device:                  before.Dev,
+		inode:                   before.Ino,
 	}, nil
 }
 

internal/diagnosticexport/exporter.go

@@ -26,13 +26,23 @@ type ObjectStore interface {
 }
 
 type Summary struct {
-	SourceBundles   int   `json:"source_bundles"`
-	ExportedBundles int   `json:"exported_bundles"`
-	PendingBundles  int   `json:"pending_bundles"`
-	SourceBytes     int64 `json:"source_bytes"`
-	ExportedBytes   int64 `json:"exported_bytes"`
-	ScannedBundles  int   `json:"scanned_bundles"`
-	DeletedBundles  int   `json:"deleted_bundles"`
+	SourceBundles               int                      `json:"source_bundles"`
+	ExportedBundles             int                      `json:"exported_bundles"`
+	PendingBundles              int                      `json:"pending_bundles"`
+	SourceBytes                 int64                    `json:"source_bytes"`
+	ExportedBytes               int64                    `json:"exported_bytes"`
+	ScannedBundles              int                      `json:"scanned_bundles"`
+	DeletedBundles              int                      `json:"deleted_bundles"`
+	ToolCacheEventCount         int                      `json:"tool_cache_event_count"`
+	RejectedToolCacheEventCount int                      `json:"rejected_tool_cache_event_count"`
+	ToolCacheEvents             []ObservedToolCacheEvent `json:"-"`
+}
+
+type ObservedToolCacheEvent struct {
+	Event      ToolCacheEvent
+	Repository string
+	Runner     string
+	CapturedAt time.Time
 }
 
 type Exporter struct {
@@ -123,6 +133,7 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) {
 			}
 			delete(state.Exports, name)
 			summary.DeletedBundles++
+			summary.addToolCacheEvents(bundle)
 			continue
 		}
 		remote, err := e.Store.Head(ctx, e.Config.Bucket, bundle.ObjectKey)
@@ -184,6 +195,7 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) {
 		summary.DeletedBundles++
 		summary.ExportedBundles++
 		summary.ExportedBytes += bundleBytes
+		summary.addToolCacheEvents(bundle)
 	}
 	_, remainingBundles, remainingBytes, scanErr := ListBundleBatch(ctx, e.Config, 1)
 	if scanErr != nil {
@@ -236,6 +248,17 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) {
 	return summary, nil
 }
 
+func (summary *Summary) addToolCacheEvents(bundle Bundle) {
+	summary.RejectedToolCacheEventCount += bundle.RejectedToolCacheEvents
+	for _, event := range bundle.ToolCacheEvents {
+		summary.ToolCacheEvents = append(summary.ToolCacheEvents, ObservedToolCacheEvent{
+			Event: event, Repository: bundle.Manifest.Instance.Repository,
+			Runner: bundle.Manifest.Instance.Name, CapturedAt: bundle.CapturedAt,
+		})
+		summary.ToolCacheEventCount++
+	}
+}
+
 func (e Exporter) saveFailure(
 	state State,
 	now time.Time,

internal/diagnosticexport/toolcache.go

@@ -0,0 +1,81 @@
+package diagnosticexport
+
+import (
+	"archive/tar"
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"strings"
+)
+
+const (
+	toolCacheMarker    = "nddev_tool_cache_event="
+	maxToolCacheEvents = 64
+	maxToolCacheLine   = 16 * 1024
+)
+
+type ToolCacheEvent struct {
+	Source      string `json:"source"`
+	CacheResult string `json:"cache_result"`
+	SHA256      string `json:"sha256"`
+	Bytes       int64  `json:"bytes"`
+	DurationMS  int64  `json:"duration_ms"`
+}
+
+func ExtractToolCacheEvents(content []byte) ([]ToolCacheEvent, int) {
+	compressed := bytes.NewReader(content)
+	decompressor, err := gzip.NewReader(compressed)
+	if err != nil {
+		return nil, 1
+	}
+	defer decompressor.Close()
+	archive := tar.NewReader(decompressor)
+	events := make([]ToolCacheEvent, 0)
+	rejected := 0
+	for len(events) < maxToolCacheEvents {
+		header, err := archive.Next()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return events, rejected + 1
+		}
+		if header.Typeflag != tar.TypeReg || !strings.HasPrefix(header.Name, "runner/") {
+			continue
+		}
+		scanner := bufio.NewScanner(io.LimitReader(archive, 2*1024*1024))
+		scanner.Buffer(make([]byte, 4096), maxToolCacheLine)
+		for scanner.Scan() && len(events) < maxToolCacheEvents {
+			line := scanner.Text()
+			index := strings.Index(line, toolCacheMarker)
+			if index < 0 {
+				continue
+			}
+			var event ToolCacheEvent
+			decoder := json.NewDecoder(strings.NewReader(line[index+len(toolCacheMarker):]))
+			decoder.DisallowUnknownFields()
+			if err := decoder.Decode(&event); err != nil || !validToolCacheEvent(event) {
+				rejected++
+				continue
+			}
+			events = append(events, event)
+		}
+		if scanner.Err() != nil {
+			rejected++
+		}
+	}
+	return events, rejected
+}
+
+func validToolCacheEvent(event ToolCacheEvent) bool {
+	if event.Source == "" || len(event.Source) > 32 || event.CacheResult == "" || len(event.CacheResult) > 96 ||
+		event.Bytes < 0 || event.DurationMS < 0 || len(event.SHA256) != 64 {
+		return false
+	}
+	_, err := hex.DecodeString(event.SHA256)
+	return err == nil
+}

internal/diagnosticexport/toolcache_test.go

@@ -0,0 +1,67 @@
+package diagnosticexport
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/NDDev-OpenNetwork/github-actions/internal/workerdiagnostics"
+)
+
+func TestExtractToolCacheEventsFromRunnerArtifacts(t *testing.T) {
+	t.Parallel()
+	archive := toolCacheArchive(t, map[string]string{
+		"runner/Worker.log": "prefix nddev_tool_cache_event={\"source\":\"upstream\",\"cache_result\":\"miss\",\"sha256\":\"" + strings.Repeat("a", 64) + "\",\"bytes\":42,\"duration_ms\":7}\nnddev_tool_cache_event={bad}\n",
+		"incus/qemu.log":    "nddev_tool_cache_event={\"source\":\"ignored\"}\n",
+	})
+	events, rejected := ExtractToolCacheEvents(archive)
+	if len(events) != 1 || rejected != 1 {
+		t.Fatalf("events=%#v rejected=%d", events, rejected)
+	}
+	if events[0].Source != "upstream" || events[0].Bytes != 42 || events[0].DurationMS != 7 {
+		t.Fatalf("event=%#v", events[0])
+	}
+}
+
+func TestSummaryBindsToolCacheEventToBundleIdentity(t *testing.T) {
+	t.Parallel()
+	var summary Summary
+	captured := time.Date(2026, 8, 24, 10, 0, 0, 0, time.UTC)
+	summary.addToolCacheEvents(Bundle{
+		Manifest:   workerdiagnostics.Manifest{Instance: workerdiagnostics.Instance{Name: "runner-1", Repository: "example/repository"}},
+		CapturedAt: captured, RejectedToolCacheEvents: 2,
+		ToolCacheEvents: []ToolCacheEvent{{Source: "cache", CacheResult: "hit", SHA256: strings.Repeat("a", 64)}},
+	})
+	if summary.ToolCacheEventCount != 1 || summary.RejectedToolCacheEventCount != 2 || len(summary.ToolCacheEvents) != 1 {
+		t.Fatalf("summary=%#v", summary)
+	}
+	observed := summary.ToolCacheEvents[0]
+	if observed.Repository != "example/repository" || observed.Runner != "runner-1" || !observed.CapturedAt.Equal(captured) {
+		t.Fatalf("observed=%#v", observed)
+	}
+}
+
+func toolCacheArchive(t *testing.T, files map[string]string) []byte {
+	t.Helper()
+	var buffer bytes.Buffer
+	compressed := gzip.NewWriter(&buffer)
+	archive := tar.NewWriter(compressed)
+	for name, content := range files {
+		if err := archive.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(content))}); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := archive.Write([]byte(content)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := archive.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := compressed.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return buffer.Bytes()
+}

internal/schedulerrecovery/command_test.go

@@ -47,11 +47,13 @@ func TestCommandExecutorRejectsShellAndTimeouts(t *testing.T) {
 func writeExecutable(t *testing.T, directory, name, content string) string {
 	t.Helper()
 	path := filepath.Join(directory, name)
-	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o700)
+	temporary := path + ".tmp"
+	file, err := os.OpenFile(temporary, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o700)
 	require.NoError(t, err)
 	_, err = file.WriteString(content)
 	require.NoError(t, err)
 	require.NoError(t, file.Sync())
 	require.NoError(t, file.Close())
+	require.NoError(t, os.Rename(temporary, path))
 	return path
 }