diff --git a/docs/meta.md b/docs/meta.md
index 1f6a8b9..bb98e93 100644
--- a/docs/meta.md
+++ b/docs/meta.md
@@ -14,7 +14,7 @@ you who owns the value.
 | Prefix | Owner and use | Keys |
 |---|---|---|
 | `cocoonset.cocoonstack.io/` | CocoonSet CRD group, pod selector labels, and CocoonSet-level fields the operator mirrors onto a managed pod | `name`, `role`, `slot`, `mode`, `image`, `os`, `storage`, `snapshot-policy`, `network`, `managed`, `force-pull`, `generation`, `hibernated-on-node` |
-| `vm.cocoonstack.io/` | VM-instance metadata — observed runtime state plus the per-VM spec the operator hands to vk-cocoon | `id`, `name`, `ip`, `vnc-port`, `hibernate`, `restore-from-hibernate`, `keep-snapshot-on-delete`, `fork-from`, `clone-from-dir`, `conn-type`, `backend`, `no-direct-io`, `probe-port`, `lifecycle-state`, `lifecycle-observed-generation`, `lifecycle-state-message`, `snapshot-annotations` |
+| `vm.cocoonstack.io/` | VM-instance metadata — observed runtime state plus the per-VM spec the operator hands to vk-cocoon | `id`, `name`, `ip`, `vnc-port`, `hibernate`, `restore-from-hibernate`, `keep-snapshot-on-delete`, `fork-from`, `clone-from-dir`, `conn-type`, `backend`, `no-direct-io`, `probe-port`, `lifecycle-state`, `lifecycle-observed-generation`, `lifecycle-state-message`, `snapshot-annotations`, `snapshot`, `snapshot-status` |
 | `cocoonstack.io/` | Node labels vk-cocoon stamps on its virtual node and the operator selects on | `pool`, `snapshot-cpu-class` |
 
 `meta.KindCocoonSet` and `meta.TolerationKey`
@@ -138,6 +138,18 @@ registry; the engine never interprets them. `meta.ReadSnapshotAnnotations`
 rejects a malformed value and any key under the `cocoonstack.` prefix, which the
 snapshot wire format reserves for its own annotations.
 
+`meta.AnnotationSnapshotRequest` (`vm.cocoonstack.io/snapshot`) asks vk-cocoon
+to push a live snapshot of the pod's running VM to a registry-relative
+`repo[:tag]` (the tag defaults to `latest`). The VM pauses only for the save
+and keeps running; the snapshot keeps its NICs, so it can serve as a clone
+source. vk-cocoon serves a request once the VM is Ready and no hibernate is
+asked for, and answers in `meta.AnnotationSnapshotStatus`
+(`vm.cocoonstack.io/snapshot-status`), a JSON `meta.SnapshotStatus` whose `ref`
+echoes the request and whose `state` is `pushed` or `failed`; a `failed` answer
+carries a `message`. A ref is answered once, pushed or failed, and re-patching
+an answered ref does nothing: every push, a retry included, uses a new tag, and
+the client waits for a status carrying that ref.
+
 ## Lifecycle status
 
 `meta.LifecycleStatus` is the typed contract for the annotation triple
diff --git a/meta/keys.go b/meta/keys.go
index 1fe7dcc..7e6cd97 100644
--- a/meta/keys.go
+++ b/meta/keys.go
@@ -76,6 +76,8 @@ const (
 	AnnotationLifecycleStateMessage = "vm.cocoonstack.io/lifecycle-state-message"
 	// AnnotationSnapshotAnnotations carries a JSON object of OCI annotations stamped on every snapshot pushed for the pod.
 	AnnotationSnapshotAnnotations = "vm.cocoonstack.io/snapshot-annotations"
+	AnnotationSnapshotRequest     = "vm.cocoonstack.io/snapshot"
+	AnnotationSnapshotStatus      = "vm.cocoonstack.io/snapshot-status"
 
 	// RoleMain identifies the main agent VM (slot 0).
 	RoleMain = "main"
diff --git a/meta/snapshot.go b/meta/snapshot.go
index 63276cc..8516827 100644
--- a/meta/snapshot.go
+++ b/meta/snapshot.go
@@ -8,7 +8,25 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-const reservedSnapshotAnnotationPrefix = "cocoonstack."
+const (
+	SnapshotStatePushed SnapshotState = "pushed"
+	SnapshotStateFailed SnapshotState = "failed"
+
+	reservedSnapshotAnnotationPrefix = "cocoonstack."
+)
+
+type SnapshotState string
+
+type SnapshotStatus struct {
+	Ref     string        `json:"ref"`
+	State   SnapshotState `json:"state"`
+	Message string        `json:"message,omitempty"`
+}
+
+func (s SnapshotStatus) Annotations() map[string]any {
+	raw, _ := json.Marshal(s)
+	return map[string]any{AnnotationSnapshotStatus: string(raw)}
+}
 
 // ReadSnapshotAnnotations reads the caller's OCI annotations for the pod's pushed snapshots; keys under the cocoonstack. prefix are reserved.
 func ReadSnapshotAnnotations(pod *corev1.Pod) (map[string]string, error) {
@@ -27,3 +45,11 @@ func ReadSnapshotAnnotations(pod *corev1.Pod) (map[string]string, error) {
 	}
 	return annotations, nil
 }
+
+func ReadSnapshotStatus(pod *corev1.Pod) SnapshotStatus {
+	var s SnapshotStatus
+	if raw := pod.Annotations[AnnotationSnapshotStatus]; raw != "" && json.Unmarshal([]byte(raw), &s) != nil {
+		return SnapshotStatus{}
+	}
+	return s
+}
diff --git a/meta/snapshot_test.go b/meta/snapshot_test.go
index 47ef74c..816ad14 100644
--- a/meta/snapshot_test.go
+++ b/meta/snapshot_test.go
@@ -39,3 +39,25 @@ func TestReadSnapshotAnnotations(t *testing.T) {
 		})
 	}
 }
+
+func TestSnapshotStatusRoundTrip(t *testing.T) {
+	want := SnapshotStatus{Ref: "user-snapshots/u1/s1", State: SnapshotStateFailed, Message: "vm is not running"}
+	pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{}}}
+	for k, v := range want.Annotations() {
+		pod.Annotations[k] = v.(string)
+	}
+	if got := ReadSnapshotStatus(pod); got != want {
+		t.Fatalf("got %+v, want %+v", got, want)
+	}
+}
+
+func TestReadSnapshotStatusUnanswered(t *testing.T) {
+	for name, raw := range map[string]string{"absent": "", "malformed": `{"ref":`} {
+		t.Run(name, func(t *testing.T) {
+			pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{AnnotationSnapshotStatus: raw}}}
+			if got := ReadSnapshotStatus(pod); got.Ref != "" {
+				t.Errorf("ref = %q, want unanswered", got.Ref)
+			}
+		})
+	}
+}
diff --git a/snapshot/push.go b/snapshot/push.go
index 6ff656b..4706de7 100644
--- a/snapshot/push.go
+++ b/snapshot/push.go
@@ -23,6 +23,7 @@ import (
 // PushOptions configures a snapshot push operation.
 type PushOptions struct {
 	Name        string
+	LocalName   string
 	Tag         string
 	BaseImage   string // optional cocoonstack.snapshot.baseimage annotation
 	Source      string
@@ -53,9 +54,10 @@ func (p *Pusher) Push(ctx context.Context, opts PushOptions) error {
 		p = &Pusher{Uploader: s.UploadSession(), Cocoon: p.Cocoon}
 	}
 
-	stream, wait, err := p.Cocoon.Export(ctx, opts.Name)
+	localName := cmp.Or(opts.LocalName, opts.Name)
+	stream, wait, err := p.Cocoon.Export(ctx, localName)
 	if err != nil {
-		return fmt.Errorf("cocoon export %s: %w", opts.Name, err)
+		return fmt.Errorf("cocoon export %s: %w", localName, err)
 	}
 
 	entries, readErr := p.readAndUploadEntries(ctx, opts, stream)
diff --git a/snapshot/snapshot_test.go b/snapshot/snapshot_test.go
index 18db7bf..1645470 100644
--- a/snapshot/snapshot_test.go
+++ b/snapshot/snapshot_test.go
@@ -147,6 +147,35 @@ func TestPushOmitsBaseImageAnnotationWhenEmpty(t *testing.T) {
 	}
 }
 
+func TestPushExportsLocalNameAndUploadsUnderName(t *testing.T) {
+	for _, tt := range []struct {
+		name       string
+		localName  string
+		wantExport string
+	}{
+		{name: "defaults to name", wantExport: "user/s1"},
+		{name: "local name", localName: "snap-myvm", wantExport: "snap-myvm"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			cocoon := &fakeCocoon{exportTar: buildExportTar(t, snapshotExportConfig{Name: "myvm"}, map[string][]byte{
+				"config.json": []byte(`{}`),
+			})}
+			uploader := newFakeUploader()
+			pusher := &Pusher{Uploader: uploader, Cocoon: cocoon}
+
+			if err := pusher.Push(t.Context(), PushOptions{Name: "user/s1", LocalName: tt.localName, Tag: "v2"}); err != nil {
+				t.Fatalf("Push: %v", err)
+			}
+			if len(cocoon.exported) != 1 || cocoon.exported[0] != tt.wantExport {
+				t.Errorf("exported %v, want [%s]", cocoon.exported, tt.wantExport)
+			}
+			if _, ok := uploader.manifests["user/s1:v2"]; !ok {
+				t.Error("no manifest at user/s1:v2")
+			}
+		})
+	}
+}
+
 func TestPushStampsCallerAnnotations(t *testing.T) {
 	cocoon := &fakeCocoon{exportTar: buildExportTar(t, snapshotExportConfig{Name: "myvm"}, map[string][]byte{
 		"config.json": []byte(`{}`),
@@ -529,9 +558,11 @@ func (f *fakeUploader) GetBlob(_ context.Context, _, digest string) (io.ReadClos
 
 type fakeCocoon struct {
 	exportTar []byte
+	exported  []string
 }
 
-func (f *fakeCocoon) Export(_ context.Context, _ string) (io.ReadCloser, func() error, error) {
+func (f *fakeCocoon) Export(_ context.Context, name string) (io.ReadCloser, func() error, error) {
+	f.exported = append(f.exported, name)
 	return io.NopCloser(bytes.NewReader(f.exportTar)), func() error { return nil }, nil
 }