refactor(BA-7353): move the kernel payload into common/interchange and carry only what the manager reads

Review follow-up. `common/data` is where a component keeps its own value objects, so a
payload that crosses a component boundary had no documented home — `dto/` is organized
by target component, and a payload has two sides.

- Add `common/interchange/`, with `AGENTS.md` stating the membership test: one component
  produces it, another consumes it, so it has to survive serialization. Record the split
  between `interchange` / `dto` / `schema` / `data` in `common/KNOWLEDGE.md`, and list the
  package (along with the previously missing `schema/`) in `common/AGENTS.md`.
- Carry only what `update_kernel_status_running()` reads, which is its one consumer:
  `container_id`, `kernel_host`, the two repl ports, `service_ports`, `attached_devices`
  and `allocations`. Dropped `id` (the event already names the kernel), `agent_addr` (the
  event names its source), the legacy stdio ports, and the parts of the resource spec —
  slots, mounts, unified devices, scratch size — that nothing reads.
- Return `list[ResourceSlotEntry]` rather than the legacy `ResourceSlot`; a caller that
  needs the dict form goes through `ResourceSlotEntry.inputs_to_resource_slot()`.
- Name the levels of the allocation mapping (`DeviceAllocation`, `PerDeviceAllocation`)
  instead of nesting three dicts inline, and apply the `DeviceAllocation` alias that
  `agent/resources.py` already defined but did not use.

With the spec no longer travelling whole, `KernelResourceSpecData` / `MountData` /
`to_data()` would only serve the agent's own log, so they are gone and
`to_json_serializable_dict()` is left as it was.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: fregataa

src/ai/backend/agent/agent.py

@@ -2713,7 +2713,7 @@ async def create_kernel(
                     "create_kernel(kernel:{}, session:{}) resource spec prepared: {}",
                     kernel_id,
                     session_id,
-                    resource_spec.to_data().model_dump_json(),
+                    resource_spec.to_json(),
                 )
 
                 # Mount backend-specific intrinsic mounts (e.g., scratch directories)

src/ai/backend/agent/resources.py

@@ -40,9 +40,7 @@
     InvalidResourceConfigError,
     ResourceOverAllocatedError,
 )
-from ai.backend.common.data.kernel.types import KernelResourceSpecData, MountData
 from ai.backend.common.etcd import AsyncEtcd
-from ai.backend.common.identifier.resource_slot import ResourceSlotName
 from ai.backend.common.json import dump_json_str, load_json
 from ai.backend.common.plugin import AbstractPlugin, BasePluginContext
 from ai.backend.common.types import (
@@ -57,7 +55,6 @@
     MountPermission,
     MountTypes,
     ResourceSlot,
-    ResourceSlotEntry,
     SlotName,
     SlotTypes,
     aobject,
@@ -126,7 +123,7 @@ class KernelResourceSpec:
     slots: ResourceSlot
     """Stores the original user-requested resource slots."""
 
-    allocations: MutableMapping[DeviceName, Mapping[SlotName, Mapping[DeviceId, Decimal]]]
+    allocations: MutableMapping[DeviceName, DeviceAllocation]
     """
     Represents the resource allocations for each slot (device) type and devices.
     """
@@ -265,34 +262,35 @@ async def aread_from_file(cls, file: AsyncTextIOWrapper) -> Self:
         text = "\n".join(await file.readlines())
         return cls.read_from_string(text)
 
-    def to_data(self) -> KernelResourceSpecData:
-        """
-        Render this spec as the value events and logs carry it as.
-        """
-        return KernelResourceSpecData(
-            slots=ResourceSlotEntry.from_resource_slot(self.slots),
-            allocations={
-                dev_name: {
-                    ResourceSlotName(str(slot_name)): dict(per_device_alloc)
-                    for slot_name, per_device_alloc in dev_alloc.items()
-                }
-                for dev_name, dev_alloc in self.allocations.items()
-            },
-            scratch_disk_size=self.scratch_disk_size,
-            mounts=[
-                MountData(
-                    type=mount.type,
-                    source=mount.source,
-                    target=mount.target,
-                    permission=mount.permission,
-                )
-                for mount in self.mounts
-            ],
-            unified_devices=[
-                (device_name, ResourceSlotName(str(slot_name)))
-                for device_name, slot_name in self.unified_devices
-            ],
-        )
+    def to_json_serializable_dict(self) -> Mapping[str, Any]:
+        o = attrs.asdict(self)
+        for slot_name, alloc in o["slots"].items():
+            if known_slot_types.get(slot_name, "count") == "bytes":
+                o["slots"] = f"{BinarySize(alloc):s}"
+            else:
+                o["slots"] = str(alloc)
+        serialized_allocations = {}
+        for dev_name, dev_alloc in o["allocations"].items():
+            serialized_dev_alloc = {}
+            for slot_name, per_device_alloc in dev_alloc.items():
+                serialized_per_device_alloc = {}
+                for dev_id, alloc in per_device_alloc.items():
+                    if known_slot_types.get(slot_name, "count") == "bytes":
+                        serialized_alloc = f"{BinarySize(alloc):s}"
+                    else:
+                        serialized_alloc = str(alloc)
+                    serialized_per_device_alloc[str(dev_id)] = serialized_alloc
+                serialized_dev_alloc[str(slot_name)] = serialized_per_device_alloc
+            serialized_allocations[str(dev_name)] = serialized_dev_alloc
+        o["allocations"] = serialized_allocations
+        o["mounts"] = list(map(str, self.mounts))
+        o["unified_devices"] = [
+            (str(device_name), str(slot_name)) for device_name, slot_name in self.unified_devices
+        ]
+        return o
+
+    def to_json(self) -> str:
+        return dump_json_str(self.to_json_serializable_dict())
 
     @classmethod
     def __get_pydantic_core_schema__(

src/ai/backend/common/AGENTS.md

@@ -26,6 +26,8 @@
 | `common/events/` | Event type definitions and dispatcher — see existing `AbstractEvent` subclasses |
 | `common/bgtask/` | Background task framework — extend `BaseBackgroundTaskHandler` |
 | `common/dto/` | Inter-component DTOs — see `common/dto/AGENTS.md` |
+| `common/interchange/` | Payloads one component sends another (event/RPC bodies) — see `common/interchange/AGENTS.md` |
+| `common/schema/` | Pydantic types the manager persists as a JSON DB column — see `common/schema/AGENTS.md` |
 | `common/exception.py` | Root `BackendAIError` and `ErrorCode` — all component exceptions inherit from here |
 | `common/types.py` | Common base types used across layers |
 

src/ai/backend/common/KNOWLEDGE.md

@@ -27,3 +27,20 @@ types belong here is the core of this document.
 - The second criterion: types that, if placed in a higher layer, **would force an upward import** (`common/schema/` states this explicitly and is used only by the manager).
 - Single-consumer subpackages (`stage/` — agent-only, `resilience/` — manager-only) are not defects.
 - "It is convenient to put it somewhere neutral" is not a membership reason.
+
+## Shared types are split by what crosses, not by who owns them
+
+A type shared by two components still has to land in one of four places, and the
+question that separates them is what boundary it crosses:
+
+| Package | Crosses |
+|---------|---------|
+| `interchange/` | A component boundary as a serialized body (event/RPC payload) |
+| `dto/` | An API boundary, per target component |
+| `schema/` | The persistence boundary — the manager stores it as a JSON column |
+| `data/` | Nothing; it is passed within a component |
+
+`interchange/` exists because the first row had no home: `dto/` is organized by target
+component, and a payload has two sides, so neither side owns it. Its membership test is
+the consumer, not the producer — a field the receiving component never reads belongs in
+the producing component, not on the wire.

src/ai/backend/common/data/kernel/types.py

@@ -0,0 +1,34 @@
+# `common/interchange` guardrails
+
+The bodies components send each other. A type belongs here when one component produces it,
+another consumes it, and it therefore has to survive serialization — event payloads and RPC
+payloads.
+
+## Rules
+
+- **Contents**: pydantic models only. Every field must be JSON-representable and round-trip
+  unchanged — no `Any`, and no type that only survives pickling (`ResourceSlot`, `SlotName`,
+  `BinarySize`, `Path` used as a bare value).
+- **Carry what the consumer reads.** A payload mirrors the contract between the two components,
+  not the producer's internal struct. A field no consumer reads does not belong on the wire; keep
+  it in the producing component.
+- **Organized by domain**, one module per domain (`interchange/kernel.py`), not by component —
+  a payload has two sides, so neither side owns the directory.
+- **Dependency direction (leaf)**: depend only on lower `common` modules (`common.types`,
+  `common.identifier`). MUST NOT import from `manager` / `agent` / `storage`, or from
+  `common.dto`.
+- **No business logic.** A conversion that only reshapes the payload's own fields is fine
+  (`to_resource_slot_entries()`); anything that needs a repository, a config, or another entity is
+  not.
+
+## How this differs from its neighbours
+
+| Package | Holds |
+|---------|-------|
+| `interchange/` | Payloads crossing a component boundary — serialized by definition |
+| `dto/` | API request/response contracts, per target component (`dto/manager/v2/...`) |
+| `schema/` | Pydantic types the manager persists as a JSON DB column |
+| `data/` | Value objects passed within a component |
+
+When a payload type also has to be persisted as a column, it belongs in `schema/`, and
+`interchange/` may reference it.

src/ai/backend/common/interchange/AGENTS.md

@@ -0,0 +1 @@
+AGENTS.md

src/ai/backend/common/interchange/CLAUDE.md

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Annotated
+
+from pydantic import Field
+
+from ai.backend.common.identifier.resource_slot import ResourceSlotName
+from ai.backend.common.types import (
+    BackendAISchema,
+    ContainerId,
+    DeviceId,
+    DeviceName,
+    ResourceSlotEntry,
+    ServicePortProtocols,
+)
+
+__all__ = (
+    "AttachedDeviceData",
+    "DeviceAllocation",
+    "DeviceCapacityData",
+    "KernelCreationInfo",
+    "PerDeviceAllocation",
+    "ServicePortData",
+)
+
+
+type AllocationAmount = Annotated[
+    Decimal,
+    # An unbounded allocation is expressed as `Decimal("Infinity")`, which the default
+    # `Decimal` constraint rejects. It survives the wire as the string "Infinity".
+    Field(allow_inf_nan=True),
+]
+
+type PerDeviceAllocation = dict[DeviceId, AllocationAmount]
+"""How much of one slot each individual device contributed."""
+
+type DeviceAllocation = dict[ResourceSlotName, PerDeviceAllocation]
+"""The slots one device type served, and the per-device amounts behind each."""
+
+
+class DeviceCapacityData(BackendAISchema):
+    mem: int | None = None
+    proc: int | None = None
+
+
+class AttachedDeviceData(BackendAISchema):
+    device_id: DeviceId
+    model_name: str
+    data: DeviceCapacityData = Field(default_factory=DeviceCapacityData)
+
+
+class ServicePortData(BackendAISchema):
+    name: str
+    protocol: ServicePortProtocols
+    container_ports: list[int] = Field(default_factory=list)
+    host_ports: list[int | None] = Field(default_factory=list)
+    is_inference: bool = False
+
+
+class KernelCreationInfo(BackendAISchema):
+    """
+    What the agent reports about a kernel once its container is up.
+
+    This is the contract with the manager, not a rendering of the agent's own resource
+    spec: it carries the facts the manager records against the kernel and nothing else.
+    """
+
+    container_id: ContainerId
+    kernel_host: str
+    repl_in_port: int
+    repl_out_port: int
+    service_ports: list[ServicePortData] = Field(default_factory=list)
+    attached_devices: dict[DeviceName, list[AttachedDeviceData]] = Field(default_factory=dict)
+    allocations: dict[DeviceName, DeviceAllocation] = Field(default_factory=dict)
+
+    def to_resource_slot_entries(self) -> list[ResourceSlotEntry]:
+        """
+        Sum the per-device allocations into the occupancy the manager accounts by.
+
+        A slot holding no device allocation is left out rather than reported as zero,
+        which is the distinction a caller storing the result as occupancy depends on.
+        """
+        entries: list[ResourceSlotEntry] = []
+        for device_allocation in self.allocations.values():
+            for slot_name, per_device in device_allocation.items():
+                if not per_device:
+                    continue
+                total = sum(per_device.values(), Decimal(0))
+                entries.append(ResourceSlotEntry(resource_type=slot_name, quantity=str(total)))
+        return entries