fix(sandbox): skip identical skill installs and prune old snapshots (#2791)

* fix(sandbox): skip identical skill installs and prune old snapshots

Re-uploading the same archive of a ready skill was still booting a billed sandbox and growing a new generation. Superceded snapshots also accumulated on the provider because the ledger never deleted them.

Skip the rebuild when the digest already matches a skill still in the live image, and let the reaper delete superseded provider snapshots after 24h while leaving extras not in the ledger untouched.

* fix(sandbox): retry stale same-zip installs and prune active leftovers

A re-upload of the same archive skipped every installing row, so a dead process could not recover until the reaper. Skip only recent in-flight runs, still refresh the stored bundle, and prune non-live active snapshots after retention.

* fix(sandbox): beat install liveness and guard the snapshot prune

A fixed window on the submission time cannot tell a slow install from a
dead one: a single agent command may take installCommandTimeout and an
install runs several of them, so a two-minute window called live runs
dead, reset their progress and pushed the reaper's own deadline out.
Stamp InstallingSince while the run works and read that instead, which
also stops the reaper from failing installs that are merely long.

Answer the skip question from the image only for a ready row. The ledger
records which skill a snapshot carries, not which archive, so a row
installing a new bundle would look finished from the previous bundle's
snapshot and report a success that never happened.

Prune now refuses to delete once the credentials no longer address the
account that built the image: the delete would come back not-found, and
the ledger would record another account's still-billed snapshots as
gone. It also waits for an eligible row before building a provider
client, since most configs have nothing to prune on most sweeps.

Author: lyingbug

internal/application/service/tenant_skill_bundle.go

@@ -36,7 +36,8 @@ type SkillBundle struct {
 	Description  string
 	Instructions string
 	// SHA256 is over the uploaded bytes, so re-uploading the same archive is
-	// recognisable in the UI and in the ledger.
+	// recognisable in the UI and in the ledger, and a ready skill with this
+	// digest can skip a billed snapshot rebuild.
 	SHA256 string
 	// Files maps skill-root-relative paths to contents, SKILL.md included.
 	Files map[string][]byte

internal/application/service/tenant_skill_install.go

@@ -8,6 +8,7 @@ import (
 	"path"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -63,6 +64,12 @@ func (s *TenantSkillService) InstallSkill(
 	if err != nil {
 		return "", err
 	}
+	if s.canSkipInstall(ctx, existing, bundle) {
+		if err := s.refreshSkippedBundle(ctx, existing, archive); err != nil {
+			return "", fmt.Errorf("store bundle for skill %s: %w", existing.ID, err)
+		}
+		return existing.ID, nil
+	}
 
 	skillID := uuid.NewString()
 	now := s.now()
@@ -201,6 +208,14 @@ func (s *TenantSkillService) runInstall(
 		return nil
 	}
 
+	// From here on this run is the row's owner, and everything below can take
+	// minutes. The heartbeat is what tells a second upload of the same archive
+	// (and the reaper) that those minutes are work rather than a dead process.
+	// It is deferred before it is stopped explicitly below, so a failure path
+	// still stops it ahead of the deferred failSkill.
+	stopHeartbeat := s.startInstallHeartbeat(ctx, tenantID, configID, skillID)
+	defer stopHeartbeat()
+
 	// The name comes from SKILL.md and is already validated on parse, so a
 	// rejection here means the bundle was accepted by a looser rule than the
 	// one the image path enforces. Failing before any sandbox work keeps that
@@ -349,6 +364,10 @@ func (s *TenantSkillService) runInstall(
 		return err
 	}
 	pointerSwitched = true
+	// The heartbeat writes the whole row, so it must be gone before the
+	// terminal "ready" write below: a beat landing after it would put the row
+	// back to installing and have the reaper fail a skill that is serving.
+	stopHeartbeat()
 	s.markPreviousSnapshotsSuperseded(ctx, tenantID, configID, installRowID)
 
 	// The terminal write is the one that must not be best-effort: the pointer
@@ -873,9 +892,10 @@ func (s *TenantSkillService) failSkill(
 // installStillOwnsTheRow is the lock-side counterpart of InstallSkill's
 // optimistic row write. A remove that ran first deleted the row; a newer
 // upload of the same name replaced BundleSHA256; a queued remove flipped the
-// status. Any of those means this run must not snapshot — failSkill would
-// stamp the newer owner's row, and a snapshot with no matching row is an
-// orphan the ledger cannot name.
+// status; a sibling retry of the same archive found the first run had already
+// landed in the live image. Any of those means this run must not snapshot —
+// failSkill would stamp the newer owner's row, and a snapshot with no matching
+// row is an orphan the ledger cannot name.
 func (s *TenantSkillService) installStillOwnsTheRow(
 	ctx context.Context, tenantID uint64, configID, skillID string, bundle *SkillBundle,
 ) (bool, error) {
@@ -892,9 +912,147 @@ func (s *TenantSkillService) installStillOwnsTheRow(
 	if bundle != nil && current.BundleSHA256 != "" && current.BundleSHA256 != bundle.SHA256 {
 		return false, nil
 	}
+	if current.Status == types.SkillStatusReady {
+		_, inImage, ok := s.skillFilesInLiveImage(ctx, current)
+		if ok && inImage {
+			return false, nil
+		}
+	}
 	return true, nil
 }
 
+// canSkipInstall reports whether this upload is a no-op. Re-uploading the
+// exact archive of a skill that is already ready (and still in the live image)
+// must not boot a billed sandbox or grow a new snapshot. An install of the
+// same bytes that is still beating is the same situation: the first run owns
+// the work.
+//
+// Only a ready row is answered from the image. An installing row is answered
+// from the heartbeat alone, deliberately: the ledger records which skill an
+// install snapshotted, not which archive, so a row that is installing bundle
+// B while the image still carries the earlier bundle A would look "already
+// installed" and this upload would report a success that never happened.
+//
+// A failed skill with the same digest is a retry: the previous attempt never
+// made it into the image. A removal in flight is not a skip either — taking
+// the row back to installing is how an upload cancels it.
+func (s *TenantSkillService) canSkipInstall(
+	ctx context.Context, existing *types.TenantSkillEntity, bundle *SkillBundle,
+) bool {
+	if existing == nil || bundle == nil {
+		return false
+	}
+	if existing.BundleSHA256 == "" || existing.BundleSHA256 != bundle.SHA256 {
+		return false
+	}
+	switch existing.Status {
+	case types.SkillStatusInstalling:
+		return s.installIsInFlight(existing)
+	case types.SkillStatusReady:
+		_, inImage, ok := s.skillFilesInLiveImage(ctx, existing)
+		return ok && inImage
+	default:
+		return false
+	}
+}
+
+// installIsInFlight reports whether an installing row still belongs to a live
+// process. The answer is the heartbeat: a running install restamps
+// InstallingSince every skillInstallHeartbeatInterval, so silence past
+// skillInstallInFlightSkip means the process is gone and the next upload must
+// be allowed to start a new run rather than wait for the stuck-run reaper.
+//
+// Reading the submission time instead would force a choice between calling a
+// slow install dead — a single agent command may take installCommandTimeout,
+// and an install runs several — and leaving a dead one unrecoverable.
+func (s *TenantSkillService) installIsInFlight(existing *types.TenantSkillEntity) bool {
+	if existing == nil || existing.InstallingSince == nil {
+		return false
+	}
+	return !existing.InstallingSince.Before(s.clock()().Add(-skillInstallInFlightSkip))
+}
+
+// startInstallHeartbeat keeps this run's liveness visible while it works, and
+// returns the stop function that must be called before any terminal write.
+//
+// The heartbeat writes the whole row, so it would otherwise race the "ready"
+// write past the pointer switch and revive an installing status. Both callers
+// stop it before that point: runInstall stops it the moment the pointer moves,
+// and the deferred stop runs before the deferred failSkill.
+func (s *TenantSkillService) startInstallHeartbeat(
+	ctx context.Context, tenantID uint64, configID, skillID string,
+) func() {
+	interval := s.installHeartbeat
+	if interval <= 0 {
+		interval = skillInstallHeartbeatInterval
+	}
+	beatCtx, stop := context.WithCancel(ctx)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-beatCtx.Done():
+				return
+			case <-ticker.C:
+				s.beatInstallHeartbeat(beatCtx, tenantID, configID, skillID)
+			}
+		}
+	}()
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			stop()
+			<-done
+		})
+	}
+}
+
+// beatInstallHeartbeat stamps InstallingSince for a row this run still owns.
+// A row that has left the installing status belongs to a newer upload, a
+// queued removal, or a finished run, and reviving its timestamp would hide
+// one of those from the reaper.
+func (s *TenantSkillService) beatInstallHeartbeat(
+	ctx context.Context, tenantID uint64, configID, skillID string,
+) {
+	current, err := s.skills.GetSkill(ctx, tenantID, configID, skillID)
+	if err != nil {
+		logger.Warnf(ctx, "[skill] load %s for install heartbeat failed: %v", skillID, err)
+		return
+	}
+	if current == nil || current.Status != types.SkillStatusInstalling {
+		return
+	}
+	at := s.clock()()
+	current.InstallingSince = &at
+	if err := s.skills.UpdateSkill(ctx, current); err != nil {
+		logger.Warnf(ctx, "[skill] install heartbeat for %s failed: %v", skillID, err)
+	}
+}
+
+// refreshSkippedBundle stores the uploaded archive even when the image work
+// is skipped. read_skill serves file contents from it, so a re-upload of a
+// ready skill is how a missing object-store blob gets repaired without
+// growing a new snapshot. A failure here is returned to the caller rather
+// than turning the ready row into a failed install.
+func (s *TenantSkillService) refreshSkippedBundle(
+	ctx context.Context, existing *types.TenantSkillEntity, archive []byte,
+) error {
+	if existing == nil {
+		return nil
+	}
+	ref, err := s.saveBundle(ctx, existing.TenantID, existing.ID, archive)
+	if err != nil {
+		return err
+	}
+	return s.updateSkillFields(ctx, existing.TenantID, existing.SandboxConfigID, existing.ID,
+		func(e *types.TenantSkillEntity) {
+			e.BundleRef = ref
+		})
+}
+
 // startMaintenanceSession opens the session one image operation runs in. The
 // operation name is carried into the session because the transcript is kept
 // deliberately, for troubleshooting: filing a removal's under "Skill install"