fix(worker): claim cached ephemeral inputs

Keep startup-scanned cache hits owned while inference uses them and reconcile their actual size against capacity.

Assisted-by: Codex:gpt-6

core/services/worker/ephemeral_capacity.go

@@ -257,6 +257,66 @@ func (g *EphemeralCapacityGuard) Commit(path string) error {
 	return nil
 }
 
+// Claim makes an existing regular file request-owned until Release. It
+// serializes with recovery deletion, preserves bytes already discovered by a
+// startup scan, and only admits growth that fits the configured capacity.
+// Repeated claims of the same committed file are idempotent.
+func (g *EphemeralCapacityGuard) Claim(path string) error {
+	cleanPath, root, err := g.registeredPath(path)
+	if err != nil {
+		return err
+	}
+
+	g.mu.Lock()
+	defer g.mu.Unlock()
+
+	info, err := os.Lstat(cleanPath)
+	if err != nil {
+		return fmt.Errorf("stating claimed ephemeral file %q: %w", cleanPath, err)
+	}
+	if !info.Mode().IsRegular() {
+		return fmt.Errorf("claimed ephemeral path %q is not a regular file", cleanPath)
+	}
+
+	entry, found := g.entries[cleanPath]
+	if found && entry.isOwned() && entry.state != ephemeralCapacityCommitted {
+		return &EphemeralReservationConflictError{
+			Path: cleanPath, ActiveBytes: entry.reserved, RequestedBytes: info.Size(),
+		}
+	}
+
+	accounted := int64(0)
+	if found {
+		accounted = entry.baseline + entry.reserved
+	}
+	delta := info.Size() - accounted
+	if delta > 0 {
+		if err := g.checkCapacityLocked(root, delta); err != nil {
+			// The file already occupies the filesystem, so keep accounting
+			// truthful even though a new request cannot claim it. Preserve
+			// ownership if an earlier claim is still awaiting Release.
+			state := ephemeralCapacityExisting
+			owned := false
+			if found && entry.state == ephemeralCapacityCommitted && entry.isOwned() {
+				state = ephemeralCapacityCommitted
+				owned = true
+			}
+			entry = ephemeralCapacityEntry{
+				state: state, owned: owned, baseline: info.Size(),
+			}
+			g.entries[cleanPath] = entry
+			g.usage += delta
+			return err
+		}
+	}
+	g.usage += delta
+	entry = ephemeralCapacityEntry{
+		state: ephemeralCapacityCommitted, owned: true, baseline: info.Size(),
+	}
+	g.entries[cleanPath] = entry
+	return nil
+}
+
 // Release forgets all accounting for path. It is safe to call repeatedly.
 func (g *EphemeralCapacityGuard) Release(path string) error {
 	cleanPath, _, err := g.registeredPath(path)

core/services/worker/file_staging.go

@@ -287,7 +287,7 @@ func ensureWorkerFile(ctx context.Context, fm *storage.FileManager, capacity *Ep
 		if !info.Mode().IsRegular() {
 			return "", fmt.Errorf("ephemeral cache path %q is not a regular file", cachePath)
 		}
-		if err := capacity.Account(cachePath, info.Size()); err != nil {
+		if err := capacity.Claim(cachePath); err != nil {
 			return "", err
 		}
 		return cachePath, nil

core/services/worker/file_staging_release_test.go

@@ -70,6 +70,79 @@ func (m *releaseMessagingClient) IsConnected() bool { return true }
 func (m *releaseMessagingClient) Close()            {}
 
 var _ = Describe("Worker exact-key staging release", func() {
+	It("claims a startup-scanned cache hit against stale recovery until release", func() {
+		cacheDir := GinkgoT().TempDir()
+		root := filepath.Join(cacheDir, "ephemeral")
+		key := "ephemeral/audio/request-id/input.wav"
+		cachePath := filepath.Join(cacheDir, filepath.FromSlash(key))
+		Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed())
+		Expect(os.WriteFile(cachePath, []byte("data"), 0o600)).To(Succeed())
+		old := time.Now().Add(-2 * time.Hour)
+		Expect(os.Chtimes(cachePath, old, old)).To(Succeed())
+		Expect(os.Chtimes(filepath.Dir(cachePath), old, old)).To(Succeed())
+
+		guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0)
+		Expect(err).NotTo(HaveOccurred())
+		store := &stagingObjectStore{payload: []byte("unused")}
+		fm, err := storage.NewFileManager(store, cacheDir)
+		Expect(err).NotTo(HaveOccurred())
+
+		localPath, err := ensureWorkerFile(context.Background(), fm, guard, key)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(localPath).To(Equal(cachePath))
+		Expect(store.getCalls).To(BeZero())
+		CleanEphemeralRoots([]string{root}, time.Hour, guard)
+		Expect(cachePath).To(BeAnExistingFile())
+
+		Expect(guard.Release(cachePath)).To(Succeed())
+		CleanEphemeralRoots([]string{root}, time.Hour, guard)
+		Expect(cachePath).NotTo(BeAnExistingFile())
+	})
+
+	It("makes repeated cache-hit claims idempotent", func() {
+		cacheDir := GinkgoT().TempDir()
+		root := filepath.Join(cacheDir, "ephemeral")
+		key := "ephemeral/audio/request-id/input.wav"
+		cachePath := filepath.Join(cacheDir, filepath.FromSlash(key))
+		Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed())
+		Expect(os.WriteFile(cachePath, []byte("data"), 0o600)).To(Succeed())
+		guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0)
+		Expect(err).NotTo(HaveOccurred())
+		fm, err := storage.NewFileManager(&stagingObjectStore{}, cacheDir)
+		Expect(err).NotTo(HaveOccurred())
+
+		for range 2 {
+			localPath, ensureErr := ensureWorkerFile(context.Background(), fm, guard, key)
+			Expect(ensureErr).NotTo(HaveOccurred())
+			Expect(localPath).To(Equal(cachePath))
+		}
+		err = guard.Reserve(filepath.Join(root, "other", "request-id", "input.wav"), 1)
+		var capacityErr *EphemeralCapacityError
+		Expect(errors.As(err, &capacityErr)).To(BeTrue())
+		Expect(capacityErr.UsageBytes).To(Equal(int64(4)))
+	})
+
+	It("capacity-checks growth of a startup-scanned cache file", func() {
+		cacheDir := GinkgoT().TempDir()
+		root := filepath.Join(cacheDir, "ephemeral")
+		key := "ephemeral/audio/request-id/input.wav"
+		cachePath := filepath.Join(cacheDir, filepath.FromSlash(key))
+		Expect(os.MkdirAll(filepath.Dir(cachePath), 0o750)).To(Succeed())
+		Expect(os.WriteFile(cachePath, []byte("12"), 0o600)).To(Succeed())
+		guard, err := NewEphemeralCapacityGuard([]string{root}, 4, 0)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(os.WriteFile(cachePath, []byte("12345"), 0o600)).To(Succeed())
+		fm, err := storage.NewFileManager(&stagingObjectStore{}, cacheDir)
+		Expect(err).NotTo(HaveOccurred())
+
+		_, err = ensureWorkerFile(context.Background(), fm, guard, key)
+		var capacityErr *EphemeralCapacityError
+		Expect(errors.As(err, &capacityErr)).To(BeTrue())
+		Expect(capacityErr.RequestedBytes).To(Equal(int64(3)))
+		Expect(capacityErr.UsageBytes).To(Equal(int64(2)))
+		Expect(guard.HasActiveReservation(cachePath)).To(BeFalse())
+	})
+
 	It("reserves S3 object size before download and releases it with the exact key", func() {
 		cacheDir := GinkgoT().TempDir()
 		root := filepath.Join(cacheDir, "ephemeral")