feat(lifecycle): expose pool allocation summary

kubernetes/internal/controller/allocator.go

@@ -269,6 +269,8 @@ func NewAnnoAllocationSyncer(client client.Client) AllocationSyncer {
 }
 
 func (syncer *annoAllocationSyncer) SetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, allocation *SandboxAllocation) error {
+	allocation.PoolRef = sandbox.Spec.PoolRef
+	allocation.Generation = sandbox.Generation
 	js, err := json.Marshal(allocation)
 	if err != nil {
 		return err

kubernetes/internal/controller/allocator_test.go

@@ -350,14 +350,20 @@ func newTestSyncer(sandbox *sandboxv1alpha1.BatchSandbox) (*annoAllocationSyncer
 
 func TestSetAllocation_AddsFinalizer(t *testing.T) {
 	sandbox := &sandboxv1alpha1.BatchSandbox{
-		ObjectMeta: metav1.ObjectMeta{Name: "sbx1", Namespace: "default"},
+		ObjectMeta: metav1.ObjectMeta{Name: "sbx1", Namespace: "default", Generation: 7},
 		Spec:       sandboxv1alpha1.BatchSandboxSpec{PoolRef: "pool1"},
 	}
 	syncer, sbx := newTestSyncer(sandbox)
 
 	err := syncer.SetAllocation(context.Background(), sbx, &SandboxAllocation{Pods: []string{"pod1"}})
 	assert.NoError(t, err)
 	assert.Contains(t, sbx.Finalizers, FinalizerPoolAllocation)
+
+	allocation, err := syncer.GetAllocation(context.Background(), sbx)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"pod1"}, allocation.Pods)
+	assert.Equal(t, "pool1", allocation.PoolRef)
+	assert.Equal(t, int64(7), allocation.Generation)
 }
 
 func TestSetReleased_FinalizerBehavior(t *testing.T) {

kubernetes/internal/controller/apis.go

@@ -39,7 +39,9 @@ const (
 var AnnotationSandboxEndpoints = pkgutils.AnnotationEndpoints
 
 type SandboxAllocation struct {
-	Pods []string `json:"pods"`
+	Pods       []string `json:"pods"`
+	PoolRef    string   `json:"poolRef"`
+	Generation int64    `json:"generation"`
 }
 
 type AllocationRelease struct {

server/opensandbox_server/api/schema.py

@@ -566,6 +566,17 @@ class Config:
         populate_by_name = True
 
 
+class AllocationSummary(BaseModel):
+    """Current runtime-confirmed pool allocation summary."""
+    mode: Literal["pool"] = Field("pool", description="Allocation mode.")
+    pool_ref: str = Field(..., alias="poolRef", description="Concrete pool reference currently allocated.")
+    state: Literal["allocated"] = Field("allocated", description="Current confirmed allocation state.")
+
+    class Config:
+        populate_by_name = True
+        extra = "forbid"
+
+
 class Sandbox(BaseModel):
     """
     Runtime execution environment provisioned from a container image.
@@ -592,6 +603,14 @@ class Sandbox(BaseModel):
         None,
         description="Opaque extension data restored from provider-specific storage",
     )
+    allocation: Optional[AllocationSummary] = Field(
+        None,
+        description=(
+            "Current runtime-confirmed pool allocation summary. Omitted unless the runtime confirms "
+            "an active pool allocation; it is not a request echo, allocation history, readiness signal, "
+            "or Kubernetes introspection result."
+        ),
+    )
     entrypoint: Optional[List[str]] = Field(None, description="The command to execute as the sandbox's entry process")
     expires_at: Optional[datetime] = Field(
         None,

server/opensandbox_server/services/k8s/workload_mapper.py

@@ -14,9 +14,17 @@
 
 from __future__ import annotations
 
+import json
+import re
 from typing import Any, Optional
 
-from opensandbox_server.api.schema import ImageSpec, PlatformSpec, Sandbox, SandboxStatus
+from opensandbox_server.api.schema import (
+    AllocationSummary,
+    ImageSpec,
+    PlatformSpec,
+    Sandbox,
+    SandboxStatus,
+)
 from opensandbox_server.extensions import extract_extensions_from_mapping
 from opensandbox_server.services.constants import SANDBOX_ID_LABEL, SANDBOX_SNAPSHOT_ID_LABEL
 
@@ -43,6 +51,7 @@ def _build_sandbox_from_workload(workload: Any, workload_provider: Any) -> Sandb
     snapshot_id = labels.get(SANDBOX_SNAPSHOT_ID_LABEL)
     expires_at = workload_provider.get_expiration(workload)
     status_info = workload_provider.get_status(workload)
+    workload_status = workload.get("status", {}) if isinstance(workload, dict) else workload.status
 
     user_metadata = {
         k: v for k, v in labels.items() if not _is_opensandbox_label(k)
@@ -67,6 +76,7 @@ def _build_sandbox_from_workload(workload: Any, workload_provider: Any) -> Sandb
     if not snapshot_id:
         image_spec = ImageSpec(uri=image_uri) if image_uri else ImageSpec(uri="unknown")
     platform_spec = _extract_platform_from_workload(workload)
+    allocation = _extract_confirmed_pool_allocation(metadata, spec, workload_status)
     return Sandbox(
         id=sandbox_id,
         status=SandboxStatus(
@@ -83,9 +93,77 @@ def _build_sandbox_from_workload(workload: Any, workload_provider: Any) -> Sandb
         snapshotId=snapshot_id,
         entrypoint=entrypoint,
         platform=platform_spec,
+        allocation=allocation,
     )
 
 
+_POD_NAME_PATTERN = re.compile(r"^[a-z0-9](?:[-a-z0-9.]{0,251}[a-z0-9])?$")
+
+
+def _extract_confirmed_pool_allocation(
+    metadata: Any,
+    spec: Any,
+    status: Any,
+) -> Optional[AllocationSummary]:
+    """Return a summary only when current pool allocation evidence is complete."""
+    pool_ref = _field(spec, "poolRef", "pool_ref")
+    if not isinstance(pool_ref, str) or not pool_ref.strip() or pool_ref == "*":
+        return None
+
+    if _field(metadata, "deletionTimestamp", "deletion_timestamp"):
+        return None
+    finalizers = _field(metadata, "finalizers")
+    if (
+        not isinstance(finalizers, list)
+        or "pool.sandbox.opensandbox.io/pool-allocation" not in finalizers
+    ):
+        return None
+
+    annotations = _field(metadata, "annotations")
+    if not isinstance(annotations, dict):
+        return None
+    raw_allocation = annotations.get("sandbox.opensandbox.io/alloc-status")
+    if not isinstance(raw_allocation, str):
+        return None
+    try:
+        annotation = json.loads(raw_allocation)
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(annotation, dict) or annotation.get("poolRef") != pool_ref:
+        return None
+
+    pods = annotation.get("pods")
+    if (
+        not isinstance(pods, list)
+        or not pods
+        or any(not isinstance(pod, str) or not _POD_NAME_PATTERN.fullmatch(pod) for pod in pods)
+        or len(set(pods)) != len(pods)
+    ):
+        return None
+    allocated = _field(status, "allocated")
+    if (
+        not isinstance(allocated, int)
+        or isinstance(allocated, bool)
+        or allocated != len(pods)
+    ):
+        return None
+
+    return AllocationSummary(poolRef=pool_ref)
+
+
+def _field(value: Any, *names: str) -> Any:
+    if isinstance(value, dict):
+        for name in names:
+            if name in value:
+                return value[name]
+        return None
+    for name in names:
+        field = getattr(value, name, None)
+        if field is not None:
+            return field
+    return None
+
+
 def _extract_platform_from_workload(workload: Any) -> Optional[PlatformSpec]:
     if isinstance(workload, dict):
         spec = workload.get("spec") or {}

server/tests/k8s/test_workload_mapper.py

@@ -12,6 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
+from types import SimpleNamespace
+
+import pytest
+
 from opensandbox_server.services.k8s.workload_mapper import (
     _build_sandbox_from_workload,
     _extract_platform_from_workload,
@@ -51,6 +56,99 @@ def test_restores_extensions_from_annotations(self):
 
         assert sandbox.extensions == {"opensandbox.extensions.custom-label": "中文数据"}
 
+    def test_returns_confirmed_pool_allocation_for_dict_workload(self):
+        sandbox = _build_sandbox_from_workload(_allocated_workload(), _WorkloadProvider())
+
+        assert sandbox.allocation is not None
+        assert sandbox.allocation.model_dump(by_alias=True) == {
+            "mode": "pool",
+            "poolRef": "pool-runc",
+            "state": "allocated",
+        }
+
+    def test_returns_confirmed_pool_allocation_for_object_workload(self):
+        workload = SimpleNamespace(
+            metadata=SimpleNamespace(
+                labels={"opensandbox.io/id": "sandbox-1"},
+                annotations={
+                    "sandbox.opensandbox.io/alloc-status": json.dumps(
+                        {"pods": ["pod-1"], "poolRef": "pool-runc", "generation": 4}
+                    )
+                },
+                finalizers=["pool.sandbox.opensandbox.io/pool-allocation"],
+                deletion_timestamp=None,
+                creation_timestamp="2026-06-22T00:00:00Z",
+            ),
+            spec=SimpleNamespace(pool_ref="pool-runc", containers=[]),
+            status=SimpleNamespace(allocated=1),
+        )
+
+        sandbox = _build_sandbox_from_workload(workload, _WorkloadProvider())
+
+        assert sandbox.allocation is not None
+        assert sandbox.allocation.pool_ref == "pool-runc"
+
+    @pytest.mark.parametrize(
+        ("name", "mutate"),
+        [
+            ("wrong annotation pool reference", lambda w: _set_annotation(w, {"pods": ["pod-1"], "poolRef": "other", "generation": 4})),
+            ("missing annotation pool reference", lambda w: _set_annotation(w, {"pods": ["pod-1"], "generation": 4})),
+            ("legacy pods-only annotation", lambda w: _set_annotation(w, {"pods": ["pod-1"]})),
+            ("deleting", lambda w: w["metadata"].update({"deletionTimestamp": "2026-06-23T00:00:00Z"})),
+            ("missing finalizer", lambda w: w["metadata"].update({"finalizers": []})),
+            ("missing allocation annotation", lambda w: w["metadata"].update({"annotations": {}})),
+            ("invalid annotation JSON", lambda w: w["metadata"]["annotations"].update({"sandbox.opensandbox.io/alloc-status": "{"})),
+            ("empty pods", lambda w: _set_annotation(w, {"pods": [], "poolRef": "pool-runc", "generation": 4})),
+            ("empty pod name", lambda w: _set_annotation(w, {"pods": [""], "poolRef": "pool-runc", "generation": 4})),
+            ("invalid pod name", lambda w: _set_annotation(w, {"pods": ["Pod-1"], "poolRef": "pool-runc", "generation": 4})),
+            ("duplicate pod names", lambda w: _set_annotation(w, {"pods": ["pod-1", "pod-1"], "poolRef": "pool-runc", "generation": 4})),
+            ("allocated count mismatch", lambda w: w["status"].update({"allocated": 2})),
+            ("wildcard pool reference", lambda w: w["spec"].update({"poolRef": "*"})),
+            ("non-pool workload", lambda w: w["spec"].pop("poolRef")),
+        ],
+    )
+    def test_omits_unconfirmed_pool_allocation(self, name, mutate):
+        workload = _allocated_workload()
+        mutate(workload)
+
+        sandbox = _build_sandbox_from_workload(workload, _WorkloadProvider())
+
+        assert sandbox.allocation is None, name
+
+    def test_renewal_generation_drift_remains_confirmed(self):
+        workload = _allocated_workload()
+        workload["metadata"]["generation"] = 12
+        _set_annotation(
+            workload,
+            {"pods": ["pod-1"], "poolRef": "pool-runc", "generation": 4},
+        )
+
+        sandbox = _build_sandbox_from_workload(workload, _WorkloadProvider())
+
+        assert sandbox.allocation is not None
+        assert sandbox.allocation.pool_ref == "pool-runc"
+
+
+def _allocated_workload(pool_ref="pool-runc"):
+    return {
+        "metadata": {
+            "labels": {"opensandbox.io/id": "sandbox-1"},
+            "annotations": {
+                "sandbox.opensandbox.io/alloc-status": json.dumps(
+                    {"pods": ["pod-1"], "poolRef": pool_ref, "generation": 4}
+                )
+            },
+            "finalizers": ["pool.sandbox.opensandbox.io/pool-allocation"],
+            "creationTimestamp": "2026-06-22T00:00:00Z",
+        },
+        "spec": {"poolRef": pool_ref, "template": None},
+        "status": {"allocated": 1},
+    }
+
+
+def _set_annotation(workload, allocation):
+    workload["metadata"]["annotations"]["sandbox.opensandbox.io/alloc-status"] = json.dumps(allocation)
+
 
 class TestExtractPlatformFromWorkload:
     """Regression tests for _extract_platform_from_workload.

server/tests/test_routes_get_sandbox.py

@@ -18,7 +18,7 @@
 from fastapi.testclient import TestClient
 
 from opensandbox_server.api import lifecycle
-from opensandbox_server.api.schema import ImageSpec, Sandbox, SandboxStatus
+from opensandbox_server.api.schema import AllocationSummary, ImageSpec, Sandbox, SandboxStatus
 
 
 def test_get_sandbox_returns_service_payload(
@@ -37,6 +37,7 @@ def get_sandbox(sandbox_id: str) -> Sandbox:
                 image=ImageSpec(uri="python:3.11"),
                 status=SandboxStatus(state="Running"),
                 metadata={"team": "infra"},
+                allocation=AllocationSummary(poolRef="pool-runc"),
                 entrypoint=["python", "-V"],
                 expiresAt=now + timedelta(hours=1),
                 createdAt=now,
@@ -51,6 +52,11 @@ def get_sandbox(sandbox_id: str) -> Sandbox:
     assert payload["id"] == "sbx-001"
     assert payload["status"]["state"] == "Running"
     assert payload["image"]["uri"] == "python:3.11"
+    assert payload["allocation"] == {
+        "mode": "pool",
+        "poolRef": "pool-runc",
+        "state": "allocated",
+    }
 
 
 def test_get_sandbox_propagates_not_found(
@@ -108,6 +114,7 @@ def get_sandbox(sandbox_id: str) -> Sandbox:
     payload = response.json()
     assert "expiresAt" not in payload
     assert "metadata" not in payload
+    assert "allocation" not in payload
     assert "reason" not in payload["status"]
     assert "message" not in payload["status"]
     assert "lastTransitionAt" not in payload["status"]

specs/sandbox-lifecycle.yml

@@ -1079,6 +1079,13 @@ components:
             type: string
           description: Opaque extension data restored from provider-specific storage
 
+        allocation:
+          $ref: '#/components/schemas/AllocationSummary'
+          description: |
+            Current runtime-confirmed pool allocation. Omitted unless an active pool
+            allocation is confirmed; this is not a request echo, allocation history,
+            readiness signal, or Kubernetes introspection result.
+
         entrypoint:
           type: array
           items:
@@ -1104,6 +1111,23 @@ components:
         - status
         - createdAt
         - entrypoint
+    AllocationSummary:
+      type: object
+      description: Public summary of a confirmed active pool allocation.
+      properties:
+        mode:
+          type: string
+          enum: [pool]
+          description: Allocation mode.
+        poolRef:
+          type: string
+          description: Concrete pool reference currently allocated.
+        state:
+          type: string
+          enum: [allocated]
+          description: Current confirmed allocation state.
+      required: [mode, poolRef, state]
+      additionalProperties: false
     SandboxState:
       type: string
       description: |