diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go
index f65aa9356..415f494a8 100644
--- a/cmd/ateom-gvisor/main.go
+++ b/cmd/ateom-gvisor/main.go
@@ -34,6 +34,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/compute/metadata"
+	"github.com/agent-substrate/substrate/cmd/ateom-gvisor/internal/cgroupstats"
 	"github.com/agent-substrate/substrate/internal/actorlog"
 	"github.com/agent-substrate/substrate/internal/ateinterceptors"
 	"github.com/agent-substrate/substrate/internal/ateomnet"
@@ -365,6 +366,12 @@ type AteomService struct {
 	// own cgroup scope, which setupCgroupDelegation prepares. A field rather
 	// than a constant so tests can point GetWorkloadStats at a fixture tree.
 	cgroupRoot string
+
+	// readSandboxCgroup overrides cgroupstats.Read when set. Only tests set it:
+	// it is the seam that lets them interleave a lifecycle transition with the
+	// stats handlers' lock-free read, the way containerStatsReader does for the
+	// micro-VM runtime. nil means the real read.
+	readSandboxCgroup func(dir string) (cgroupstats.Sample, error)
 }
 
 var _ ateompb.AteomServer = (*AteomService)(nil)
diff --git a/cmd/ateom-gvisor/stats.go b/cmd/ateom-gvisor/stats.go
index 4514af413..68a320059 100644
--- a/cmd/ateom-gvisor/stats.go
+++ b/cmd/ateom-gvisor/stats.go
@@ -27,6 +27,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/agent-substrate/substrate/cmd/ateom-gvisor/internal/cgroupstats"
+	"github.com/agent-substrate/substrate/internal/ateomstats"
 	"github.com/agent-substrate/substrate/internal/proto/ateompb"
 )
 
@@ -112,17 +113,16 @@ func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWor
 		return nil, status.Errorf(codes.NotFound, "ateom is executing actor %q, not the requested %q", active.UID, req.GetActorUid())
 	}
 
-	observedAt := time.Now()
-	sample, err := cgroupstats.Read(filepath.Join(s.cgroupRoot, sandboxCgroupContainer))
+	sample, err := s.sampleSandbox(active)
 	if err != nil {
-		// The requested actor is the active one but its cgroup is not there. Most
-		// often that is a poll landing in the boot: the ateom retains the
+		// The requested actor is the active one but its cgroup is not there.
+		// Most often that is a poll landing in the boot: the ateom retains the
 		// attribution from the moment it accepts the actor, before runsc has
-		// created the leaf. The other way in is a sandbox that went away between
-		// the check above and the read, which the next CheckpointWorkload turns
-		// into the NOT_FOUND above. Either way it is "no numbers right now" and the
-		// caller should take the next sample, so FAILED_PRECONDITION. Anything else
-		// is a real read failure.
+		// created the leaf. The other way in is a sandbox that went away
+		// underneath the read, which the next CheckpointWorkload turns into the
+		// NOT_FOUND above. Either way it is "no numbers right now" and the
+		// caller should take the next sample, so FAILED_PRECONDITION. Anything
+		// else is a real read failure.
 		if errors.Is(err, fs.ErrNotExist) {
 			return nil, status.Error(codes.FailedPrecondition, "no sandbox cgroup to measure yet")
 		}
@@ -144,7 +144,75 @@ func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWor
 		return nil, status.Errorf(codes.NotFound, "ateom stopped executing actor %q while the sample was being taken", req.GetActorUid())
 	}
 
-	return &ateompb.GetWorkloadStatsResponse{
+	return &ateompb.GetWorkloadStatsResponse{Sample: sample}, nil
+}
+
+// GetActiveWorkloadStats implements
+// ateompb.Ateom/GetActiveWorkloadStats: the discovery read, sampling
+// whatever is executing with no identity asserted. Same lock discipline as
+// GetWorkloadStats above, for the same reasons.
+func (s *AteomService) GetActiveWorkloadStats(ctx context.Context, req *ateompb.GetActiveWorkloadStatsRequest) (*ateompb.GetActiveWorkloadStatsResponse, error) {
+	active := s.activeActor.Load()
+	if active == nil {
+		return noSample(ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD), nil
+	}
+
+	sample, err := s.sampleSandbox(active)
+	if err != nil {
+		// A missing cgroup is a workload with no numbers yet -- a poll landing
+		// in the boot -- which for a caller with no prior knowledge is as
+		// normal a finding as an available ateom, so it is a reason, not an
+		// error. Anything else is a real read failure.
+		if errors.Is(err, fs.ErrNotExist) {
+			return noSample(ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET), nil
+		}
+		return nil, status.Errorf(codes.Internal, "reading sandbox cgroup: %v", err)
+	}
+
+	// Same re-check as GetWorkloadStats, different answer: with no uid asserted
+	// there is no "requested actor" for NOT_FOUND to disown, and a transition
+	// underneath the read just means these numbers cannot be attributed to any
+	// single actor. Report the reason as of now -- the next tick resolves it
+	// either way.
+	if latest := s.activeActor.Load(); latest != active {
+		reason := ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET
+		if latest == nil {
+			reason = ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD
+		}
+		return noSample(reason), nil
+	}
+
+	return &ateompb.GetActiveWorkloadStatsResponse{
+		Result: &ateompb.GetActiveWorkloadStatsResponse_Sample{Sample: sample},
+	}, nil
+}
+
+// noSample is the discovery read's "nothing to give, and that is normal"
+// answer.
+func noSample(reason ateompb.NoSampleReason) *ateompb.GetActiveWorkloadStatsResponse {
+	return &ateompb.GetActiveWorkloadStatsResponse{
+		Result: &ateompb.GetActiveWorkloadStatsResponse_NoSampleReason{NoSampleReason: reason},
+	}
+}
+
+// sampleSandbox reads the sandbox cgroup and builds the sample attributed to
+// active. Errors come back raw -- notably fs.ErrNotExist for a cgroup that is
+// not there yet -- because the two RPCs disagree on what that means: an error
+// code for the keyed read, a normal EXECUTING answer for the discovery read.
+// Callers re-check s.activeActor against the pointer they loaded after this
+// returns; the read holds no lock.
+func (s *AteomService) sampleSandbox(active *ateomstats.ActorAttribution) (*ateompb.WorkloadStatsSample, error) {
+	read := s.readSandboxCgroup
+	if read == nil {
+		read = cgroupstats.Read
+	}
+	observedAt := time.Now()
+	sample, err := read(filepath.Join(s.cgroupRoot, sandboxCgroupContainer))
+	if err != nil {
+		return nil, err
+	}
+
+	return &ateompb.WorkloadStatsSample{
 		Atespace:               active.Ref.Atespace,
 		ActorName:              active.Ref.Name,
 		ActorUid:               active.UID,
diff --git a/cmd/ateom-gvisor/stats_test.go b/cmd/ateom-gvisor/stats_test.go
index 09915f4b8..b46ff5e90 100644
--- a/cmd/ateom-gvisor/stats_test.go
+++ b/cmd/ateom-gvisor/stats_test.go
@@ -28,6 +28,7 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/testing/protocmp"
 
+	"github.com/agent-substrate/substrate/cmd/ateom-gvisor/internal/cgroupstats"
 	"github.com/agent-substrate/substrate/internal/ateomstats"
 	"github.com/agent-substrate/substrate/internal/proto/ateompb"
 	"github.com/agent-substrate/substrate/internal/resources"
@@ -89,13 +90,13 @@ func TestGetWorkloadStats(t *testing.T) {
 		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
 	}
 
-	if got.GetObservedAtUnixNano() < before || got.GetObservedAtUnixNano() > after {
-		t.Errorf("GetWorkloadStats() observed_at_unix_nano = %d, want within [%d, %d]", got.GetObservedAtUnixNano(), before, after)
+	if got.GetSample().GetObservedAtUnixNano() < before || got.GetSample().GetObservedAtUnixNano() > after {
+		t.Errorf("GetWorkloadStats() observed_at_unix_nano = %d, want within [%d, %d]", got.GetSample().GetObservedAtUnixNano(), before, after)
 	}
 	// Checked above; zeroed so the rest can be compared as a whole.
-	got.ObservedAtUnixNano = 0
+	got.GetSample().ObservedAtUnixNano = 0
 
-	want := &ateompb.GetWorkloadStatsResponse{
+	want := &ateompb.GetWorkloadStatsResponse{Sample: &ateompb.WorkloadStatsSample{
 		Atespace:               "space-a",
 		ActorName:              "actor-a",
 		ActorUid:               "uid-a",
@@ -107,7 +108,7 @@ func TestGetWorkloadStats(t *testing.T) {
 		MemoryPeakBytes:        209715200,
 		MemoryWorkingSetBytes:  136314880,
 		CpuUsageUsec:           1234567,
-	}
+	}}
 	if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" {
 		t.Errorf("GetWorkloadStats() mismatch (-want +got):\n%s", diff)
 	}
@@ -218,3 +219,122 @@ func TestAteomServiceStartsAvailable(t *testing.T) {
 		t.Errorf("new AteomService.activeActor = %v, want nil", got)
 	}
 }
+
+func TestGetActiveWorkloadStats(t *testing.T) {
+	s := newStatsService(t, healthyCgroup)
+	s.activeActor.Store(&testActor)
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() error = %v, want nil", err)
+	}
+	if got.GetSample() == nil {
+		t.Fatalf("GetActiveWorkloadStats() = %v, want a sample", got)
+	}
+
+	// The keyed read against the same fixture is the reference: the discovery
+	// read must produce the identical sample, since both are the same
+	// measurement with a different addressing mode.
+	want, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"})
+	if err != nil {
+		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
+	}
+	sample := got.GetSample()
+	sample.ObservedAtUnixNano = 0
+	want.GetSample().ObservedAtUnixNano = 0
+	if diff := cmp.Diff(want.GetSample(), sample, protocmp.Transform()); diff != "" {
+		t.Errorf("discovery sample differs from keyed sample (-keyed +discovery):\n%s", diff)
+	}
+}
+
+// TestGetActiveWorkloadStatsAvailable pins the contract that makes the
+// discovery read scrapeable: an idle ateom is a reason, never an error.
+func TestGetActiveWorkloadStatsAvailable(t *testing.T) {
+	s := newStatsService(t, healthyCgroup)
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() on an available ateom: error = %v, want nil", err)
+	}
+	if got.GetNoSampleReason() != ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD {
+		t.Errorf("GetActiveWorkloadStats() = %v, want NO_WORKLOAD reason", got)
+	}
+}
+
+// TestGetActiveWorkloadStatsBooting: executing but nothing to measure yet is
+// a NOT_MEASURABLE_YET reason, not an error, unlike the keyed read's
+// FAILED_PRECONDITION. A blind caller finds boots as routinely as idle
+// workers.
+func TestGetActiveWorkloadStatsBooting(t *testing.T) {
+	s := newStatsService(t, nil) // no cgroup directory: a poll landing mid-boot
+	s.activeActor.Store(&testActor)
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() mid-boot: error = %v, want nil", err)
+	}
+	if got.GetNoSampleReason() != ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET {
+		t.Errorf("GetActiveWorkloadStats() mid-boot = %v, want NOT_MEASURABLE_YET reason", got)
+	}
+}
+
+// The transition tests cover the re-check that runs after the lock-free
+// measurement, via the readSandboxCgroup seam: flipping activeActor inside the
+// read lands in exactly the window a checkpoint plus a fresh run (or a
+// checkpoint alone) can land in.
+
+func TestGetActiveWorkloadStatsTransition(t *testing.T) {
+	otherActor := testActor
+	otherActor.UID = "uid-b"
+
+	tests := []struct {
+		name string
+		to   *ateomstats.ActorAttribution
+		want ateompb.NoSampleReason
+	}{
+		// A new actor took the slot: there is a workload, its numbers are just
+		// not attributable this tick.
+		{name: "to another actor", to: &otherActor, want: ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET},
+		// A checkpoint emptied the slot: report what is true now.
+		{name: "to available", to: nil, want: ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := newStatsService(t, healthyCgroup)
+			s.activeActor.Store(&testActor)
+			s.readSandboxCgroup = func(dir string) (cgroupstats.Sample, error) {
+				s.activeActor.Store(tc.to)
+				return cgroupstats.Read(dir)
+			}
+
+			got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+			if err != nil {
+				t.Fatalf("GetActiveWorkloadStats() during transition: error = %v, want nil", err)
+			}
+			if got.GetSample() != nil {
+				t.Errorf("GetActiveWorkloadStats() during transition returned sample %v, want none", got.GetSample())
+			}
+			if got.GetNoSampleReason() != tc.want {
+				t.Errorf("GetActiveWorkloadStats() during transition = %v, want %v reason", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestGetWorkloadStatsTransition pins the keyed read's side of the same
+// window: the caller asserted an actor that is gone by the time the sample
+// exists, so the answer is NOT_FOUND -- its mapping wants re-resolving.
+func TestGetWorkloadStatsTransition(t *testing.T) {
+	s := newStatsService(t, healthyCgroup)
+	s.activeActor.Store(&testActor)
+	s.readSandboxCgroup = func(dir string) (cgroupstats.Sample, error) {
+		s.activeActor.Store(nil)
+		return cgroupstats.Read(dir)
+	}
+
+	_, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"})
+	if got := status.Code(err); got != codes.NotFound {
+		t.Errorf("GetWorkloadStats() during transition: code = %v, want %v (err: %v)", got, codes.NotFound, err)
+	}
+}
diff --git a/cmd/ateom-microvm/stats.go b/cmd/ateom-microvm/stats.go
index 426c2f4e5..c9c472746 100644
--- a/cmd/ateom-microvm/stats.go
+++ b/cmd/ateom-microvm/stats.go
@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"google.golang.org/grpc/codes"
@@ -26,6 +27,7 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/agentstats"
 	"github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/third_party/kata/agentpb"
+	"github.com/agent-substrate/substrate/internal/ateomstats"
 	"github.com/agent-substrate/substrate/internal/proto/ateompb"
 )
 
@@ -102,37 +104,19 @@ func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWor
 		return nil, status.Errorf(codes.NotFound, "ateom is executing actor %q, not the requested %q", active.UID, req.GetActorUid())
 	}
 
-	// The requested actor is the one here, but there is no guest to ask yet.
-	// Usually that is a poll landing in the boot or the restore: the ateom
-	// retains the attribution from the moment it accepts the actor, and the
-	// target is only published once the containers are up. It is also what a
-	// teardown looks like from here, since teardownActor clears the target
-	// before it closes the connection, and what a restore whose post-restore
-	// agent dial failed looks like for the rest of that activation.
-	//
-	// FAILED_PRECONDITION in all three: the answer is "no numbers right now",
-	// the caller should take the next sample, and for the teardown the next
-	// sample is the NOT_FOUND above.
-	target := s.guestStats.Load()
-	if target == nil {
-		return nil, status.Error(codes.FailedPrecondition, "no guest agent connection to measure yet")
-	}
-	// Belt and braces against the one thing that must never happen. The target
-	// is published and cleared under lock alongside the attribution, so this
-	// should be unreachable; if the two ever disagree, decline rather than
-	// report a stale guest's numbers under the requested actor's name.
-	if target.actorUID != active.UID {
-		return nil, status.Errorf(codes.FailedPrecondition, "guest agent connection belongs to actor %q, not %q", target.actorUID, active.UID)
-	}
-
-	observedAt := time.Now()
-	sample, err := sumContainerStats(ctx, target)
+	sample, err := s.sampleGuest(ctx, active)
 	if err != nil {
-		// Not Internal: a guest that has stopped answering is a routine state
-		// here, not a bug. Either the sandbox is going away — which the next
-		// CheckpointWorkload turns into the NOT_FOUND above — or the agent is
-		// briefly unreachable, and the next poll gets a number.
-		return nil, status.Errorf(codes.FailedPrecondition, "no container stats from the guest agent: %v", err)
+		if errors.Is(err, errStaleGuestTarget) {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+		// "No numbers right now", never NOT_FOUND: the requested actor IS the
+		// one here. The reasons are all routine -- a poll landing in the boot
+		// or the restore before the target is published, a teardown that
+		// cleared the target ahead of closing the connection, a restore whose
+		// post-restore agent dial failed, or a guest that has stopped
+		// answering. The caller should take the next sample; after a teardown
+		// the next sample is the NOT_FOUND above.
+		return nil, status.Error(codes.FailedPrecondition, err.Error())
 	}
 
 	// Re-check that the same workload is still the active one. The calls above
@@ -150,7 +134,100 @@ func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWor
 		return nil, status.Errorf(codes.NotFound, "ateom stopped executing actor %q while the sample was being taken", req.GetActorUid())
 	}
 
-	return &ateompb.GetWorkloadStatsResponse{
+	return &ateompb.GetWorkloadStatsResponse{Sample: sample}, nil
+}
+
+// GetActiveWorkloadStats implements
+// ateompb.Ateom/GetActiveWorkloadStats: the discovery read, sampling
+// whatever is executing with no identity asserted. Same lock discipline as
+// GetWorkloadStats above, for the same reasons.
+func (s *AteomService) GetActiveWorkloadStats(ctx context.Context, req *ateompb.GetActiveWorkloadStatsRequest) (*ateompb.GetActiveWorkloadStatsResponse, error) {
+	active := s.activeActor.Load()
+	if active == nil {
+		return noSample(ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD), nil
+	}
+
+	sample, err := s.sampleGuest(ctx, active)
+	if err != nil {
+		if errors.Is(err, errStaleGuestTarget) {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+		// Every routine way sampleGuest declines is a workload with no numbers
+		// yet -- boot, restore, teardown in progress, a guest that has stopped
+		// answering -- and for a caller with no prior knowledge each is as
+		// normal a finding as an available ateom. A reason, not an error.
+		return noSample(ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET), nil
+	}
+
+	// Same re-check as GetWorkloadStats, different answer: with no uid asserted
+	// there is no "requested actor" for NOT_FOUND to disown, and a transition
+	// underneath the read just means these numbers cannot be attributed to any
+	// single actor. Report the reason as of now -- the next tick resolves it
+	// either way.
+	if latest := s.activeActor.Load(); latest != active {
+		reason := ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET
+		if latest == nil {
+			reason = ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD
+		}
+		return noSample(reason), nil
+	}
+
+	return &ateompb.GetActiveWorkloadStatsResponse{
+		Result: &ateompb.GetActiveWorkloadStatsResponse_Sample{Sample: sample},
+	}, nil
+}
+
+// noSample is the discovery read's "nothing to give, and that is normal"
+// answer.
+func noSample(reason ateompb.NoSampleReason) *ateompb.GetActiveWorkloadStatsResponse {
+	return &ateompb.GetActiveWorkloadStatsResponse{
+		Result: &ateompb.GetActiveWorkloadStatsResponse_NoSampleReason{NoSampleReason: reason},
+	}
+}
+
+// errStaleGuestTarget is the one bug-shaped failure sampleGuest can return:
+// the published guest target and the attribution disagree, which the lifecycle
+// RPCs write together under lock and so should never happen. Both stats reads
+// map it to Internal; everything else sampleGuest returns is routine.
+var errStaleGuestTarget = errors.New("guest agent connection belongs to a different actor")
+
+// sampleGuest reads the guest's container cgroups through the agent and builds
+// the sample attributed to active. With one exception its errors mean "no
+// numbers right now" rather than a bug -- a guest that has stopped answering
+// is routine here, and unlike the gVisor runtime's local file reads, a vsock
+// call offers no error type that separates "gone" from "broken". The
+// exception is errStaleGuestTarget, above. Errors come back raw because the
+// two RPCs express the routine ones differently: an error code for the keyed
+// read, a NoSampleReason for the discovery read. Callers re-check
+// s.activeActor against the pointer they loaded after this returns; the read
+// holds no lock.
+func (s *AteomService) sampleGuest(ctx context.Context, active *ateomstats.ActorAttribution) (*ateompb.WorkloadStatsSample, error) {
+	// The actor is the one here, but there is no guest to ask yet. Usually that
+	// is a poll landing in the boot or the restore: the ateom retains the
+	// attribution from the moment it accepts the actor, and the target is only
+	// published once the containers are up. It is also what a teardown looks
+	// like from here, since teardownActor clears the target before it closes
+	// the connection, and what a restore whose post-restore agent dial failed
+	// looks like for the rest of that activation.
+	target := s.guestStats.Load()
+	if target == nil {
+		return nil, errors.New("no guest agent connection to measure yet")
+	}
+	// Belt and braces against the one thing that must never happen. The target
+	// is published and cleared under lock alongside the attribution, so this
+	// should be unreachable; if the two ever disagree, decline rather than
+	// report a stale guest's numbers under the requested actor's name.
+	if target.actorUID != active.UID {
+		return nil, fmt.Errorf("%w: %q, not %q", errStaleGuestTarget, target.actorUID, active.UID)
+	}
+
+	observedAt := time.Now()
+	sample, err := sumContainerStats(ctx, target)
+	if err != nil {
+		return nil, fmt.Errorf("no container stats from the guest agent: %w", err)
+	}
+
+	return &ateompb.WorkloadStatsSample{
 		Atespace:               active.Ref.Atespace,
 		ActorName:              active.Ref.Name,
 		ActorUid:               active.UID,
diff --git a/cmd/ateom-microvm/stats_test.go b/cmd/ateom-microvm/stats_test.go
index 7345f1380..0fcbe70b5 100644
--- a/cmd/ateom-microvm/stats_test.go
+++ b/cmd/ateom-microvm/stats_test.go
@@ -123,6 +123,12 @@ type fakeAgent struct {
 	stats map[string]*agentpb.CgroupStats
 	errs  map[string]error
 
+	// onCall, when set, runs at the top of every StatsContainer. It is how a
+	// test interleaves a lifecycle transition with the handlers' lock-free
+	// read: the handler has loaded activeActor by the time the agent is asked,
+	// so flipping it here lands in the window the re-check guards.
+	onCall func()
+
 	// calls records the container ids asked for, in order, so a test can tell
 	// "summed two containers" from "read one twice".
 	calls []string
@@ -132,6 +138,9 @@ type fakeAgent struct {
 }
 
 func (f *fakeAgent) StatsContainer(ctx context.Context, containerID string) (*agentpb.CgroupStats, error) {
+	if f.onCall != nil {
+		f.onCall()
+	}
 	f.calls = append(f.calls, containerID)
 	_, ok := ctx.Deadline()
 	f.deadlines = append(f.deadlines, ok)
@@ -175,13 +184,13 @@ func TestGetWorkloadStats(t *testing.T) {
 		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
 	}
 
-	if got.GetObservedAtUnixNano() < before || got.GetObservedAtUnixNano() > after {
-		t.Errorf("GetWorkloadStats() observed_at_unix_nano = %d, want within [%d, %d]", got.GetObservedAtUnixNano(), before, after)
+	if got.GetSample().GetObservedAtUnixNano() < before || got.GetSample().GetObservedAtUnixNano() > after {
+		t.Errorf("GetWorkloadStats() observed_at_unix_nano = %d, want within [%d, %d]", got.GetSample().GetObservedAtUnixNano(), before, after)
 	}
 	// Checked above; zeroed so the rest can be compared as a whole.
-	got.ObservedAtUnixNano = 0
+	got.GetSample().ObservedAtUnixNano = 0
 
-	want := &ateompb.GetWorkloadStatsResponse{
+	want := &ateompb.GetWorkloadStatsResponse{Sample: &ateompb.WorkloadStatsSample{
 		Atespace:               "space-a",
 		ActorName:              "actor-a",
 		ActorUid:               "uid-a",
@@ -193,7 +202,7 @@ func TestGetWorkloadStats(t *testing.T) {
 		MemoryPeakBytes:        209715200,
 		MemoryWorkingSetBytes:  136314880,
 		CpuUsageUsec:           1234567,
-	}
+	}}
 	if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" {
 		t.Errorf("GetWorkloadStats() mismatch (-want +got):\n%s", diff)
 	}
@@ -221,19 +230,19 @@ func TestGetWorkloadStatsSumsContainers(t *testing.T) {
 		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
 	}
 
-	if want := uint64(1500); got.GetMemoryCurrentBytes() != want {
-		t.Errorf("memory_current_bytes = %d, want %d", got.GetMemoryCurrentBytes(), want)
+	if want := uint64(1500); got.GetSample().GetMemoryCurrentBytes() != want {
+		t.Errorf("memory_current_bytes = %d, want %d", got.GetSample().GetMemoryCurrentBytes(), want)
 	}
 	// The sum of the peaks, which is an upper bound on the peak of the sum: the
 	// two containers need not have peaked at the same moment.
-	if want := uint64(4800); got.GetMemoryPeakBytes() != want {
-		t.Errorf("memory_peak_bytes = %d, want %d", got.GetMemoryPeakBytes(), want)
+	if want := uint64(4800); got.GetSample().GetMemoryPeakBytes() != want {
+		t.Errorf("memory_peak_bytes = %d, want %d", got.GetSample().GetMemoryPeakBytes(), want)
 	}
-	if want := uint64(1200); got.GetMemoryWorkingSetBytes() != want {
-		t.Errorf("memory_working_set_bytes = %d, want %d", got.GetMemoryWorkingSetBytes(), want)
+	if want := uint64(1200); got.GetSample().GetMemoryWorkingSetBytes() != want {
+		t.Errorf("memory_working_set_bytes = %d, want %d", got.GetSample().GetMemoryWorkingSetBytes(), want)
 	}
-	if want := uint64(10); got.GetCpuUsageUsec() != want {
-		t.Errorf("cpu_usage_usec = %d, want %d", got.GetCpuUsageUsec(), want)
+	if want := uint64(10); got.GetSample().GetCpuUsageUsec() != want {
+		t.Errorf("cpu_usage_usec = %d, want %d", got.GetSample().GetCpuUsageUsec(), want)
 	}
 
 	if want := []string{"app_ovl", "sidecar_ovl"}; !cmp.Equal(want, agent.calls) {
@@ -257,11 +266,11 @@ func TestGetWorkloadStatsSkipsUnreadableContainer(t *testing.T) {
 	if err != nil {
 		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
 	}
-	if want := uint64(1000); got.GetMemoryCurrentBytes() != want {
-		t.Errorf("memory_current_bytes = %d, want %d", got.GetMemoryCurrentBytes(), want)
+	if want := uint64(1000); got.GetSample().GetMemoryCurrentBytes() != want {
+		t.Errorf("memory_current_bytes = %d, want %d", got.GetSample().GetMemoryCurrentBytes(), want)
 	}
-	if want := uint64(5); got.GetCpuUsageUsec() != want {
-		t.Errorf("cpu_usage_usec = %d, want %d", got.GetCpuUsageUsec(), want)
+	if want := uint64(5); got.GetSample().GetCpuUsageUsec() != want {
+		t.Errorf("cpu_usage_usec = %d, want %d", got.GetSample().GetCpuUsageUsec(), want)
 	}
 }
 
@@ -276,7 +285,7 @@ func TestGetWorkloadStatsCountsAnsweredContainer(t *testing.T) {
 	if err != nil {
 		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
 	}
-	if got.GetMemoryCurrentBytes() != 0 || got.GetCpuUsageUsec() != 0 {
+	if got.GetSample().GetMemoryCurrentBytes() != 0 || got.GetSample().GetCpuUsageUsec() != 0 {
 		t.Errorf("GetWorkloadStats() = %v, want an all-zero measurement", got)
 	}
 }
@@ -332,8 +341,8 @@ func TestGetWorkloadStatsErrors(t *testing.T) {
 		},
 		{
 			// Should be unreachable — the two atomics are written together under
-			// lock — so what is pinned here is that disagreeing state declines
-			// instead of misattributing.
+			// lock — so a disagreement is an invariant violation, not a routine
+			// state: Internal, unlike every other way sampleGuest declines.
 			name: "guest agent connection belongs to another actor",
 			service: func() *AteomService {
 				s := newStatsService(healthy, "app_ovl")
@@ -341,7 +350,7 @@ func TestGetWorkloadStatsErrors(t *testing.T) {
 				return s
 			},
 			actorUID: "uid-a",
-			want:     codes.FailedPrecondition,
+			want:     codes.Internal,
 		},
 		{
 			// Not one container gone but the guest as a whole not answering: the
@@ -414,3 +423,138 @@ func TestAteomServiceStartsAvailable(t *testing.T) {
 		t.Errorf("new AteomService.guestStats = %v, want nil", got)
 	}
 }
+
+func TestGetActiveWorkloadStats(t *testing.T) {
+	agent := &fakeAgent{stats: map[string]*agentpb.CgroupStats{
+		"app_ovl": containerStats(157286400, 209715200, 20971520, 1234567000),
+	}}
+	s := newStatsService(agent, "app_ovl")
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() error = %v, want nil", err)
+	}
+	if got.GetSample() == nil {
+		t.Fatalf("GetActiveWorkloadStats() = %v, want a sample", got)
+	}
+
+	// The keyed read against the same fake is the reference: the discovery read
+	// must produce the identical sample, since both are the same measurement
+	// with a different addressing mode.
+	want, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"})
+	if err != nil {
+		t.Fatalf("GetWorkloadStats() error = %v, want nil", err)
+	}
+	sample := got.GetSample()
+	sample.ObservedAtUnixNano = 0
+	want.GetSample().ObservedAtUnixNano = 0
+	if diff := cmp.Diff(want.GetSample(), sample, protocmp.Transform()); diff != "" {
+		t.Errorf("discovery sample differs from keyed sample (-keyed +discovery):\n%s", diff)
+	}
+}
+
+// TestGetActiveWorkloadStatsAvailable pins the contract that makes the
+// discovery read scrapeable: an idle ateom is a reason, never an error.
+func TestGetActiveWorkloadStatsAvailable(t *testing.T) {
+	s := &AteomService{}
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() on an available ateom: error = %v, want nil", err)
+	}
+	if got.GetNoSampleReason() != ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD {
+		t.Errorf("GetActiveWorkloadStats() = %v, want NO_WORKLOAD reason", got)
+	}
+}
+
+// TestGetActiveWorkloadStatsBooting: executing but no guest target yet is a
+// NOT_MEASURABLE_YET reason, not an error, unlike the keyed read's
+// FAILED_PRECONDITION. A blind caller finds boots as routinely as idle
+// workers.
+func TestGetActiveWorkloadStatsBooting(t *testing.T) {
+	s := &AteomService{}
+	s.activeActor.Store(&testActor) // attribution retained, target not published
+
+	got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if err != nil {
+		t.Fatalf("GetActiveWorkloadStats() mid-boot: error = %v, want nil", err)
+	}
+	if got.GetNoSampleReason() != ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET {
+		t.Errorf("GetActiveWorkloadStats() mid-boot = %v, want NOT_MEASURABLE_YET reason", got)
+	}
+}
+
+// The transition tests cover the re-check that runs after the lock-free
+// measurement: the fake agent's onCall hook flips activeActor while the
+// handler is mid-read, which is exactly the window a checkpoint plus a fresh
+// run (or a checkpoint alone) can land in.
+
+func TestGetActiveWorkloadStatsTransition(t *testing.T) {
+	otherActor := testActor
+	otherActor.UID = "uid-b"
+
+	tests := []struct {
+		name string
+		to   *ateomstats.ActorAttribution
+		want ateompb.NoSampleReason
+	}{
+		// A new actor took the slot: there is a workload, its numbers are just
+		// not attributable this tick.
+		{name: "to another actor", to: &otherActor, want: ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET},
+		// A checkpoint emptied the slot: report what is true now.
+		{name: "to available", to: nil, want: ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			agent := &fakeAgent{stats: map[string]*agentpb.CgroupStats{
+				"app_ovl": containerStats(1000, 2000, 100, 5000),
+			}}
+			s := newStatsService(agent, "app_ovl")
+			agent.onCall = func() { s.activeActor.Store(tc.to) }
+
+			got, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+			if err != nil {
+				t.Fatalf("GetActiveWorkloadStats() during transition: error = %v, want nil", err)
+			}
+			if got.GetSample() != nil {
+				t.Errorf("GetActiveWorkloadStats() during transition returned sample %v, want none", got.GetSample())
+			}
+			if got.GetNoSampleReason() != tc.want {
+				t.Errorf("GetActiveWorkloadStats() during transition = %v, want %v reason", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestGetWorkloadStatsTransition pins the keyed read's side of the same
+// window: the caller asserted an actor that is gone by the time the sample
+// exists, so the answer is NOT_FOUND -- its mapping wants re-resolving.
+func TestGetWorkloadStatsTransition(t *testing.T) {
+	agent := &fakeAgent{stats: map[string]*agentpb.CgroupStats{
+		"app_ovl": containerStats(1000, 2000, 100, 5000),
+	}}
+	s := newStatsService(agent, "app_ovl")
+	agent.onCall = func() { s.activeActor.Store(nil) }
+
+	_, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"})
+	if got := status.Code(err); got != codes.NotFound {
+		t.Errorf("GetWorkloadStats() during transition: code = %v, want %v (err: %v)", got, codes.NotFound, err)
+	}
+}
+
+// TestGetActiveWorkloadStatsStaleTarget pins that the one bug-shaped failure
+// stays an error on the discovery read too: a target/attribution disagreement
+// is an invariant violation, not a NOT_MEASURABLE_YET to skip past silently.
+func TestGetActiveWorkloadStatsStaleTarget(t *testing.T) {
+	agent := &fakeAgent{stats: map[string]*agentpb.CgroupStats{
+		"app_ovl": containerStats(1000, 2000, 100, 5000),
+	}}
+	s := newStatsService(agent, "app_ovl")
+	s.guestStats.Store(&guestStatsTarget{actorUID: "uid-b", agent: agent, workloadIDs: []string{"app_ovl"}})
+
+	_, err := s.GetActiveWorkloadStats(context.Background(), &ateompb.GetActiveWorkloadStatsRequest{})
+	if got := status.Code(err); got != codes.Internal {
+		t.Errorf("GetActiveWorkloadStats() with stale target: code = %v, want %v (err: %v)", got, codes.Internal, err)
+	}
+}
diff --git a/internal/proto/ateompb/ateom.pb.go b/internal/proto/ateompb/ateom.pb.go
index 3ff384612..a0c83adb8 100644
--- a/internal/proto/ateompb/ateom.pb.go
+++ b/internal/proto/ateompb/ateom.pb.go
@@ -212,6 +212,62 @@ func (StatsSource) EnumDescriptor() ([]byte, []int) {
 	return file_ateom_proto_rawDescGZIP(), []int{2}
 }
 
+// NoSampleReason is why the discovery read has no sample to give -- every
+// value is a normal state a caller with no prior knowledge routinely finds,
+// never an error.
+type NoSampleReason int32
+
+const (
+	NoSampleReason_NO_SAMPLE_REASON_UNSPECIFIED NoSampleReason = 0
+	// Nothing is executing: the ateom is "available".
+	NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD NoSampleReason = 1
+	// A workload is executing but there are no numbers to give yet: a poll
+	// landing in a boot or a restore, a teardown in progress, or a lifecycle
+	// transition underneath the read. Transient -- take the next sample.
+	NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET NoSampleReason = 2
+)
+
+// Enum value maps for NoSampleReason.
+var (
+	NoSampleReason_name = map[int32]string{
+		0: "NO_SAMPLE_REASON_UNSPECIFIED",
+		1: "NO_SAMPLE_REASON_NO_WORKLOAD",
+		2: "NO_SAMPLE_REASON_NOT_MEASURABLE_YET",
+	}
+	NoSampleReason_value = map[string]int32{
+		"NO_SAMPLE_REASON_UNSPECIFIED":        0,
+		"NO_SAMPLE_REASON_NO_WORKLOAD":        1,
+		"NO_SAMPLE_REASON_NOT_MEASURABLE_YET": 2,
+	}
+)
+
+func (x NoSampleReason) Enum() *NoSampleReason {
+	p := new(NoSampleReason)
+	*p = x
+	return p
+}
+
+func (x NoSampleReason) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (NoSampleReason) Descriptor() protoreflect.EnumDescriptor {
+	return file_ateom_proto_enumTypes[3].Descriptor()
+}
+
+func (NoSampleReason) Type() protoreflect.EnumType {
+	return &file_ateom_proto_enumTypes[3]
+}
+
+func (x NoSampleReason) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use NoSampleReason.Descriptor instead.
+func (NoSampleReason) EnumDescriptor() ([]byte, []int) {
+	return file_ateom_proto_rawDescGZIP(), []int{3}
+}
+
 type RunWorkloadRequest struct {
 	state                  protoimpl.MessageState `protogen:"open.v1"`
 	Atespace               string                 `protobuf:"bytes,1,opt,name=atespace,proto3" json:"atespace,omitempty"`
@@ -1083,14 +1139,14 @@ func (x *GetWorkloadStatsRequest) GetActorUid() string {
 	return ""
 }
 
-// GetWorkloadStatsResponse is one resource-usage sample for the executing
-// workload. The unit of measurement is the SANDBOX, which today equals the
-// actor. Per-container attribution is not reported: the micro-VM source could
-// give it, since the guest keeps a cgroup per container and ateom sums them,
-// but the gVisor source cannot split one at all without the sentry's own
-// accounting, and a field only one runtime could ever fill would be worse than
-// none.
-type GetWorkloadStatsResponse struct {
+// WorkloadStatsSample is one resource-usage sample for an executing workload,
+// shared by both stats reads. The unit of measurement is the SANDBOX, which
+// today equals the actor. Per-container attribution is not reported: the
+// micro-VM source could give it, since the guest keeps a cgroup per container
+// and ateom sums them, but the gVisor source cannot split one at all without
+// the sentry's own accounting, and a field only one runtime could ever fill
+// would be worse than none.
+type WorkloadStatsSample struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Identity of the measured actor, retained by ateom from the
 	// RunWorkloadRequest / RestoreWorkloadRequest that started it. Echoed back so
@@ -1135,20 +1191,20 @@ type GetWorkloadStatsResponse struct {
 	sizeCache          protoimpl.SizeCache
 }
 
-func (x *GetWorkloadStatsResponse) Reset() {
-	*x = GetWorkloadStatsResponse{}
+func (x *WorkloadStatsSample) Reset() {
+	*x = WorkloadStatsSample{}
 	mi := &file_ateom_proto_msgTypes[13]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *GetWorkloadStatsResponse) String() string {
+func (x *WorkloadStatsSample) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*GetWorkloadStatsResponse) ProtoMessage() {}
+func (*WorkloadStatsSample) ProtoMessage() {}
 
-func (x *GetWorkloadStatsResponse) ProtoReflect() protoreflect.Message {
+func (x *WorkloadStatsSample) ProtoReflect() protoreflect.Message {
 	mi := &file_ateom_proto_msgTypes[13]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -1160,95 +1216,263 @@ func (x *GetWorkloadStatsResponse) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use GetWorkloadStatsResponse.ProtoReflect.Descriptor instead.
-func (*GetWorkloadStatsResponse) Descriptor() ([]byte, []int) {
+// Deprecated: Use WorkloadStatsSample.ProtoReflect.Descriptor instead.
+func (*WorkloadStatsSample) Descriptor() ([]byte, []int) {
 	return file_ateom_proto_rawDescGZIP(), []int{13}
 }
 
-func (x *GetWorkloadStatsResponse) GetAtespace() string {
+func (x *WorkloadStatsSample) GetAtespace() string {
 	if x != nil {
 		return x.Atespace
 	}
 	return ""
 }
 
-func (x *GetWorkloadStatsResponse) GetActorName() string {
+func (x *WorkloadStatsSample) GetActorName() string {
 	if x != nil {
 		return x.ActorName
 	}
 	return ""
 }
 
-func (x *GetWorkloadStatsResponse) GetActorUid() string {
+func (x *WorkloadStatsSample) GetActorUid() string {
 	if x != nil {
 		return x.ActorUid
 	}
 	return ""
 }
 
-func (x *GetWorkloadStatsResponse) GetActorTemplateNamespace() string {
+func (x *WorkloadStatsSample) GetActorTemplateNamespace() string {
 	if x != nil {
 		return x.ActorTemplateNamespace
 	}
 	return ""
 }
 
-func (x *GetWorkloadStatsResponse) GetActorTemplateName() string {
+func (x *WorkloadStatsSample) GetActorTemplateName() string {
 	if x != nil {
 		return x.ActorTemplateName
 	}
 	return ""
 }
 
-func (x *GetWorkloadStatsResponse) GetSandboxClass() SandboxClass {
+func (x *WorkloadStatsSample) GetSandboxClass() SandboxClass {
 	if x != nil {
 		return x.SandboxClass
 	}
 	return SandboxClass_SANDBOX_CLASS_UNSPECIFIED
 }
 
-func (x *GetWorkloadStatsResponse) GetSource() StatsSource {
+func (x *WorkloadStatsSample) GetSource() StatsSource {
 	if x != nil {
 		return x.Source
 	}
 	return StatsSource_STATS_SOURCE_UNSPECIFIED
 }
 
-func (x *GetWorkloadStatsResponse) GetMemoryCurrentBytes() uint64 {
+func (x *WorkloadStatsSample) GetMemoryCurrentBytes() uint64 {
 	if x != nil {
 		return x.MemoryCurrentBytes
 	}
 	return 0
 }
 
-func (x *GetWorkloadStatsResponse) GetMemoryPeakBytes() uint64 {
+func (x *WorkloadStatsSample) GetMemoryPeakBytes() uint64 {
 	if x != nil {
 		return x.MemoryPeakBytes
 	}
 	return 0
 }
 
-func (x *GetWorkloadStatsResponse) GetMemoryWorkingSetBytes() uint64 {
+func (x *WorkloadStatsSample) GetMemoryWorkingSetBytes() uint64 {
 	if x != nil {
 		return x.MemoryWorkingSetBytes
 	}
 	return 0
 }
 
-func (x *GetWorkloadStatsResponse) GetCpuUsageUsec() uint64 {
+func (x *WorkloadStatsSample) GetCpuUsageUsec() uint64 {
 	if x != nil {
 		return x.CpuUsageUsec
 	}
 	return 0
 }
 
-func (x *GetWorkloadStatsResponse) GetObservedAtUnixNano() int64 {
+func (x *WorkloadStatsSample) GetObservedAtUnixNano() int64 {
 	if x != nil {
 		return x.ObservedAtUnixNano
 	}
 	return 0
 }
 
+type GetWorkloadStatsResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Sample        *WorkloadStatsSample   `protobuf:"bytes,1,opt,name=sample,proto3" json:"sample,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *GetWorkloadStatsResponse) Reset() {
+	*x = GetWorkloadStatsResponse{}
+	mi := &file_ateom_proto_msgTypes[14]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *GetWorkloadStatsResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetWorkloadStatsResponse) ProtoMessage() {}
+
+func (x *GetWorkloadStatsResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_ateom_proto_msgTypes[14]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetWorkloadStatsResponse.ProtoReflect.Descriptor instead.
+func (*GetWorkloadStatsResponse) Descriptor() ([]byte, []int) {
+	return file_ateom_proto_rawDescGZIP(), []int{14}
+}
+
+func (x *GetWorkloadStatsResponse) GetSample() *WorkloadStatsSample {
+	if x != nil {
+		return x.Sample
+	}
+	return nil
+}
+
+type GetActiveWorkloadStatsRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *GetActiveWorkloadStatsRequest) Reset() {
+	*x = GetActiveWorkloadStatsRequest{}
+	mi := &file_ateom_proto_msgTypes[15]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *GetActiveWorkloadStatsRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetActiveWorkloadStatsRequest) ProtoMessage() {}
+
+func (x *GetActiveWorkloadStatsRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_ateom_proto_msgTypes[15]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetActiveWorkloadStatsRequest.ProtoReflect.Descriptor instead.
+func (*GetActiveWorkloadStatsRequest) Descriptor() ([]byte, []int) {
+	return file_ateom_proto_rawDescGZIP(), []int{15}
+}
+
+type GetActiveWorkloadStatsResponse struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Exactly one of the two is set: either a measurement, or the reason there
+	// is none. An ateom serves one actor at a time, so the sample slot is
+	// singular; the sample is self-describing (see the attribution rule on the
+	// rpc), and a sample being present is itself the statement that a workload
+	// is executing.
+	//
+	// Types that are valid to be assigned to Result:
+	//
+	//	*GetActiveWorkloadStatsResponse_Sample
+	//	*GetActiveWorkloadStatsResponse_NoSampleReason
+	Result        isGetActiveWorkloadStatsResponse_Result `protobuf_oneof:"result"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *GetActiveWorkloadStatsResponse) Reset() {
+	*x = GetActiveWorkloadStatsResponse{}
+	mi := &file_ateom_proto_msgTypes[16]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *GetActiveWorkloadStatsResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetActiveWorkloadStatsResponse) ProtoMessage() {}
+
+func (x *GetActiveWorkloadStatsResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_ateom_proto_msgTypes[16]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use GetActiveWorkloadStatsResponse.ProtoReflect.Descriptor instead.
+func (*GetActiveWorkloadStatsResponse) Descriptor() ([]byte, []int) {
+	return file_ateom_proto_rawDescGZIP(), []int{16}
+}
+
+func (x *GetActiveWorkloadStatsResponse) GetResult() isGetActiveWorkloadStatsResponse_Result {
+	if x != nil {
+		return x.Result
+	}
+	return nil
+}
+
+func (x *GetActiveWorkloadStatsResponse) GetSample() *WorkloadStatsSample {
+	if x != nil {
+		if x, ok := x.Result.(*GetActiveWorkloadStatsResponse_Sample); ok {
+			return x.Sample
+		}
+	}
+	return nil
+}
+
+func (x *GetActiveWorkloadStatsResponse) GetNoSampleReason() NoSampleReason {
+	if x != nil {
+		if x, ok := x.Result.(*GetActiveWorkloadStatsResponse_NoSampleReason); ok {
+			return x.NoSampleReason
+		}
+	}
+	return NoSampleReason_NO_SAMPLE_REASON_UNSPECIFIED
+}
+
+type isGetActiveWorkloadStatsResponse_Result interface {
+	isGetActiveWorkloadStatsResponse_Result()
+}
+
+type GetActiveWorkloadStatsResponse_Sample struct {
+	Sample *WorkloadStatsSample `protobuf:"bytes,1,opt,name=sample,proto3,oneof"`
+}
+
+type GetActiveWorkloadStatsResponse_NoSampleReason struct {
+	NoSampleReason NoSampleReason `protobuf:"varint,2,opt,name=no_sample_reason,json=noSampleReason,proto3,enum=ateom.NoSampleReason,oneof"`
+}
+
+func (*GetActiveWorkloadStatsResponse_Sample) isGetActiveWorkloadStatsResponse_Result() {}
+
+func (*GetActiveWorkloadStatsResponse_NoSampleReason) isGetActiveWorkloadStatsResponse_Result() {}
+
 var File_ateom_proto protoreflect.FileDescriptor
 
 const file_ateom_proto_rawDesc = "" +
@@ -1334,8 +1558,8 @@ const file_ateom_proto_rawDesc = "" +
 	"\x0f_egress_gateway\"\x19\n" +
 	"\x17RestoreWorkloadResponse\"6\n" +
 	"\x17GetWorkloadStatsRequest\x12\x1b\n" +
-	"\tactor_uid\x18\x01 \x01(\tR\bactorUid\"\xb2\x04\n" +
-	"\x18GetWorkloadStatsResponse\x12\x1a\n" +
+	"\tactor_uid\x18\x01 \x01(\tR\bactorUid\"\xad\x04\n" +
+	"\x13WorkloadStatsSample\x12\x1a\n" +
 	"\batespace\x18\x01 \x01(\tR\batespace\x12\x1d\n" +
 	"\n" +
 	"actor_name\x18\x02 \x01(\tR\tactorName\x12\x1b\n" +
@@ -1349,7 +1573,14 @@ const file_ateom_proto_rawDesc = "" +
 	"\x18memory_working_set_bytes\x18\n" +
 	" \x01(\x04R\x15memoryWorkingSetBytes\x12$\n" +
 	"\x0ecpu_usage_usec\x18\v \x01(\x04R\fcpuUsageUsec\x121\n" +
-	"\x15observed_at_unix_nano\x18\f \x01(\x03R\x12observedAtUnixNano*\x84\x01\n" +
+	"\x15observed_at_unix_nano\x18\f \x01(\x03R\x12observedAtUnixNano\"N\n" +
+	"\x18GetWorkloadStatsResponse\x122\n" +
+	"\x06sample\x18\x01 \x01(\v2\x1a.ateom.WorkloadStatsSampleR\x06sample\"\x1f\n" +
+	"\x1dGetActiveWorkloadStatsRequest\"\xa3\x01\n" +
+	"\x1eGetActiveWorkloadStatsResponse\x124\n" +
+	"\x06sample\x18\x01 \x01(\v2\x1a.ateom.WorkloadStatsSampleH\x00R\x06sample\x12A\n" +
+	"\x10no_sample_reason\x18\x02 \x01(\x0e2\x15.ateom.NoSampleReasonH\x00R\x0enoSampleReasonB\b\n" +
+	"\x06result*\x84\x01\n" +
 	"\rSnapshotScope\x12\x1e\n" +
 	"\x1aSNAPSHOT_SCOPE_UNSPECIFIED\x10\x00\x12\x17\n" +
 	"\x13SNAPSHOT_SCOPE_FULL\x10\x01\x12\x17\n" +
@@ -1362,12 +1593,17 @@ const file_ateom_proto_rawDesc = "" +
 	"\vStatsSource\x12\x1c\n" +
 	"\x18STATS_SOURCE_UNSPECIFIED\x10\x00\x12\x17\n" +
 	"\x13STATS_SOURCE_CGROUP\x10\x01\x12\x1c\n" +
-	"\x18STATS_SOURCE_GUEST_AGENT\x10\x022\xd7\x02\n" +
+	"\x18STATS_SOURCE_GUEST_AGENT\x10\x02*}\n" +
+	"\x0eNoSampleReason\x12 \n" +
+	"\x1cNO_SAMPLE_REASON_UNSPECIFIED\x10\x00\x12 \n" +
+	"\x1cNO_SAMPLE_REASON_NO_WORKLOAD\x10\x01\x12'\n" +
+	"#NO_SAMPLE_REASON_NOT_MEASURABLE_YET\x10\x022\xc0\x03\n" +
 	"\x05Ateom\x12F\n" +
 	"\vRunWorkload\x12\x19.ateom.RunWorkloadRequest\x1a\x1a.ateom.RunWorkloadResponse\"\x00\x12[\n" +
 	"\x12CheckpointWorkload\x12 .ateom.CheckpointWorkloadRequest\x1a!.ateom.CheckpointWorkloadResponse\"\x00\x12R\n" +
 	"\x0fRestoreWorkload\x12\x1d.ateom.RestoreWorkloadRequest\x1a\x1e.ateom.RestoreWorkloadResponse\"\x00\x12U\n" +
-	"\x10GetWorkloadStats\x12\x1e.ateom.GetWorkloadStatsRequest\x1a\x1f.ateom.GetWorkloadStatsResponse\"\x00B=Z;github.com/agent-substrate/substrate/internal/proto/ateompbb\x06proto3"
+	"\x10GetWorkloadStats\x12\x1e.ateom.GetWorkloadStatsRequest\x1a\x1f.ateom.GetWorkloadStatsResponse\"\x00\x12g\n" +
+	"\x16GetActiveWorkloadStats\x12$.ateom.GetActiveWorkloadStatsRequest\x1a%.ateom.GetActiveWorkloadStatsResponse\"\x00B=Z;github.com/agent-substrate/substrate/internal/proto/ateompbb\x06proto3"
 
 var (
 	file_ateom_proto_rawDescOnce sync.Once
@@ -1381,60 +1617,69 @@ func file_ateom_proto_rawDescGZIP() []byte {
 	return file_ateom_proto_rawDescData
 }
 
-var file_ateom_proto_enumTypes = make([]protoimpl.EnumInfo, 3)
-var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 17)
+var file_ateom_proto_enumTypes = make([]protoimpl.EnumInfo, 4)
+var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 20)
 var file_ateom_proto_goTypes = []any{
-	(SnapshotScope)(0),                 // 0: ateom.SnapshotScope
-	(SandboxClass)(0),                  // 1: ateom.SandboxClass
-	(StatsSource)(0),                   // 2: ateom.StatsSource
-	(*RunWorkloadRequest)(nil),         // 3: ateom.RunWorkloadRequest
-	(*EgressGateway)(nil),              // 4: ateom.EgressGateway
-	(*WorkloadSpec)(nil),               // 5: ateom.WorkloadSpec
-	(*Container)(nil),                  // 6: ateom.Container
-	(*DurableDirVolumeMount)(nil),      // 7: ateom.DurableDirVolumeMount
-	(*Readyz)(nil),                     // 8: ateom.Readyz
-	(*HTTPGetAction)(nil),              // 9: ateom.HTTPGetAction
-	(*RunWorkloadResponse)(nil),        // 10: ateom.RunWorkloadResponse
-	(*CheckpointWorkloadRequest)(nil),  // 11: ateom.CheckpointWorkloadRequest
-	(*CheckpointWorkloadResponse)(nil), // 12: ateom.CheckpointWorkloadResponse
-	(*RestoreWorkloadRequest)(nil),     // 13: ateom.RestoreWorkloadRequest
-	(*RestoreWorkloadResponse)(nil),    // 14: ateom.RestoreWorkloadResponse
-	(*GetWorkloadStatsRequest)(nil),    // 15: ateom.GetWorkloadStatsRequest
-	(*GetWorkloadStatsResponse)(nil),   // 16: ateom.GetWorkloadStatsResponse
-	nil,                                // 17: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry
-	nil,                                // 18: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry
-	nil,                                // 19: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry
+	(SnapshotScope)(0),                     // 0: ateom.SnapshotScope
+	(SandboxClass)(0),                      // 1: ateom.SandboxClass
+	(StatsSource)(0),                       // 2: ateom.StatsSource
+	(NoSampleReason)(0),                    // 3: ateom.NoSampleReason
+	(*RunWorkloadRequest)(nil),             // 4: ateom.RunWorkloadRequest
+	(*EgressGateway)(nil),                  // 5: ateom.EgressGateway
+	(*WorkloadSpec)(nil),                   // 6: ateom.WorkloadSpec
+	(*Container)(nil),                      // 7: ateom.Container
+	(*DurableDirVolumeMount)(nil),          // 8: ateom.DurableDirVolumeMount
+	(*Readyz)(nil),                         // 9: ateom.Readyz
+	(*HTTPGetAction)(nil),                  // 10: ateom.HTTPGetAction
+	(*RunWorkloadResponse)(nil),            // 11: ateom.RunWorkloadResponse
+	(*CheckpointWorkloadRequest)(nil),      // 12: ateom.CheckpointWorkloadRequest
+	(*CheckpointWorkloadResponse)(nil),     // 13: ateom.CheckpointWorkloadResponse
+	(*RestoreWorkloadRequest)(nil),         // 14: ateom.RestoreWorkloadRequest
+	(*RestoreWorkloadResponse)(nil),        // 15: ateom.RestoreWorkloadResponse
+	(*GetWorkloadStatsRequest)(nil),        // 16: ateom.GetWorkloadStatsRequest
+	(*WorkloadStatsSample)(nil),            // 17: ateom.WorkloadStatsSample
+	(*GetWorkloadStatsResponse)(nil),       // 18: ateom.GetWorkloadStatsResponse
+	(*GetActiveWorkloadStatsRequest)(nil),  // 19: ateom.GetActiveWorkloadStatsRequest
+	(*GetActiveWorkloadStatsResponse)(nil), // 20: ateom.GetActiveWorkloadStatsResponse
+	nil,                                    // 21: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry
+	nil,                                    // 22: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry
+	nil,                                    // 23: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry
 }
 var file_ateom_proto_depIdxs = []int32{
-	5,  // 0: ateom.RunWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
-	17, // 1: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry
-	4,  // 2: ateom.RunWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway
-	6,  // 3: ateom.WorkloadSpec.containers:type_name -> ateom.Container
-	8,  // 4: ateom.Container.readyz:type_name -> ateom.Readyz
-	7,  // 5: ateom.Container.durable_dir_volume_mounts:type_name -> ateom.DurableDirVolumeMount
-	9,  // 6: ateom.Readyz.http_get:type_name -> ateom.HTTPGetAction
-	5,  // 7: ateom.CheckpointWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
-	18, // 8: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry
+	6,  // 0: ateom.RunWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
+	21, // 1: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry
+	5,  // 2: ateom.RunWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway
+	7,  // 3: ateom.WorkloadSpec.containers:type_name -> ateom.Container
+	9,  // 4: ateom.Container.readyz:type_name -> ateom.Readyz
+	8,  // 5: ateom.Container.durable_dir_volume_mounts:type_name -> ateom.DurableDirVolumeMount
+	10, // 6: ateom.Readyz.http_get:type_name -> ateom.HTTPGetAction
+	6,  // 7: ateom.CheckpointWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
+	22, // 8: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry
 	0,  // 9: ateom.CheckpointWorkloadRequest.scope:type_name -> ateom.SnapshotScope
-	5,  // 10: ateom.RestoreWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
-	19, // 11: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry
+	6,  // 10: ateom.RestoreWorkloadRequest.spec:type_name -> ateom.WorkloadSpec
+	23, // 11: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry
 	0,  // 12: ateom.RestoreWorkloadRequest.scope:type_name -> ateom.SnapshotScope
-	4,  // 13: ateom.RestoreWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway
-	1,  // 14: ateom.GetWorkloadStatsResponse.sandbox_class:type_name -> ateom.SandboxClass
-	2,  // 15: ateom.GetWorkloadStatsResponse.source:type_name -> ateom.StatsSource
-	3,  // 16: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest
-	11, // 17: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest
-	13, // 18: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest
-	15, // 19: ateom.Ateom.GetWorkloadStats:input_type -> ateom.GetWorkloadStatsRequest
-	10, // 20: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse
-	12, // 21: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse
-	14, // 22: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse
-	16, // 23: ateom.Ateom.GetWorkloadStats:output_type -> ateom.GetWorkloadStatsResponse
-	20, // [20:24] is the sub-list for method output_type
-	16, // [16:20] is the sub-list for method input_type
-	16, // [16:16] is the sub-list for extension type_name
-	16, // [16:16] is the sub-list for extension extendee
-	0,  // [0:16] is the sub-list for field type_name
+	5,  // 13: ateom.RestoreWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway
+	1,  // 14: ateom.WorkloadStatsSample.sandbox_class:type_name -> ateom.SandboxClass
+	2,  // 15: ateom.WorkloadStatsSample.source:type_name -> ateom.StatsSource
+	17, // 16: ateom.GetWorkloadStatsResponse.sample:type_name -> ateom.WorkloadStatsSample
+	17, // 17: ateom.GetActiveWorkloadStatsResponse.sample:type_name -> ateom.WorkloadStatsSample
+	3,  // 18: ateom.GetActiveWorkloadStatsResponse.no_sample_reason:type_name -> ateom.NoSampleReason
+	4,  // 19: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest
+	12, // 20: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest
+	14, // 21: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest
+	16, // 22: ateom.Ateom.GetWorkloadStats:input_type -> ateom.GetWorkloadStatsRequest
+	19, // 23: ateom.Ateom.GetActiveWorkloadStats:input_type -> ateom.GetActiveWorkloadStatsRequest
+	11, // 24: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse
+	13, // 25: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse
+	15, // 26: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse
+	18, // 27: ateom.Ateom.GetWorkloadStats:output_type -> ateom.GetWorkloadStatsResponse
+	20, // 28: ateom.Ateom.GetActiveWorkloadStats:output_type -> ateom.GetActiveWorkloadStatsResponse
+	24, // [24:29] is the sub-list for method output_type
+	19, // [19:24] is the sub-list for method input_type
+	19, // [19:19] is the sub-list for extension type_name
+	19, // [19:19] is the sub-list for extension extendee
+	0,  // [0:19] is the sub-list for field type_name
 }
 
 func init() { file_ateom_proto_init() }
@@ -1444,13 +1689,17 @@ func file_ateom_proto_init() {
 	}
 	file_ateom_proto_msgTypes[0].OneofWrappers = []any{}
 	file_ateom_proto_msgTypes[10].OneofWrappers = []any{}
+	file_ateom_proto_msgTypes[16].OneofWrappers = []any{
+		(*GetActiveWorkloadStatsResponse_Sample)(nil),
+		(*GetActiveWorkloadStatsResponse_NoSampleReason)(nil),
+	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_ateom_proto_rawDesc), len(file_ateom_proto_rawDesc)),
-			NumEnums:      3,
-			NumMessages:   17,
+			NumEnums:      4,
+			NumMessages:   20,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/internal/proto/ateompb/ateom.proto b/internal/proto/ateompb/ateom.proto
index f190fcd15..877f8fe47 100644
--- a/internal/proto/ateompb/ateom.proto
+++ b/internal/proto/ateompb/ateom.proto
@@ -77,6 +77,27 @@ service Ateom {
   // "booting" distinguishable from "not here" at all, and it means a workload
   // that dies during boot is attributable rather than anonymous.
   rpc GetWorkloadStats(GetWorkloadStatsRequest) returns (GetWorkloadStatsResponse) {}
+
+  // GetActiveWorkloadStats samples whatever this ateom is currently
+  // executing, without asserting an identity. It is the discovery read for a
+  // scraper that enumerates ateoms and holds no worker-to-actor mapping;
+  // GetWorkloadStats above is the verified read for a caller that must be
+  // answered about a specific actor.
+  //
+  // Every state a blind caller can find is a normal answer here, never an
+  // error: the response is either a sample or the NoSampleReason there is
+  // none -- nothing to measure ("available"), or nothing to measure YET (a
+  // poll landing in a boot or a restore). Error codes are reserved for real
+  // failures reading a sandbox that should be measurable. This is deliberately
+  // unlike GetWorkloadStats, whose caller asserts knowledge the codes then
+  // answer.
+  //
+  // Consumers MUST attribute each sample solely from the identity echoed
+  // inside it, never from a mapping they hold: without an asserted uid, the
+  // response is the only statement of who was measured. Like GetWorkloadStats
+  // it is a pure read, safe on a timer, and does not touch the lifecycle
+  // mutex.
+  rpc GetActiveWorkloadStats(GetActiveWorkloadStatsRequest) returns (GetActiveWorkloadStatsResponse) {}
 }
 
 message RunWorkloadRequest {
@@ -282,14 +303,14 @@ enum StatsSource {
   STATS_SOURCE_GUEST_AGENT = 2;
 }
 
-// GetWorkloadStatsResponse is one resource-usage sample for the executing
-// workload. The unit of measurement is the SANDBOX, which today equals the
-// actor. Per-container attribution is not reported: the micro-VM source could
-// give it, since the guest keeps a cgroup per container and ateom sums them,
-// but the gVisor source cannot split one at all without the sentry's own
-// accounting, and a field only one runtime could ever fill would be worse than
-// none.
-message GetWorkloadStatsResponse {
+// WorkloadStatsSample is one resource-usage sample for an executing workload,
+// shared by both stats reads. The unit of measurement is the SANDBOX, which
+// today equals the actor. Per-container attribution is not reported: the
+// micro-VM source could give it, since the guest keeps a cgroup per container
+// and ateom sums them, but the gVisor source cannot split one at all without
+// the sentry's own accounting, and a field only one runtime could ever fill
+// would be worse than none.
+message WorkloadStatsSample {
   // Identity of the measured actor, retained by ateom from the
   // RunWorkloadRequest / RestoreWorkloadRequest that started it. Echoed back so
   // the caller can attribute the sample without holding its own mapping from
@@ -334,3 +355,35 @@ message GetWorkloadStatsResponse {
 
   int64 observed_at_unix_nano = 12;
 }
+
+message GetWorkloadStatsResponse {
+  WorkloadStatsSample sample = 1;
+}
+
+message GetActiveWorkloadStatsRequest {
+}
+
+// NoSampleReason is why the discovery read has no sample to give -- every
+// value is a normal state a caller with no prior knowledge routinely finds,
+// never an error.
+enum NoSampleReason {
+  NO_SAMPLE_REASON_UNSPECIFIED = 0;
+  // Nothing is executing: the ateom is "available".
+  NO_SAMPLE_REASON_NO_WORKLOAD = 1;
+  // A workload is executing but there are no numbers to give yet: a poll
+  // landing in a boot or a restore, a teardown in progress, or a lifecycle
+  // transition underneath the read. Transient -- take the next sample.
+  NO_SAMPLE_REASON_NOT_MEASURABLE_YET = 2;
+}
+
+message GetActiveWorkloadStatsResponse {
+  // Exactly one of the two is set: either a measurement, or the reason there
+  // is none. An ateom serves one actor at a time, so the sample slot is
+  // singular; the sample is self-describing (see the attribution rule on the
+  // rpc), and a sample being present is itself the statement that a workload
+  // is executing.
+  oneof result {
+    WorkloadStatsSample sample = 1;
+    NoSampleReason no_sample_reason = 2;
+  }
+}
diff --git a/internal/proto/ateompb/ateom_grpc.pb.go b/internal/proto/ateompb/ateom_grpc.pb.go
index 391fd5f60..f3be1ce8b 100644
--- a/internal/proto/ateompb/ateom_grpc.pb.go
+++ b/internal/proto/ateompb/ateom_grpc.pb.go
@@ -33,10 +33,11 @@ import (
 const _ = grpc.SupportPackageIsVersion9
 
 const (
-	Ateom_RunWorkload_FullMethodName        = "/ateom.Ateom/RunWorkload"
-	Ateom_CheckpointWorkload_FullMethodName = "/ateom.Ateom/CheckpointWorkload"
-	Ateom_RestoreWorkload_FullMethodName    = "/ateom.Ateom/RestoreWorkload"
-	Ateom_GetWorkloadStats_FullMethodName   = "/ateom.Ateom/GetWorkloadStats"
+	Ateom_RunWorkload_FullMethodName            = "/ateom.Ateom/RunWorkload"
+	Ateom_CheckpointWorkload_FullMethodName     = "/ateom.Ateom/CheckpointWorkload"
+	Ateom_RestoreWorkload_FullMethodName        = "/ateom.Ateom/RestoreWorkload"
+	Ateom_GetWorkloadStats_FullMethodName       = "/ateom.Ateom/GetWorkloadStats"
+	Ateom_GetActiveWorkloadStats_FullMethodName = "/ateom.Ateom/GetActiveWorkloadStats"
 )
 
 // AteomClient is the client API for Ateom service.
@@ -99,6 +100,26 @@ type AteomClient interface {
 	// "booting" distinguishable from "not here" at all, and it means a workload
 	// that dies during boot is attributable rather than anonymous.
 	GetWorkloadStats(ctx context.Context, in *GetWorkloadStatsRequest, opts ...grpc.CallOption) (*GetWorkloadStatsResponse, error)
+	// GetActiveWorkloadStats samples whatever this ateom is currently
+	// executing, without asserting an identity. It is the discovery read for a
+	// scraper that enumerates ateoms and holds no worker-to-actor mapping;
+	// GetWorkloadStats above is the verified read for a caller that must be
+	// answered about a specific actor.
+	//
+	// Every state a blind caller can find is a normal answer here, never an
+	// error: the response is either a sample or the NoSampleReason there is
+	// none -- nothing to measure ("available"), or nothing to measure YET (a
+	// poll landing in a boot or a restore). Error codes are reserved for real
+	// failures reading a sandbox that should be measurable. This is deliberately
+	// unlike GetWorkloadStats, whose caller asserts knowledge the codes then
+	// answer.
+	//
+	// Consumers MUST attribute each sample solely from the identity echoed
+	// inside it, never from a mapping they hold: without an asserted uid, the
+	// response is the only statement of who was measured. Like GetWorkloadStats
+	// it is a pure read, safe on a timer, and does not touch the lifecycle
+	// mutex.
+	GetActiveWorkloadStats(ctx context.Context, in *GetActiveWorkloadStatsRequest, opts ...grpc.CallOption) (*GetActiveWorkloadStatsResponse, error)
 }
 
 type ateomClient struct {
@@ -149,6 +170,16 @@ func (c *ateomClient) GetWorkloadStats(ctx context.Context, in *GetWorkloadStats
 	return out, nil
 }
 
+func (c *ateomClient) GetActiveWorkloadStats(ctx context.Context, in *GetActiveWorkloadStatsRequest, opts ...grpc.CallOption) (*GetActiveWorkloadStatsResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(GetActiveWorkloadStatsResponse)
+	err := c.cc.Invoke(ctx, Ateom_GetActiveWorkloadStats_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // AteomServer is the server API for Ateom service.
 // All implementations must embed UnimplementedAteomServer
 // for forward compatibility.
@@ -209,6 +240,26 @@ type AteomServer interface {
 	// "booting" distinguishable from "not here" at all, and it means a workload
 	// that dies during boot is attributable rather than anonymous.
 	GetWorkloadStats(context.Context, *GetWorkloadStatsRequest) (*GetWorkloadStatsResponse, error)
+	// GetActiveWorkloadStats samples whatever this ateom is currently
+	// executing, without asserting an identity. It is the discovery read for a
+	// scraper that enumerates ateoms and holds no worker-to-actor mapping;
+	// GetWorkloadStats above is the verified read for a caller that must be
+	// answered about a specific actor.
+	//
+	// Every state a blind caller can find is a normal answer here, never an
+	// error: the response is either a sample or the NoSampleReason there is
+	// none -- nothing to measure ("available"), or nothing to measure YET (a
+	// poll landing in a boot or a restore). Error codes are reserved for real
+	// failures reading a sandbox that should be measurable. This is deliberately
+	// unlike GetWorkloadStats, whose caller asserts knowledge the codes then
+	// answer.
+	//
+	// Consumers MUST attribute each sample solely from the identity echoed
+	// inside it, never from a mapping they hold: without an asserted uid, the
+	// response is the only statement of who was measured. Like GetWorkloadStats
+	// it is a pure read, safe on a timer, and does not touch the lifecycle
+	// mutex.
+	GetActiveWorkloadStats(context.Context, *GetActiveWorkloadStatsRequest) (*GetActiveWorkloadStatsResponse, error)
 	mustEmbedUnimplementedAteomServer()
 }
 
@@ -231,6 +282,9 @@ func (UnimplementedAteomServer) RestoreWorkload(context.Context, *RestoreWorkloa
 func (UnimplementedAteomServer) GetWorkloadStats(context.Context, *GetWorkloadStatsRequest) (*GetWorkloadStatsResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method GetWorkloadStats not implemented")
 }
+func (UnimplementedAteomServer) GetActiveWorkloadStats(context.Context, *GetActiveWorkloadStatsRequest) (*GetActiveWorkloadStatsResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method GetActiveWorkloadStats not implemented")
+}
 func (UnimplementedAteomServer) mustEmbedUnimplementedAteomServer() {}
 func (UnimplementedAteomServer) testEmbeddedByValue()               {}
 
@@ -324,6 +378,24 @@ func _Ateom_GetWorkloadStats_Handler(srv interface{}, ctx context.Context, dec f
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Ateom_GetActiveWorkloadStats_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetActiveWorkloadStatsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AteomServer).GetActiveWorkloadStats(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Ateom_GetActiveWorkloadStats_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(AteomServer).GetActiveWorkloadStats(ctx, req.(*GetActiveWorkloadStatsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // Ateom_ServiceDesc is the grpc.ServiceDesc for Ateom service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -347,6 +419,10 @@ var Ateom_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetWorkloadStats",
 			Handler:    _Ateom_GetWorkloadStats_Handler,
 		},
+		{
+			MethodName: "GetActiveWorkloadStats",
+			Handler:    _Ateom_GetActiveWorkloadStats_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "ateom.proto",