fix(resource): add create_sandbox/project_access stubs and fix ambiguous step

- Add create_sandbox() override to _ContainerBaseHandler raising NotImplementedError, matching CloudResourceHandler pattern (issue #836) - Add project_access() override to _ContainerBaseHandler raising NotImplementedError, consistent with all other stub methods - Rename @then("the import should succeed without errors") step to @then("the container handler module import should succeed without errors") to resolve AmbiguousStep collision with tdd_a2a_sdk_dependency_steps.py - Update container_handler.feature to use renamed step - Apply ruff format to container.py ISSUES CLOSED: #2907
fix(resource): implement missing container handler module for container infrastructure resource types
2026-05-30 13:08:44 -04:00 · 2026-05-30 13:08:44 -04:00 · 2026-05-30 17:06:22 +00:00 · 2026-05-30 11:45:16 -04:00 · 2026-05-30 11:09:06 -04:00 · 2026-05-30 11:09:06 -04:00
32 changed files with 2047 additions and 533 deletions
@@ -6,6 +6,14 @@ Changed `wf10_batch.robot` to be less likely to create files, and
 `plan_generation_graph.robot` to give more test answers.

 ## [Unreleased]
+- **fix(cli): add Read-Only and Writes columns to tool list output** (#1476): Rewrote
+  `list_tools()` in `src/cleveragents/cli/commands/tool.py` to render exactly the 5
+  spec-required columns (Name, Type, Source, Read-Only, Writes), removed the legacy
+  Description and Timeout columns, computes read-only/writes status from `capability`
+  metadata (rendering `✓`/`—`), and added a Summary panel showing Total, Tools,
+  Validations, Read-Only, Writes, and Namespaces counts. Adds Behave BDD tests in
+  `features/tool_cli.feature` verifying correct column names, capability rendering, and
+  Summary panel presence.
 - **Fix actor compiler to read LSP bindings from typed `lsp_binding` field** (#1488): Fixed
  `_extract_lsp_bindings()` in `actor/compiler.py` to read from `node.lsp_binding` (the typed
  `NodeLspBinding` Pydantic field on `NodeDefinition`) as the primary path, with
@@ -1,8 +1,13 @@
 """ASV benchmarks for skeleton compressor overhead.

-Measures the time to compress context fragments at various ratios
+Measures the time to compress context fragments at various budgets
 and fragment counts.  The benchmark covers the hot path that runs
 during subplan context inheritance.
+
+The caller is responsible for converting a skeleton_ratio to an
+absolute skeleton_budget before invoking compress().  These benchmarks
+use representative absolute token budgets derived from typical ratios
+applied to the fragment set sizes.
 """

 from __future__ import annotations
@@ -33,9 +38,9 @@ from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
 _SKEL_PROV = FragmentProvenance(resource_uri="bench-skeleton://default")


-def _build_fragments(count: int, tokens_each: int = 100) -> list[ContextFragment]:
+def _build_fragments(count: int, tokens_each: int = 100) -> tuple[ContextFragment, ...]:
    """Generate *count* fragments for benchmarking."""
-    return [
+    return tuple(
        ContextFragment(
            fragment_id=f"bench-{i:05d}",
            uko_node=f"bench-skeleton://file/{i}",
@@ -48,60 +53,60 @@ def _build_fragments(count: int, tokens_each: int = 100) -> list[ContextFragment
            ),
        )
        for i in range(count)
-    ]
+    )


 class SkeletonCompressorSmallSuite:
-    """Benchmark compression of a small fragment set (10 fragments)."""
+    """Benchmark compression of a small fragment set (10 fragments, 1000 tokens)."""

    def setup(self) -> None:
        self._svc = SkeletonCompressorService()
-        self._fragments = _build_fragments(10)
+        self._fragments = _build_fragments(10)  # 1000 total tokens

-    def time_compress_ratio_0(self) -> None:
-        """No compression (pass-through)."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.0)
+    def time_compress_budget_full(self) -> None:
+        """No compression (pass-through) — budget equals total tokens."""
+        self._svc.compress(self._fragments, 1000)

-    def time_compress_ratio_03(self) -> None:
-        """Default ratio compression."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.3)
+    def time_compress_budget_70pct(self) -> None:
+        """Default-ratio-equivalent compression (budget = 70% of total)."""
+        self._svc.compress(self._fragments, 700)

-    def time_compress_ratio_05(self) -> None:
-        """Moderate compression."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.5)
+    def time_compress_budget_50pct(self) -> None:
+        """Moderate compression (budget = 50% of total)."""
+        self._svc.compress(self._fragments, 500)

-    def time_compress_ratio_1(self) -> None:
-        """Maximum compression."""
-        self._svc.compress(self._fragments, skeleton_ratio=1.0)
+    def time_compress_budget_zero(self) -> None:
+        """Maximum compression — empty result."""
+        self._svc.compress(self._fragments, 0)


 class SkeletonCompressorMediumSuite:
-    """Benchmark compression of a medium fragment set (100 fragments)."""
+    """Benchmark compression of a medium fragment set (100 fragments, 10000 tokens)."""

    def setup(self) -> None:
        self._svc = SkeletonCompressorService()
-        self._fragments = _build_fragments(100)
+        self._fragments = _build_fragments(100)  # 10000 total tokens

-    def time_compress_ratio_03(self) -> None:
-        """Default ratio compression."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.3)
+    def time_compress_budget_70pct(self) -> None:
+        """Default-ratio-equivalent compression."""
+        self._svc.compress(self._fragments, 7000)

-    def time_compress_ratio_05(self) -> None:
+    def time_compress_budget_50pct(self) -> None:
        """Moderate compression."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.5)
+        self._svc.compress(self._fragments, 5000)


 class SkeletonCompressorLargeSuite:
-    """Benchmark compression of a large fragment set (1000 fragments)."""
+    """Benchmark compression of a large fragment set (1000 fragments, 100000 tokens)."""

    def setup(self) -> None:
        self._svc = SkeletonCompressorService()
-        self._fragments = _build_fragments(1000)
+        self._fragments = _build_fragments(1000)  # 100000 total tokens

-    def time_compress_ratio_05(self) -> None:
+    def time_compress_budget_50pct(self) -> None:
        """Moderate compression over 1000 fragments."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.5)
+        self._svc.compress(self._fragments, 50000)

-    def time_compress_ratio_08(self) -> None:
+    def time_compress_budget_20pct(self) -> None:
        """Heavy compression over 1000 fragments."""
-        self._svc.compress(self._fragments, skeleton_ratio=0.8)
+        self._svc.compress(self._fragments, 20000)
@@ -88,8 +88,8 @@ via the `getExtendedAgentCard` operation.

 ## Transport Modes

-| Mode   | Transport              | Class / SDK Component              | Behaviour                                              |
-|--------|------------------------|------------------------------------|--------------------------------------------------------|
+| Mode   | Transport              | Class / SDK Component              | Behaviour                                                          |
+|--------|------------------------|------------------------------------|--------------------------------------------------------------------|
 | Local  | A2A JSON-RPC over stdio | `A2aLocalFacade` + JSON-RPC stdio  | Agent as subprocess; extensions resolved in-process |
 | Server | A2A JSON-RPC over HTTP  | A2A SDK HTTP transport             | JSON-RPC 2.0 to CleverAgents server               |

@@ -165,7 +165,7 @@ Pattern applies to all entity types (`actor`, `skill`, `tool`, `validation`,
 ### Context Operations

 | Method | Service Method | Required Params |
-|--------|---------------|-----------------|
+|--------|----------------|-----------------|
 | `_cleveragents/context/show` | `ContextService.show()` | `project_name` |
 | `_cleveragents/context/inspect` | `ContextService.inspect()` | `project_name` |
 | `_cleveragents/context/simulate` | `ContextService.simulate()` | `project_name`, simulation params |
@@ -174,7 +174,7 @@ Pattern applies to all entity types (`actor`, `skill`, `tool`, `validation`,
 ### Sync Operations

 | Method | Service Method | Required Params |
-|--------|---------------|-----------------|
+|--------|----------------|-----------------|
 | `_cleveragents/sync/pull` | `SyncService.pull()` | `namespace` |
 | `_cleveragents/sync/push` | `SyncService.push()` | `namespace`, `entities` |
 | `_cleveragents/sync/status` | `SyncService.status()` | `namespace` (optional) |
@@ -182,7 +182,7 @@ Pattern applies to all entity types (`actor`, `skill`, `tool`, `validation`,
 ### Namespace Operations

 | Method | Service Method | Required Params |
-|--------|---------------|-----------------|
+|--------|----------------|-----------------|
 | `_cleveragents/namespace/list` | `NamespaceService.list()` | -- |
 | `_cleveragents/namespace/show` | `NamespaceService.show()` | `namespace` |
 | `_cleveragents/namespace/members` | `NamespaceService.members()` | `namespace` |
@@ -198,13 +198,20 @@ Pattern applies to all entity types (`actor`, `skill`, `tool`, `validation`,

 ## Streaming Events

-A2A streaming uses SSE via the `message/stream` operation, delivering typed
-events as the task progresses:
+A2A streaming uses SSE via the `message/stream` operation (client → server
+request), delivering typed events as the task progresses. The SSE data
+payloads are JSON-RPC 2.0 **notifications** (no `id` field):

-| Event Type | Emitted When |
-|-----------|-------------|
-| `TaskStatusUpdateEvent` | Task state changes (submitted -> working -> completed, etc.) |
-| `TaskArtifactUpdateEvent` | Agent produces output (response chunks, plan entries, tool results) |
+| Event Type | JSON-RPC Method | Emitted When |
+|------------|-----------------|--------------|
+| `TaskStatusUpdateEvent` | `task/statusUpdate` | Task state changes (submitted -> working -> completed, etc.) |
+| `TaskArtifactUpdateEvent` | `task/artifactUpdate` | Agent produces output (response chunks, plan entries, tool results) |
+
+> **Note:** `message/stream` is the **request** method (client → server) that
+> initiates streaming. The SSE data payloads use `task/statusUpdate` or
+> `task/artifactUpdate` as the notification method name — not `message/stream`.
+> Non-spec fields (`event_id`, `event_type`, `timestamp`) are carried in SSE
+> envelope headers (`event:` and `id:` lines), not in the data payload.

 ### Task Lifecycle States

@@ -220,16 +227,30 @@ events as the task progresses:

 ### Example Streaming Event

+SSE data payload for a task status update (JSON-RPC 2.0 notification):
+
 ```json
 {
  "jsonrpc": "2.0",
-  "method": "message/stream",
+  "method": "task/statusUpdate",
  "params": {
-    "id": "task_01HXR...",
-    "status": { "state": "working" },
-    "artifacts": [{
+    "taskId": "task_01HXR...",
+    "status": { "state": "working" }
+  }
+}
+```
+
+SSE data payload for an artifact update:
+
+```json
+{
+  "jsonrpc": "2.0",
+  "method": "task/artifactUpdate",
+  "params": {
+    "taskId": "task_01HXR...",
+    "artifact": {
      "parts": [{ "text": "I'll start by extracting..." }]
-    }]
+    }
  }
 }
 ```
@@ -277,9 +298,9 @@ response = facade.dispatch(A2aRequest(

 ### Constructor

-| Parameter  | Type                       | Default | Description                   |
-|------------|----------------------------|---------|-------------------------------|
-| `services` | `dict[str, Any] \| None`   | `None`  | Named services for routing    |
+| Parameter  | Type                             | Default | Description                    |
+|------------|----------------------------------|---------|--------------------------------|
+| `services` | `dict[str, Any] \| None`   | `None`  | Named services for routing     |

 ### Methods

@@ -299,15 +320,15 @@ registered later with `register_service()`.

 ### Service Keys

-| Key                          | Type                        | Wired Methods                       |
-|------------------------------|-----------------------------|-------------------------------------|
-| `session_service`            | `SessionWorkflow`           | Standard message operations         |
-| `plan_lifecycle_service`     | `PlanLifecycleService`      | `_cleveragents/plan/*`              |
-| `tool_registry`              | `ToolRegistry`              | `_cleveragents/registry/tool/*`     |
-| `resource_registry_service`  | `ResourceRegistryService`   | `_cleveragents/registry/resource/*` |
-| `sync_service`               | `SyncService`               | `_cleveragents/sync/*`              |
-| `namespace_service`          | `NamespaceService`          | `_cleveragents/namespace/*`         |
-| `context_service`            | `ContextService`            | `_cleveragents/context/*`           |
+| Key                              | Type                             | Wired Methods                              |
+|----------------------------------|----------------------------------|--------------------------------------------|
+| `session_service`                | `SessionWorkflow`                | Standard message operations                |
+| `plan_lifecycle_service`         | `PlanLifecycleService`           | `_cleveragents/plan/*`                     |
+| `tool_registry`                  | `ToolRegistry`                   | `_cleveragents/registry/tool/*`            |
+| `resource_registry_service`      | `ResourceRegistryService`        | `_cleveragents/registry/resource/*`        |
+| `sync_service`                   | `SyncService`                    | `_cleveragents/sync/*`                     |
+| `namespace_service`              | `NamespaceService`               | `_cleveragents/namespace/*`                |
+| `context_service`                | `ContextService`                 | `_cleveragents/context/*`                  |

 ---

@@ -402,8 +423,8 @@ CleverAgentsError
        └── A2aOperationNotFoundError
 ```

-| Exception                   | When Raised                                        |
-|-----------------------------|----------------------------------------------------|
-| `A2aNotAvailableError`      | Server-mode operation attempted without connection |
-| `A2aVersionMismatchError`   | Unsupported A2A version                            |
-| `A2aOperationNotFoundError` | Unknown method dispatched                          |
+| Exception                    | When Raised                                                        |
+|------------------------------|--------------------------------------------------------------------|
+| `A2aNotAvailableError`       | Server-mode operation attempted without connection |
+| `A2aVersionMismatchError`    | Unsupported A2A version                                            |
+| `A2aOperationNotFoundError`  | Unknown method dispatched                                          |
@@ -92,7 +92,6 @@ Feature: CLI output formats parity
    Then json_keys and yaml_keys should match

  # Direct formatting module tests
-  @tdd_issue @tdd_issue_4364 @tdd_expected_fail
  Scenario: Format output handles all format types for dict
    When I call format_output with a dict and format json
    Then the format result should be valid JSON dict
@@ -102,13 +101,33 @@ Feature: CLI output formats parity
    Then the format result should contain plain key-value pairs
    When I call format_output with a dict and format table
    Then the format result should contain table output
+
+  # fix(cli): format_output() rich format must produce styled output, not JSON
+  Scenario: Format output rich format produces styled terminal output not JSON
    When I call format_output with a dict and format rich
-    Then the format result should be valid JSON dict
+    Then the format result should contain rich styled output
+    And the format result should not be valid JSON
+
+  # fix(cli): format_output() color format must produce ANSI-colored output not plain text
+  Scenario: Format output color format produces ANSI-colored output not plain text
+    When I call format_output with a dict and format color
+    Then the format result should contain color styled output
+    And the format result should not be plain text only

  Scenario: Format output handles list data
    When I call format_output with a list and format plain
    Then the format result should contain separator lines

+  Scenario: Format output rich format handles list data
+    When I call format_output with a list and format rich
+    Then the format result should contain rich styled list output
+    And the format result should not be valid JSON
+
+  Scenario: Format output color format handles list data
+    When I call format_output with a list and format color
+    Then the format result should contain color styled list output
+    And the format result should not be plain text only
+
  Scenario: Serialize value handles enums and nested dicts
    When I call serialize_value with enum and nested data
    Then the serialized result should have string enum values
@@ -0,0 +1,277 @@
+@container-handler
+Feature: Container infrastructure resource handler module
+  Verifies that the container handler module exists and implements all
+  five handler classes referenced in _resource_registry_container.py:
+  ContainerRuntimeHandler, ContainerImageHandler, ContainerChildHandler,
+  ContainerVolumeHandler, ContainerNetworkHandler.
+
+  All handlers must conform to the ResourceHandler protocol and be
+  importable from cleveragents.resource.handlers.container.
+
+  # ── Module importability ─────────────────────────────────
+
+  Scenario: Container handler module is importable
+    When I import the container handler module
+    Then the container handler module import should succeed without errors
+
+  Scenario Outline: Handler class is importable from container module
+    When I import "<class_name>" from the container handler module
+    Then the class should be importable
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  # ── Protocol conformance ─────────────────────────────────
+
+  Scenario Outline: Handler class conforms to ResourceHandler protocol
+    Given a "<class_name>" instance
+    Then the instance should satisfy the ResourceHandler protocol
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  # ── Handler resolution via resolver ─────────────────────
+
+  Scenario Outline: Handler resolves via resolve_handler for container types
+    When I resolve the handler "<handler_ref>"
+    Then the handler should resolve without HandlerResolutionError
+
+    Examples:
+      | handler_ref                                                              |
+      | cleveragents.resource.handlers.container:ContainerRuntimeHandler         |
+      | cleveragents.resource.handlers.container:ContainerImageHandler           |
+      | cleveragents.resource.handlers.container:ContainerChildHandler           |
+      | cleveragents.resource.handlers.container:ContainerVolumeHandler          |
+      | cleveragents.resource.handlers.container:ContainerNetworkHandler         |
+
+  # ── ContainerRuntimeHandler ──────────────────────────────
+
+  Scenario: ContainerRuntimeHandler has correct type label
+    Given a "ContainerRuntimeHandler" instance
+    Then the handler type_label should be "container-runtime"
+
+  Scenario: ContainerRuntimeHandler resolve raises NotImplementedError
+    Given a "ContainerRuntimeHandler" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call resolve on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerRuntimeHandler read raises NotImplementedError
+    Given a "ContainerRuntimeHandler" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call read on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerRuntimeHandler content_hash returns EMPTY_CONTENT_HASH for no location
+    Given a "ContainerRuntimeHandler" instance
+    And a container resource of type "container-runtime" with no location
+    When I call content_hash on the container handler
+    Then the content_hash result should be the EMPTY_CONTENT_HASH sentinel
+
+  Scenario: ContainerRuntimeHandler content_hash returns identity hash for location
+    Given a "ContainerRuntimeHandler" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call content_hash on the container handler
+    Then the content_hash result should be a non-empty hex string
+
+  # ── ContainerImageHandler ────────────────────────────────
+
+  Scenario: ContainerImageHandler has correct type label
+    Given a "ContainerImageHandler" instance
+    Then the handler type_label should be "container-image"
+
+  Scenario: ContainerImageHandler resolve raises NotImplementedError
+    Given a "ContainerImageHandler" instance
+    And a container resource of type "container-image" with location "ubuntu:22.04"
+    When I call resolve on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerImageHandler content_hash returns identity hash for image ref
+    Given a "ContainerImageHandler" instance
+    And a container resource of type "container-image" with location "ubuntu:22.04"
+    When I call content_hash on the container handler
+    Then the content_hash result should be a non-empty hex string
+
+  # ── ContainerChildHandler ────────────────────────────────
+
+  Scenario: ContainerChildHandler has correct type label
+    Given a "ContainerChildHandler" instance
+    Then the handler type_label should be "container-child"
+
+  Scenario: ContainerChildHandler resolve raises NotImplementedError
+    Given a "ContainerChildHandler" instance
+    And a container resource of type "container-mount" with location "/mnt/data"
+    When I call resolve on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerChildHandler read raises NotImplementedError
+    Given a "ContainerChildHandler" instance
+    And a container resource of type "container-mount" with location "/mnt/data"
+    When I call read on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  # ── ContainerVolumeHandler ───────────────────────────────
+
+  Scenario: ContainerVolumeHandler has correct type label
+    Given a "ContainerVolumeHandler" instance
+    Then the handler type_label should be "container-volume"
+
+  Scenario: ContainerVolumeHandler resolve raises NotImplementedError
+    Given a "ContainerVolumeHandler" instance
+    And a container resource of type "container-volume" with location "my-volume"
+    When I call resolve on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerVolumeHandler content_hash returns identity hash for volume name
+    Given a "ContainerVolumeHandler" instance
+    And a container resource of type "container-volume" with location "my-volume"
+    When I call content_hash on the container handler
+    Then the content_hash result should be a non-empty hex string
+
+  # ── ContainerNetworkHandler ──────────────────────────────
+
+  Scenario: ContainerNetworkHandler has correct type label
+    Given a "ContainerNetworkHandler" instance
+    Then the handler type_label should be "container-network"
+
+  Scenario: ContainerNetworkHandler resolve raises NotImplementedError
+    Given a "ContainerNetworkHandler" instance
+    And a container resource of type "container-network" with location "bridge"
+    When I call resolve on the container handler
+    Then a container handler NotImplementedError should be raised
+
+  Scenario: ContainerNetworkHandler content_hash returns identity hash for network name
+    Given a "ContainerNetworkHandler" instance
+    And a container resource of type "container-network" with location "bridge"
+    When I call content_hash on the container handler
+    Then the content_hash result should be a non-empty hex string
+
+  # ── CRUD stubs ───────────────────────────────────────────
+
+  Scenario Outline: Handler write raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call write on the container handler with data b"test"
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  Scenario Outline: Handler delete raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call delete on the container handler
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  Scenario Outline: Handler list_children raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call list_children on the container handler
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  Scenario Outline: Handler diff raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call diff on the container handler with other_location "/other"
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  Scenario Outline: Handler discover_children raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call discover_children on the container handler
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  # ── Lifecycle stubs ──────────────────────────────────────
+
+  Scenario Outline: Handler create_checkpoint raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call create_checkpoint on the container handler
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  Scenario Outline: Handler rollback_to raises NotImplementedError
+    Given a "<class_name>" instance
+    And a container resource of type "container-runtime" with location "/var/run/docker.sock"
+    When I call rollback_to on the container handler
+    Then a container handler NotImplementedError should be raised
+
+    Examples:
+      | class_name                |
+      | ContainerRuntimeHandler   |
+      | ContainerImageHandler     |
+      | ContainerChildHandler     |
+      | ContainerVolumeHandler    |
+      | ContainerNetworkHandler   |
+
+  # ── Registry integration ─────────────────────────────────
+
+  Scenario: container-runtime handler resolves without error after module creation
+    When I resolve the handler for resource type "container-runtime"
+    Then the handler should resolve without HandlerResolutionError
+
+  Scenario: container-image handler resolves without error after module creation
+    When I resolve the handler for resource type "container-image"
+    Then the handler should resolve without HandlerResolutionError
+
+  Scenario: container-volume handler resolves without error after module creation
+    When I resolve the handler for resource type "container-volume"
+    Then the handler should resolve without HandlerResolutionError
+
+  Scenario: container-network handler resolves without error after module creation
+    When I resolve the handler for resource type "container-network"
+    Then the handler should resolve without HandlerResolutionError
@@ -204,9 +204,17 @@ Feature: Legacy Data Migrator Coverage
    Given I have a legacy project with read-only JSON files
    When I run the legacy data migration
    Then the migration should handle backup errors gracefully
+
  Scenario: get_all_for_project is called only once for multiple plans
    Given I have a legacy project with multiple plans to migrate efficiently
    When I run the legacy data migration with query tracking
    Then get_all_for_project should be called exactly once
    And the migration should return true
    And all plans should be migrated correctly
+
+  Scenario: Map existing plan IDs correctly during migration
+    Given I have a legacy project with plans already in database
+    When I run the legacy data migration
+    Then existing plans should not be duplicated
+    And the migration should return true
+    And the existing plan id should be mapped without type suppression
@@ -429,11 +429,6 @@ Feature: Services __init__ lazy-import coverage (round 3)
    Then svcov3 the attribute "PersistentSessionService" should be resolved

  # ---------- skeleton_compressor ----------
-  Scenario: svcov3 lazy-load CompressionResult
-    Given svcov3 a fresh services module
-    When svcov3 I access lazy attribute "CompressionResult"
-    Then svcov3 the attribute "CompressionResult" should be resolved
-
  Scenario: svcov3 lazy-load SkeletonCompressorService
    Given svcov3 a fresh services module
    When svcov3 I access lazy attribute "SkeletonCompressorService"
@@ -6,134 +6,100 @@ Feature: Skeleton compressor
  Background:
    Given a skeleton compressor service

-  # --- ratio validation -------------------------------------------------
+  # --- budget validation -------------------------------------------------

-  Scenario: Reject ratio below 0.0
+  Scenario: Reject negative skeleton_budget
    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio -0.1
-    Then the compressor should raise a ValueError for invalid ratio
+    When I compress with skeleton_budget -1
+    Then the compressor should raise a ValueError for invalid budget

-  Scenario: Reject ratio above 1.0
+  Scenario: Accept budget 0 returns empty
    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 1.5
-    Then the compressor should raise a ValueError for invalid ratio
+    When I compress with skeleton_budget 0
+    Then the result should contain zero fragments

-  Scenario: Accept ratio 0.0
+  Scenario: Accept large budget returns all fragments
    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 0.0
+    When I compress with skeleton_budget 1000
    Then all fragments should be returned unchanged

-  Scenario: Accept ratio 1.0
+  Scenario: Accept budget at boundary 500
    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 1.0
-    Then only the highest-relevance fragment should be returned
-
-  Scenario: Accept ratio at boundary 0.5
-    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 500
    Then compressed tokens should be at most 500

-  # --- default handling -------------------------------------------------
-
-  Scenario: Default ratio applied when None
-    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio not specified
-    Then the metadata ratio should equal the default 0.15
-
  # --- stable ordering --------------------------------------------------

  Scenario: Fragments with equal relevance are ordered by id
    Given three fragments with equal relevance 0.5
-    When I compress with skeleton_ratio 0.0
+    When I compress with skeleton_budget 10000
    Then fragments should be ordered by fragment_id ascending

  Scenario: Fragments are ordered by relevance descending
    Given fragments with relevances 0.9, 0.3, and 0.7
-    When I compress with skeleton_ratio 0.0
+    When I compress with skeleton_budget 10000
    Then the first fragment should have relevance 0.9
    And the last fragment should have relevance 0.3

-  # --- metadata ----------------------------------------------------------
-
-  Scenario: Metadata records correct token counts
-    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 0.5
-    Then metadata original_tokens should be 1000
-    And metadata compressed_tokens should be at most 500
-
-  Scenario: Metadata records source decision IDs
-    Given fragments with known decision IDs
-    When I compress with skeleton_ratio 0.0
-    Then metadata should contain all source decision IDs
-
-  Scenario: Metadata ratio matches input
-    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 0.7
-    Then metadata ratio should be 0.7
-
  # --- edge cases -------------------------------------------------------

  Scenario: Empty fragment list compresses to empty
    Given an empty fragment list
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 500
    Then the result should contain zero fragments
-    And metadata original_tokens should be 0
-    And metadata compressed_tokens should equal 0

-  Scenario: Single fragment at ratio 0.5
+  Scenario: Single fragment fits within budget
    Given a single fragment with 100 tokens
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 200
+    Then the result should contain one fragment
+
+  Scenario: Single fragment kept even when it exceeds budget
+    Given a single fragment with 100 tokens
+    When I compress with skeleton_budget 50
    Then the result should contain one fragment

  # --- argument validation ----------------------------------------------

-  Scenario: Reject non-list fragments argument
-    When I compress with a non-list fragments argument
+  Scenario: Reject non-tuple fragments argument
+    When I compress with a non-tuple fragments argument
    Then the compressor should raise a TypeError

  Scenario: Reject fragment with negative token count
    Given a fragment with negative token count
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 500
    Then the compressor should raise a ValueError for invalid fragment

  Scenario: Reject fragment with empty id
    Given a fragment with empty fragment_id
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 500
    Then the compressor should raise a ValueError for invalid fragment

  Scenario: Reject fragment with relevance out of range
    Given a fragment with relevance 1.5
-    When I compress with skeleton_ratio 0.5
+    When I compress with skeleton_budget 500
    Then the compressor should raise a ValueError for invalid fragment

-  Scenario: Reject non-ContextFragment item in list
-    When I compress with a list containing a non-fragment item
+  Scenario: Reject non-ContextFragment item in tuple
+    When I compress with a tuple containing a non-fragment item
    Then the compressor should raise a TypeError for invalid item

-  Scenario: Reject non-numeric skeleton_ratio
+  Scenario: Reject non-integer skeleton_budget
    Given context fragments with total tokens 1000
-    When I compress with a non-numeric skeleton_ratio
-    Then the compressor should raise a TypeError for invalid ratio type
+    When I compress with a non-integer skeleton_budget
+    Then the compressor should raise a TypeError for invalid budget type

  Scenario: Reject skeleton metadata with compressed exceeding original
    When I create skeleton metadata with compressed exceeding original
    Then a validation error should be raised for compressed exceeding original

-  # --- compression summary (original vs compressed) ----------------------
-
-  Scenario: Compression summary stored in plan metadata
-    Given context fragments with total tokens 1000
-    When I compress with skeleton_ratio 0.6
-    Then compressed_tokens should be less than original_tokens
-
-  # --- skeleton_ratio integration with plan model ------------------------
+  # --- skeleton_budget integration with plan model -----------------------

  Scenario: Plan model accepts skeleton_metadata
    Given a skeleton metadata with ratio 0.5 and 1000 original tokens and 500 compressed
    When I attach skeleton_metadata to a plan
    Then the plan should expose skeleton metadata in cli dict

-  # --- _validate_fragments edge-cases (lines 153-163) --------------------
+  # --- _validate_fragments edge-cases ------------------------------------

  Scenario: Reject fragment with negative token_count
    Given a context fragment constructed with negative token_count
@@ -149,3 +115,9 @@ Feature: Skeleton compressor
    Given a context fragment constructed with empty fragment_id
    When I validate the invalid fragments
    Then the compressor should raise a ValueError mentioning "fragment_id must be non-empty"
+
+  # --- structural subtype assertion -------------------------------------
+
+  Scenario: SkeletonCompressorService satisfies SkeletonCompressor protocol
+    When I check that SkeletonCompressorService satisfies the SkeletonCompressor protocol
+    Then the structural subtype assertion should pass
@@ -310,3 +310,63 @@ Feature: Skill YAML schema validation
    And the skill config model_dump should contain key "mcp_servers"
    And the skill config model_dump should contain key "agent_skill_folders"

+
+  # ────────────────────────────────────────────────────────────
+  # skill: wrapper key scenarios
+  # ────────────────────────────────────────────────────────────
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: Load spec-compliant YAML with skill: wrapper key
+    Given a spec-compliant skill YAML with skill: wrapper key
+    When I validate the skill schema
+    Then the skill schema validation should succeed
+    And the skill config name should be "local/wrapped-skill"
+    And the skill config should have 1 tools
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: Load spec-compliant YAML with cleveragents: header and skill: wrapper
+    Given a spec-compliant skill YAML with cleveragents: header and skill: wrapper
+    When I validate the skill schema
+    Then the skill schema validation should succeed
+    And the skill config name should be "local/wrapped-with-meta"
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: skill: wrapper key with None value raises ValueError
+    Given a skill YAML with skill: wrapper key with None value
+    When I validate the skill schema expecting failure
+    Then the skill schema validation should fail
+    And the skill schema error should mention "empty"
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: skill: wrapper key with non-dict value raises ValueError
+    Given a skill YAML with skill: wrapper key and string value
+    When I validate the skill schema expecting failure
+    Then the skill schema validation should fail
+    And the skill schema error should mention "mapping"
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: skill: wrapper key with list value raises ValueError
+    Given a skill YAML with skill: wrapper key and list value
+    When I validate the skill schema expecting failure
+    Then the skill schema validation should fail
+    And the skill schema error should mention "mapping"
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: Flat YAML without wrapper key still works (backward compatibility)
+    Given a skill YAML string with only the name field
+    When I validate the skill schema
+    Then the skill schema validation should succeed
+    And the skill config name should be "local/empty-skill"
+
+  @tdd_issue
+  @tdd_issue_1472
+  Scenario: cleveragents: header alone (flat format with metadata, no skill: wrapper)
+    Given a skill YAML with cleveragents: header and flat format
+    When I validate the skill schema
+    Then the skill schema validation should succeed
+    And the skill config name should be "local/flat-with-meta"
@@ -0,0 +1,151 @@
+"""Step definitions for CLI output format validation (rich / color formats)."""
+
+from __future__ import annotations
+
+import json
+
+import yaml
+from behave import then
+from behave.runner import Context
+
+
+@then("the format result should be valid JSON dict")
+def step_format_result_json_dict(context: Context) -> None:
+    parsed = json.loads(context.format_result)
+    assert isinstance(parsed, dict)
+    # Verify spec-required envelope fields are present
+    for field in ("command", "status", "exit_code", "data", "timing", "messages"):
+        assert field in parsed, f"Envelope field '{field}' missing from JSON result"
+    # Verify data field contains the original dict
+    assert isinstance(parsed["data"], dict)
+
+
+@then("the format result should be valid YAML dict")
+def step_format_result_yaml_dict(context: Context) -> None:
+    parsed = yaml.safe_load(context.format_result)
+    assert isinstance(parsed, dict)
+    # Verify spec-required envelope fields are present
+    for field in ("command", "status", "exit_code", "data", "timing", "messages"):
+        assert field in parsed, f"Envelope field '{field}' missing from YAML result"
+    # Verify data field contains the original dict
+    assert isinstance(parsed["data"], dict)
+
+
+@then("the format result should contain plain key-value pairs")
+def step_format_result_plain(context: Context) -> None:
+    assert "name: test" in context.format_result
+    assert "nested:" in context.format_result
+    assert "items:" in context.format_result
+
+
+@then("the format result should contain table output")
+def step_format_result_table(context: Context) -> None:
+    assert "name" in context.format_result
+    assert "count" in context.format_result
+
+
+@then("the format result should contain separator lines")
+def step_format_result_separator(context: Context) -> None:
+    assert "---" in context.format_result
+
+
+@then("the serialized result should have string enum values")
+def step_serialized_enum_values(context: Context) -> None:
+    assert context.serialized["state"] == "available"
+    assert isinstance(context.serialized["nested"], dict)
+    assert context.serialized["items"][0] == "archived"
+    assert isinstance(context.serialized["ts"], str)
+
+
+@then("the format result should contain rich styled output")
+def step_format_result_rich_styled(context: Context) -> None:
+    """Assert that rich format produces styled output (ANSI codes or key-value lines).
+
+    The rich format must NOT produce raw JSON — it must produce styled terminal
+    output via the RichMaterializer (ANSI escape codes or structured key-value
+    lines with panel headers).
+    """
+    result = context.format_result
+    # Rich output must contain the data keys
+    assert "name" in result, f"Rich output missing 'name': {result!r}"
+    assert "count" in result, f"Rich output missing 'count': {result!r}"
+    # Rich output must contain styled panel header or ANSI codes
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_panel_header = "Output" in result
+    assert has_ansi or has_panel_header, (
+        f"Rich output has no ANSI codes or panel header — got: {result!r}"
+    )
+
+
+@then("the format result should not be valid JSON")
+def step_format_result_not_json(context: Context) -> None:
+    """Assert that the format result is NOT valid JSON (i.e. not the JSON fallback)."""
+    try:
+        json.loads(context.format_result)
+        raise AssertionError(
+            f"format_output() with 'rich' returned valid JSON — "
+            f"expected styled terminal output, got: {context.format_result!r}"
+        )
+    except (json.JSONDecodeError, ValueError):
+        pass  # Expected: rich output is not JSON
+
+
+@then("the format result should contain color styled output")
+def step_format_result_color_styled(context: Context) -> None:
+    """Assert that color format produces ANSI-colored output (not plain text).
+
+    The color format must NOT produce plain text — it must produce ANSI-colored
+    terminal output via the ColorMaterializer.
+    """
+    result = context.format_result
+    # Color output must contain the data keys
+    assert "name" in result, f"Color output missing 'name': {result!r}"
+    assert "count" in result, f"Color output missing 'count': {result!r}"
+    # Color output must contain ANSI escape codes or styled panel header
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_panel_header = "Output" in result
+    assert has_ansi or has_panel_header, (
+        f"Color output has no ANSI codes or panel header — got: {result!r}"
+    )
+
+
+@then("the format result should not be plain text only")
+def step_format_result_not_plain_text(context: Context) -> None:
+    """Assert that the color format result is NOT plain text (i.e. not the plain fallback).
+
+    Plain text would look like 'name: test\\ncount: 42' with no ANSI codes.
+    Color output must have ANSI escape codes or styled panel headers.
+    """
+    result = context.format_result
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_panel_header = "Output" in result
+    # If it's purely plain text (no ANSI, no panel header), that's the bug
+    is_plain_only = not has_ansi and not has_panel_header
+    assert not is_plain_only, (
+        f"format_output() with 'color' returned plain text — "
+        f"expected ANSI-colored output, got: {result!r}"
+    )
+
+
+@then("the format result should contain rich styled list output")
+def step_format_result_rich_list(context: Context) -> None:
+    """Assert rich format produces styled output for list data (panel per item)."""
+    result = context.format_result
+    assert "name" in result, f"Rich list output missing 'name': {result!r}"
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_item_panel = "Item 1" in result or "Item 2" in result
+    assert has_ansi or has_item_panel, (
+        f"Rich list output has no ANSI codes or item panels — got: {result!r}"
+    )
+
+
+@then("the format result should contain color styled list output")
+def step_format_result_color_list(context: Context) -> None:
+    """Assert color format produces styled output for list data (panel per item)."""
+    result = context.format_result
+    assert "name" in result, f"Color list output missing 'name': {result!r}"
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_item_panel = "Item 1" in result or "Item 2" in result
+    assert has_ansi or has_item_panel, (
+        f"Color list output has no ANSI codes or item panels — got: {result!r}"
+    )
@@ -441,6 +441,12 @@ def step_call_format_rich(context: Context) -> None:
    context.format_result = _capture_format_output(data, "rich")


+@when("I call format_output with a dict and format color")
+def step_call_format_color(context: Context) -> None:
+    data = {"name": "test", "count": 42}
+    context.format_result = _capture_format_output(data, "color")
+
+
@when("I call format_output with a list and format plain")
 def step_call_format_list_plain(context: Context) -> None:
    data = [
@@ -450,6 +456,24 @@ def step_call_format_list_plain(context: Context) -> None:
    context.format_result = _capture_format_output(data, "plain")


+@when("I call format_output with a list and format rich")
+def step_call_format_list_rich(context: Context) -> None:
+    data = [
+        {"name": "a", "count": 1},
+        {"name": "b", "count": 2},
+    ]
+    context.format_result = _capture_format_output(data, "rich")
+
+
+@when("I call format_output with a list and format color")
+def step_call_format_list_color(context: Context) -> None:
+    data = [
+        {"name": "a", "count": 1},
+        {"name": "b", "count": 2},
+    ]
+    context.format_result = _capture_format_output(data, "color")
+
+
@when("I call serialize_value with enum and nested data")
 def step_call_serialize_with_enum(context: Context) -> None:
    context.serialized = _serialize_value(
@@ -460,54 +484,3 @@ def step_call_serialize_with_enum(context: Context) -> None:
            "ts": datetime(2025, 6, 1, 12, 0),
        }
    )
-
-
-# ------- Additional Then steps -------
-
-
-@then("the format result should be valid JSON dict")
-def step_format_result_json_dict(context: Context) -> None:
-    parsed = json.loads(context.format_result)
-    assert isinstance(parsed, dict)
-    # Verify spec-required envelope fields are present
-    for field in ("command", "status", "exit_code", "data", "timing", "messages"):
-        assert field in parsed, f"Envelope field '{field}' missing from JSON result"
-    # Verify data field contains the original dict
-    assert isinstance(parsed["data"], dict)
-
-
-@then("the format result should be valid YAML dict")
-def step_format_result_yaml_dict(context: Context) -> None:
-    parsed = yaml.safe_load(context.format_result)
-    assert isinstance(parsed, dict)
-    # Verify spec-required envelope fields are present
-    for field in ("command", "status", "exit_code", "data", "timing", "messages"):
-        assert field in parsed, f"Envelope field '{field}' missing from YAML result"
-    # Verify data field contains the original dict
-    assert isinstance(parsed["data"], dict)
-
-
-@then("the format result should contain plain key-value pairs")
-def step_format_result_plain(context: Context) -> None:
-    assert "name: test" in context.format_result
-    assert "nested:" in context.format_result
-    assert "items:" in context.format_result
-
-
-@then("the format result should contain table output")
-def step_format_result_table(context: Context) -> None:
-    assert "name" in context.format_result
-    assert "count" in context.format_result
-
-
-@then("the format result should contain separator lines")
-def step_format_result_separator(context: Context) -> None:
-    assert "---" in context.format_result
-
-
-@then("the serialized result should have string enum values")
-def step_serialized_enum_values(context: Context) -> None:
-    assert context.serialized["state"] == "available"
-    assert isinstance(context.serialized["nested"], dict)
-    assert context.serialized["items"][0] == "archived"
-    assert isinstance(context.serialized["ts"], str)
@@ -0,0 +1,389 @@
+"""Step definitions for container_handler.feature.
+
+Tests for the container infrastructure resource handler module
+(cleveragents.resource.handlers.container).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from behave import given, then, when  # type: ignore[attr-defined]
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.resource.handlers._base import EMPTY_CONTENT_HASH
+from cleveragents.resource.handlers.protocol import ResourceHandler
+
+# ── Handler class registry ───────────────────────────────────
+
+
+def _get_handler_class(class_name: str) -> type:
+    """Import and return a handler class from the container module."""
+    from cleveragents.resource.handlers import container as _container_mod
+
+    cls = getattr(_container_mod, class_name, None)
+    if cls is None:
+        raise AssertionError(
+            f"Class '{class_name}' not found in "
+            "cleveragents.resource.handlers.container"
+        )
+    return cls
+
+
+def _make_container_resource(
+    resource_type: str,
+    location: str | None = None,
+) -> Resource:
+    """Create a minimal container resource for testing."""
+    return Resource(
+        resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
+        resource_type_name=resource_type,
+        classification=PhysVirt.PHYSICAL,
+        location=location,
+        properties={},
+        capabilities=ResourceCapabilities(
+            readable=True,
+            writable=False,
+            sandboxable=False,
+            checkpointable=False,
+        ),
+    )
+
+
+# ── When steps ───────────────────────────────────────────────
+
+
+@when("I import the container handler module")
+def step_import_container_module(context: Any) -> None:
+    """Attempt to import the container handler module."""
+    try:
+        import cleveragents.resource.handlers.container as _mod
+
+        context.import_error = None  # type: ignore[attr-defined]
+        context.container_module = _mod  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.import_error = exc  # type: ignore[attr-defined]
+
+
+@when('I import "{class_name}" from the container handler module')
+def step_import_class_from_container_module(context: Any, class_name: str) -> None:
+    """Attempt to import a specific class from the container handler module."""
+    try:
+        cls = _get_handler_class(class_name)
+        context.imported_class = cls  # type: ignore[attr-defined]
+        context.import_error = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.import_error = exc  # type: ignore[attr-defined]
+        context.imported_class = None  # type: ignore[attr-defined]
+
+
+@when('I resolve the handler "<handler_ref>"')
+def step_resolve_handler_ref_template(context: Any, handler_ref: str) -> None:
+    """Resolve a handler by reference string (template step)."""
+    step_resolve_handler_ref(context, handler_ref)
+
+
+@when('I resolve the handler "{handler_ref}"')
+def step_resolve_handler_ref(context: Any, handler_ref: str) -> None:
+    """Resolve a handler by reference string."""
+    from cleveragents.resource.handlers.resolver import (
+        HandlerResolutionError,
+        clear_handler_cache,
+        resolve_handler,
+    )
+
+    clear_handler_cache()
+    try:
+        context.resolved_handler = resolve_handler(handler_ref)  # type: ignore[attr-defined]
+        context.resolution_error = None  # type: ignore[attr-defined]
+    except HandlerResolutionError as exc:
+        context.resolution_error = exc  # type: ignore[attr-defined]
+        context.resolved_handler = None  # type: ignore[attr-defined]
+
+
+@when('I resolve the handler for resource type "{resource_type}"')
+def step_resolve_handler_for_type(context: Any, resource_type: str) -> None:
+    """Resolve a handler for a resource type via the registry."""
+    from cleveragents.application.services._resource_registry_data import BUILTIN_TYPES
+    from cleveragents.resource.handlers.resolver import (
+        HandlerResolutionError,
+        clear_handler_cache,
+        resolve_handler,
+    )
+
+    clear_handler_cache()
+    handler_ref = None
+    for type_def in BUILTIN_TYPES:
+        if type_def["name"] == resource_type:
+            handler_ref = type_def.get("handler")
+            break
+
+    if handler_ref is None:
+        context.resolution_error = AssertionError(  # type: ignore[attr-defined]
+            f"Resource type '{resource_type}' not found in BUILTIN_TYPES"
+        )
+        context.resolved_handler = None  # type: ignore[attr-defined]
+        return
+
+    try:
+        context.resolved_handler = resolve_handler(handler_ref)  # type: ignore[attr-defined]
+        context.resolution_error = None  # type: ignore[attr-defined]
+    except HandlerResolutionError as exc:
+        context.resolution_error = exc  # type: ignore[attr-defined]
+        context.resolved_handler = None  # type: ignore[attr-defined]
+
+
+@when("I call resolve on the container handler")
+def step_call_resolve(context: Any) -> None:
+    """Call resolve() on the handler and capture any exception."""
+    mock_sandbox = MagicMock()
+    try:
+        context.handler.resolve(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+            plan_id="test-plan-01",
+            slot_name="test-slot",
+            sandbox_manager=mock_sandbox,
+            access="read_only",
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call read on the container handler")
+def step_call_read(context: Any) -> None:
+    """Call read() on the handler and capture any exception."""
+    try:
+        context.handler.read(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when('I call write on the container handler with data b"test"')
+def step_call_write(context: Any) -> None:
+    """Call write() on the handler and capture any exception."""
+    try:
+        context.handler.write(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+            path="",
+            data=b"test",
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call delete on the container handler")
+def step_call_delete(context: Any) -> None:
+    """Call delete() on the handler and capture any exception."""
+    try:
+        context.handler.delete(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call list_children on the container handler")
+def step_call_list_children(context: Any) -> None:
+    """Call list_children() on the handler and capture any exception."""
+    try:
+        context.handler.list_children(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when('I call diff on the container handler with other_location "/other"')
+def step_call_diff(context: Any) -> None:
+    """Call diff() on the handler and capture any exception."""
+    try:
+        context.handler.diff(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+            other_location="/other",
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call discover_children on the container handler")
+def step_call_discover_children(context: Any) -> None:
+    """Call discover_children() on the handler and capture any exception."""
+    try:
+        context.handler.discover_children(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call create_checkpoint on the container handler")
+def step_call_create_checkpoint(context: Any) -> None:
+    """Call create_checkpoint() on the handler and capture any exception."""
+    mock_sandbox = MagicMock()
+    try:
+        context.handler.create_checkpoint(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+            plan_id="test-plan-01",
+            sandbox_manager=mock_sandbox,
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call rollback_to on the container handler")
+def step_call_rollback_to(context: Any) -> None:
+    """Call rollback_to() on the handler and capture any exception."""
+    mock_sandbox = MagicMock()
+    try:
+        context.handler.rollback_to(  # type: ignore[attr-defined]
+            resource=context.container_resource,  # type: ignore[attr-defined]
+            plan_id="test-plan-01",
+            checkpoint_id="ckpt-001",
+            sandbox_manager=mock_sandbox,
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+
+
+@when("I call content_hash on the container handler")
+def step_call_content_hash(context: Any) -> None:
+    """Call content_hash() on the handler and capture the result."""
+    try:
+        context.content_hash_result = context.handler.content_hash(  # type: ignore[attr-defined]
+            context.container_resource  # type: ignore[attr-defined]
+        )
+        context.raised_exception = None  # type: ignore[attr-defined]
+    except Exception as exc:
+        context.raised_exception = exc  # type: ignore[attr-defined]
+        context.content_hash_result = None  # type: ignore[attr-defined]
+
+
+# ── Given steps ──────────────────────────────────────────────
+
+
+@given('a "{class_name}" instance')
+def step_handler_instance(context: Any, class_name: str) -> None:
+    """Create a handler instance by class name."""
+    cls = _get_handler_class(class_name)
+    context.handler = cls()  # type: ignore[attr-defined]
+    context.handler_class_name = class_name  # type: ignore[attr-defined]
+
+
+@given('a container resource of type "{resource_type}" with location "{location}"')
+def step_container_resource_with_location(
+    context: Any, resource_type: str, location: str
+) -> None:
+    """Create a container resource with a specific location."""
+    context.container_resource = _make_container_resource(  # type: ignore[attr-defined]
+        resource_type=resource_type,
+        location=location,
+    )
+
+
+@given('a container resource of type "{resource_type}" with no location')
+def step_container_resource_no_location(context: Any, resource_type: str) -> None:
+    """Create a container resource with no location."""
+    context.container_resource = _make_container_resource(  # type: ignore[attr-defined]
+        resource_type=resource_type,
+        location=None,
+    )
+
+
+# ── Then steps ───────────────────────────────────────────────
+
+
+@then("the container handler module import should succeed without errors")
+def step_import_succeeded(context: Any) -> None:
+    """Assert that the import succeeded."""
+    assert context.import_error is None, (  # type: ignore[attr-defined]
+        f"Import failed with: {context.import_error}"  # type: ignore[attr-defined]
+    )
+
+
+@then("the class should be importable")
+def step_class_importable(context: Any) -> None:
+    """Assert that the class was imported successfully."""
+    assert context.import_error is None, (  # type: ignore[attr-defined]
+        f"Class import failed with: {context.import_error}"  # type: ignore[attr-defined]
+    )
+    assert context.imported_class is not None  # type: ignore[attr-defined]
+
+
+@then("the instance should satisfy the ResourceHandler protocol")
+def step_satisfies_protocol(context: Any) -> None:
+    """Assert that the handler instance satisfies ResourceHandler protocol."""
+    assert isinstance(context.handler, ResourceHandler), (  # type: ignore[attr-defined]
+        f"{context.handler_class_name} does not satisfy ResourceHandler protocol"  # type: ignore[attr-defined]
+    )
+
+
+@then("the handler should resolve without HandlerResolutionError")
+def step_handler_resolved(context: Any) -> None:
+    """Assert that the handler resolved without error."""
+    assert context.resolution_error is None, (  # type: ignore[attr-defined]
+        f"Handler resolution failed: {context.resolution_error}"  # type: ignore[attr-defined]
+    )
+    assert context.resolved_handler is not None  # type: ignore[attr-defined]
+
+
+@then('the handler type_label should be "{expected_label}"')
+def step_handler_type_label(context: Any, expected_label: str) -> None:
+    """Assert the handler's _type_label class variable."""
+    actual = getattr(context.handler, "_type_label", None)  # type: ignore[attr-defined]
+    assert actual == expected_label, (
+        f"Expected _type_label='{expected_label}', got '{actual}'"
+    )
+
+
+@then("a container handler NotImplementedError should be raised")
+def step_not_implemented_raised(context: Any) -> None:
+    """Assert that a NotImplementedError was raised."""
+    exc = context.raised_exception  # type: ignore[attr-defined]
+    assert exc is not None, "Expected NotImplementedError but no exception was raised"
+    assert isinstance(exc, NotImplementedError), (
+        f"Expected NotImplementedError but got {type(exc).__name__}: {exc}"
+    )
+
+
+@then("the content_hash result should be the EMPTY_CONTENT_HASH sentinel")
+def step_content_hash_is_empty(context: Any) -> None:
+    """Assert that content_hash returned the EMPTY_CONTENT_HASH sentinel."""
+    result = context.content_hash_result  # type: ignore[attr-defined]
+    assert result == EMPTY_CONTENT_HASH, (
+        f"Expected EMPTY_CONTENT_HASH but got: {result!r}"
+    )
+
+
+@then("the content_hash result should be a non-empty hex string")
+def step_content_hash_is_hex(context: Any) -> None:
+    """Assert that content_hash returned a non-empty hex string."""
+    result = context.content_hash_result  # type: ignore[attr-defined]
+    assert result is not None, "content_hash returned None"
+    assert isinstance(result, str), f"Expected str, got {type(result)}"
+    assert len(result) > 0, "content_hash returned empty string"
+    assert result != EMPTY_CONTENT_HASH, (
+        "content_hash returned EMPTY_CONTENT_HASH but expected identity hash"
+    )
+    # Verify it's a valid hex string
+    try:
+        int(result, 16)
+    except ValueError as exc:
+        raise AssertionError(
+            f"content_hash result is not a valid hex string: {result!r}"
+        ) from exc
@@ -1085,3 +1085,17 @@ def step_check_all_plans_migrated(context: Context) -> None:
        assert plan_names == expected_names, (
            f"Expected plans {expected_names}, got {plan_names}"
        )
+
+
+@then("the existing plan id should be mapped without type suppression")
+def step_check_existing_plan_id_mapped(context: Context) -> None:
+    """Check that existing plan id was mapped correctly via assert type narrowing."""
+    with context.unit_of_work.transaction() as ctx:
+        project = ctx.projects.get_by_name("existing_plans")
+        assert project is not None, "Project 'existing_plans' should exist in database"
+        plans = ctx.plans.get_all_for_project(project.id)
+        main_plans = [p for p in plans if p.name == "main"]
+        assert len(main_plans) == 1, "Should have exactly one 'main' plan"
+        assert main_plans[0].id is not None, (
+            "Existing plan id must be non-None (assert narrowing validates this)"
+        )
@@ -6,8 +6,8 @@ from behave import given, then, when
 from behave.runner import Context
 from pydantic import ValidationError

+from cleveragents.application.services.acms_service import SkeletonCompressor
 from cleveragents.application.services.skeleton_compressor import (
-    DEFAULT_SKELETON_RATIO,
    SkeletonCompressorService,
 )
 from cleveragents.domain.models.core.context_fragment import (
@@ -51,7 +51,7 @@ def _make_skel_fragment(
    )


-def _make_fragments(total_tokens: int, count: int = 4) -> list[ContextFragment]:
+def _make_fragments(total_tokens: int, count: int = 4) -> tuple[ContextFragment, ...]:
    """Create *count* fragments summing to *total_tokens*."""
    base = total_tokens // count
    remainder = total_tokens - base * count
@@ -68,7 +68,7 @@ def _make_fragments(total_tokens: int, count: int = 4) -> list[ContextFragment]:
                source_decision_id=f"01HX{'A' * 22}{i}" if i < 3 else None,
            )
        )
-    return frags
+    return tuple(frags)


 def _make_plan_id() -> str:
@@ -94,7 +94,7 @@ def step_fragments_total(context: Context, total: int) -> None:

@given("three fragments with equal relevance {rel:g}")
 def step_equal_relevance(context: Context, rel: float) -> None:
-    context.fragments = [
+    context.fragments = tuple(
        _make_skel_fragment(
            fragment_id=f"frag-{chr(ord('c') - i)}",
            content="x" * 50,
@@ -102,12 +102,12 @@ def step_equal_relevance(context: Context, rel: float) -> None:
            relevance_score=rel,
        )
        for i in range(3)
-    ]
+    )


@given("fragments with relevances 0.9, 0.3, and 0.7")
 def step_varied_relevances(context: Context) -> None:
-    context.fragments = [
+    context.fragments = (
        _make_skel_fragment(
            fragment_id="f-1", content="a", token_count=100, relevance_score=0.9
        ),
@@ -117,12 +117,12 @@ def step_varied_relevances(context: Context) -> None:
        _make_skel_fragment(
            fragment_id="f-3", content="c", token_count=100, relevance_score=0.7
        ),
-    ]
+    )


@given("fragments with known decision IDs")
 def step_known_ids(context: Context) -> None:
-    context.fragments = [
+    context.fragments = (
        _make_skel_fragment(
            fragment_id="f-1",
            content="a",
@@ -137,46 +137,46 @@ def step_known_ids(context: Context) -> None:
            relevance_score=0.5,
            source_decision_id="01HXDECISION00000000000002",
        ),
-    ]
+    )


@given("an empty fragment list")
 def step_empty_frags(context: Context) -> None:
-    context.fragments = []
+    context.fragments = ()


@given("a single fragment with {tokens:d} tokens")
 def step_single_frag(context: Context, tokens: int) -> None:
-    context.fragments = [
+    context.fragments = (
        _make_skel_fragment(
            fragment_id="only",
            content="x" * tokens,
            token_count=tokens,
            relevance_score=0.8,
        ),
-    ]
+    )


@given("a fragment with negative token count")
 def step_neg_tokens(context: Context) -> None:
    try:
-        context.fragments = [
+        context.fragments = (
            _make_skel_fragment(
                fragment_id="bad",
                content="x",
                token_count=-10,
                relevance_score=0.5,
            ),
-        ]
+        )
    except (ValueError, ValidationError) as exc:
        # Model-level validation now catches this at construction time.
        context.compressor_error = exc
-        context.fragments = []
+        context.fragments = ()


@given("a fragment with empty fragment_id")
 def step_empty_id(context: Context) -> None:
-    context.fragments = [
+    context.fragments = (
        ContextFragment(
            fragment_id="",
            uko_node="skeleton://empty-id",
@@ -185,24 +185,24 @@ def step_empty_id(context: Context) -> None:
            relevance_score=0.5,
            provenance=_SKEL_PROV,
        ),
-    ]
+    )


@given("a fragment with relevance {rel:g}")
 def step_bad_relevance(context: Context, rel: float) -> None:
    try:
-        context.fragments = [
+        context.fragments = (
            _make_skel_fragment(
                fragment_id="bad",
                content="x",
                token_count=10,
                relevance_score=rel,
            ),
-        ]
+        )
    except (ValueError, ValidationError) as exc:
        # Model-level validation now catches this at construction time.
        context.compressor_error = exc
-        context.fragments = []
+        context.fragments = ()


@given(
@@ -220,8 +220,8 @@ def step_make_metadata(context: Context, ratio: float, orig: int, comp: int) ->
 # --- When clauses ----------------------------------------------------------


-@when("I compress with skeleton_ratio {ratio:g}")
-def step_compress_ratio(context: Context, ratio: float) -> None:
+@when("I compress with skeleton_budget {budget:d}")
+def step_compress_budget(context: Context, budget: int) -> None:
    # If model-level validation already caught an error during fragment
    # construction (Given step), propagate it as the compress error.
    if getattr(context, "compressor_error", None) is not None:
@@ -229,45 +229,37 @@ def step_compress_ratio(context: Context, ratio: float) -> None:
        context.result = None
        return
    try:
-        context.result = context.service.compress(
-            context.fragments, skeleton_ratio=ratio
-        )
+        context.result = context.service.compress(context.fragments, budget)
        context.error = None
    except (ValueError, TypeError) as exc:
        context.error = exc
        context.result = None


-@when("I compress with skeleton_ratio not specified")
-def step_compress_default(context: Context) -> None:
-    context.result = context.service.compress(context.fragments)
-    context.error = None
-
-
-@when("I compress with a non-list fragments argument")
-def step_compress_non_list(context: Context) -> None:
+@when("I compress with a non-tuple fragments argument")
+def step_compress_non_tuple(context: Context) -> None:
    try:
-        context.result = context.service.compress("not-a-list", skeleton_ratio=0.5)  # type: ignore[arg-type]
+        context.result = context.service.compress("not-a-tuple", 500)  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc
        context.result = None


-@when("I compress with a list containing a non-fragment item")
+@when("I compress with a tuple containing a non-fragment item")
 def step_compress_non_fragment_item(context: Context) -> None:
    try:
        context.result = context.service.compress(
-            [
+            (
                _make_skel_fragment(
                    fragment_id="ok",
                    content="x",
                    token_count=10,
                    relevance_score=0.5,
                ),
-                "not-a-fragment",
-            ],  # type: ignore[list-item]
-            skeleton_ratio=0.5,
+                "not-a-fragment",  # type: ignore[arg-type]
+            ),
+            500,
        )
        context.error = None
    except TypeError as exc:
@@ -275,12 +267,12 @@ def step_compress_non_fragment_item(context: Context) -> None:
        context.result = None


-@when("I compress with a non-numeric skeleton_ratio")
-def step_compress_non_numeric_ratio(context: Context) -> None:
+@when("I compress with a non-integer skeleton_budget")
+def step_compress_non_integer_budget(context: Context) -> None:
    try:
        context.result = context.service.compress(
            context.fragments,
-            skeleton_ratio="bad",  # type: ignore[arg-type]
+            "bad",  # type: ignore[arg-type]
        )
        context.error = None
    except TypeError as exc:
@@ -315,11 +307,21 @@ def step_attach_to_plan(context: Context) -> None:
    )


+@when(
+    "I check that SkeletonCompressorService satisfies the SkeletonCompressor protocol"
+)
+def step_check_protocol(context: Context) -> None:
+    context.protocol_check_result = isinstance(
+        SkeletonCompressorService(), SkeletonCompressor
+    )
+    context.protocol_check_error = None
+
+
 # --- Then clauses ----------------------------------------------------------


-@then("the compressor should raise a ValueError for invalid ratio")
-def step_check_value_error_ratio(context: Context) -> None:
+@then("the compressor should raise a ValueError for invalid budget")
+def step_check_value_error_budget(context: Context) -> None:
    assert context.error is not None, "Expected ValueError"
    assert isinstance(context.error, ValueError), (
        f"Expected ValueError, got {type(context.error)}"
@@ -329,95 +331,33 @@ def step_check_value_error_ratio(context: Context) -> None:
@then("all fragments should be returned unchanged")
 def step_all_returned(context: Context) -> None:
    assert context.result is not None
-    assert len(context.result.fragments) == len(context.fragments)
-
-
-@then("only the highest-relevance fragment should be returned")
-def step_top_one(context: Context) -> None:
-    assert context.result is not None
-    assert len(context.result.fragments) == 1
-    top_relevance = max(f.relevance_score for f in context.fragments)
-    assert context.result.fragments[0].relevance_score == top_relevance
+    assert len(context.result) == len(context.fragments)


@then("compressed tokens should be at most {limit:d}")
 def step_tokens_limit(context: Context, limit: int) -> None:
    assert context.result is not None
-    assert context.result.metadata.compressed_tokens <= limit
-
-
-@then("the metadata ratio should equal the default {expected:g}")
-def step_default_ratio(context: Context, expected: float) -> None:
-    assert context.result is not None
-    assert context.result.metadata.ratio == expected
-    assert expected == DEFAULT_SKELETON_RATIO
+    total = sum(f.token_count for f in context.result)
+    assert total <= limit, f"Expected total tokens <= {limit}, got {total}"


@then("fragments should be ordered by fragment_id ascending")
 def step_ordered_by_id(context: Context) -> None:
    assert context.result is not None
-    ids = [f.fragment_id for f in context.result.fragments]
+    ids = [f.fragment_id for f in context.result]
    assert ids == sorted(ids), f"Expected sorted IDs, got {ids}"


@then("the first fragment should have relevance {rel:g}")
 def step_first_relevance(context: Context, rel: float) -> None:
    assert context.result is not None
-    assert context.result.fragments[0].relevance_score == rel
+    assert context.result[0].relevance_score == rel


@then("the last fragment should have relevance {rel:g}")
 def step_last_relevance(context: Context, rel: float) -> None:
    assert context.result is not None
-    assert context.result.fragments[-1].relevance_score == rel
-
-
-@then("metadata original_tokens should be {expected:d}")
-def step_original_tokens(context: Context, expected: int) -> None:
-    assert context.result is not None
-    assert context.result.metadata.original_tokens == expected
-
-
-@then("metadata compressed_tokens should equal {expected:d}")
-def step_compressed_equals(context: Context, expected: int) -> None:
-    assert context.result is not None
-    assert context.result.metadata.compressed_tokens == expected
-
-
-@then("metadata compressed_tokens should be at most {limit:d}")
-def step_compressed_at_most(context: Context, limit: int) -> None:
-    assert context.result is not None
-    assert context.result.metadata.compressed_tokens <= limit
-
-
-@then("metadata should contain all source decision IDs")
-def step_all_decision_ids(context: Context) -> None:
-    assert context.result is not None
-    expected_ids = {
-        f.metadata["source_decision_id"]
-        for f in context.fragments
-        if "source_decision_id" in f.metadata
-    }
-    actual_ids = set(context.result.metadata.source_decision_ids)
-    assert expected_ids == actual_ids
-
-
-@then("metadata ratio should be {expected:g}")
-def step_ratio_value(context: Context, expected: float) -> None:
-    assert context.result is not None
-    assert context.result.metadata.ratio == expected
-
-
-@then("the result should contain zero fragments")
-def step_zero_frags(context: Context) -> None:
-    assert context.result is not None
-    assert len(context.result.fragments) == 0
-
-
-@then("the result should contain one fragment")
-def step_one_frag(context: Context) -> None:
-    assert context.result is not None
-    assert len(context.result.fragments) == 1
+    assert context.result[-1].relevance_score == rel


@then("the compressor should raise a TypeError")
@@ -444,8 +384,8 @@ def step_type_error_item(context: Context) -> None:
    )


-@then("the compressor should raise a TypeError for invalid ratio type")
-def step_type_error_ratio_type(context: Context) -> None:
+@then("the compressor should raise a TypeError for invalid budget type")
+def step_type_error_budget_type(context: Context) -> None:
    assert context.error is not None, "Expected TypeError"
    assert isinstance(context.error, TypeError), (
        f"Expected TypeError, got {type(context.error)}"
@@ -457,13 +397,16 @@ def step_validation_error_compressed(context: Context) -> None:
    assert context.meta_error is not None, "Expected validation error"


-@then("compressed_tokens should be less than original_tokens")
-def step_less_tokens(context: Context) -> None:
+@then("the result should contain zero fragments")
+def step_zero_frags(context: Context) -> None:
    assert context.result is not None
-    assert (
-        context.result.metadata.compressed_tokens
-        < context.result.metadata.original_tokens
-    )
+    assert len(context.result) == 0
+
+
+@then("the result should contain one fragment")
+def step_one_frag(context: Context) -> None:
+    assert context.result is not None
+    assert len(context.result) == 1


@then("the plan should expose skeleton metadata in cli dict")
@@ -476,6 +419,16 @@ def step_plan_cli_dict(context: Context) -> None:
    assert skel["compressed_tokens"] == context.skel_meta.compressed_tokens


+@then("the structural subtype assertion should pass")
+def step_protocol_assertion_passes(context: Context) -> None:
+    assert context.protocol_check_error is None, (
+        f"Protocol check raised: {context.protocol_check_error}"
+    )
+    assert context.protocol_check_result is True, (
+        "SkeletonCompressorService does not satisfy the SkeletonCompressor protocol"
+    )
+
+
 # --- _validate_fragments edge-case steps --------------------------------


@@ -489,7 +442,7 @@ def step_frag_negative_token(context: Context) -> None:
        relevance_score=0.5,
        provenance=_SKEL_PROV,
    )
-    context.invalid_fragments = [frag]
+    context.invalid_fragments = (frag,)


@given("a context fragment constructed with relevance_score {score}")
@@ -502,7 +455,7 @@ def step_frag_bad_relevance(context: Context, score: str) -> None:
        relevance_score=float(score),
        provenance=_SKEL_PROV,
    )
-    context.invalid_fragments = [frag]
+    context.invalid_fragments = (frag,)


@given("a context fragment constructed with empty fragment_id")
@@ -515,7 +468,7 @@ def step_frag_empty_id(context: Context) -> None:
        relevance_score=0.5,
        provenance=_SKEL_PROV,
    )
-    context.invalid_fragments = [frag]
+    context.invalid_fragments = (frag,)


@when("I validate the invalid fragments")
@@ -651,3 +651,82 @@ def step_then_model_dump_contains_key(context: Context, key: str) -> None:
    assert context.skill_config is not None
    data: dict[str, Any] = context.skill_config.model_dump()
    assert key in data, f"Key '{key}' not found in model_dump: {list(data.keys())}"
+
+
+# ────────────────────────────────────────────────────────────
+# skill: wrapper key test fixtures and steps
+# ────────────────────────────────────────────────────────────
+
+_SPEC_COMPLIANT_WRAPPED_YAML = """\
+cleveragents:
+  version: "1.0"
+skill:
+  name: local/wrapped-skill
+  tools:
+    - name: builtin/shell_execute
+"""
+
+_SPEC_COMPLIANT_WRAPPED_META_YAML = """\
+cleveragents:
+  version: "2.0"
+skill:
+  name: local/wrapped-with-meta
+  description: "A wrapped skill with metadata"
+"""
+
+_SKILL_WRAPPER_NONE_YAML = """\
+skill:
+"""
+
+_SKILL_WRAPPER_STRING_YAML = """\
+skill: "just a string"
+"""
+
+_SKILL_WRAPPER_LIST_YAML = """\
+skill:
+  - item1
+  - item2
+"""
+
+_CLEVERAGENTS_FLAT_META_YAML = """\
+cleveragents:
+  version: "1.0"
+name: local/flat-with-meta
+description: "Flat config with stray metadata"
+"""
+
+
+@given("a spec-compliant skill YAML with skill: wrapper key")
+def step_given_spec_compliant_wrapper(context: Context) -> None:
+    """Provide a spec-compliant YAML with cleveragents: header and skill: wrapper."""
+    context.skill_yaml_string = _SPEC_COMPLIANT_WRAPPED_YAML
+
+
+@given("a spec-compliant skill YAML with cleveragents: header and skill: wrapper")
+def step_given_spec_compliant_with_meta(context: Context) -> None:
+    """Provide a spec-compliant YAML with both cleveragents: header and skill: wrapper."""
+    context.skill_yaml_string = _SPEC_COMPLIANT_WRAPPED_META_YAML
+
+
+@given("a skill YAML with skill: wrapper key with None value")
+def step_given_skill_wrapper_none(context: Context) -> None:
+    """Provide a YAML with skill: wrapper key but no value (None)."""
+    context.skill_yaml_string = _SKILL_WRAPPER_NONE_YAML
+
+
+@given("a skill YAML with skill: wrapper key and string value")
+def step_given_skill_wrapper_string(context: Context) -> None:
+    """Provide a YAML with skill: wrapper key but a non-dict string value."""
+    context.skill_yaml_string = _SKILL_WRAPPER_STRING_YAML
+
+
+@given("a skill YAML with skill: wrapper key and list value")
+def step_given_skill_wrapper_list(context: Context) -> None:
+    """Provide a YAML with skill: wrapper key but a non-dict list value."""
+    context.skill_yaml_string = _SKILL_WRAPPER_LIST_YAML
+
+
+@given("a skill YAML with cleveragents: header and flat format")
+def step_given_cleveragents_flat_meta(context: Context) -> None:
+    """Provide a flat-format YAML with cleveragents: metadata but no skill: wrapper."""
+    context.skill_yaml_string = _CLEVERAGENTS_FLAT_META_YAML
@@ -195,6 +195,13 @@ def step_no_mocked_tools(context: Context) -> None:
    context.mock_service.list_tools.return_value = []


+@given("there is a mocked read-only tool in the registry")
+def step_mocked_read_only_tool(context: Context) -> None:
+    tool = _make_mock_tool("local/read-only-tool", "tool", "custom")
+    tool["capability"] = {"read_only": True, "writes": False, "checkpointable": False}
+    context.mock_service.list_tools.return_value = [tool]
+
+
@given('there is a mocked tool with name "{name}"')
 def step_mocked_tool_exists(context: Context, name: str) -> None:
    context.mock_service.get_tool.return_value = _make_mock_tool(name)
@@ -632,6 +639,17 @@ def step_tool_list_empty(context: Context) -> None:
    assert "No tools found" in context.tool_result.output


+@then("the Rich tool list output contains all five spec columns")
+def step_rich_tool_list_columns(context: Context) -> None:
+    assert context.tool_result is not None
+    assert context.tool_result.exit_code == 0
+    output = context.tool_result.output
+    for col in ("Name", "Type", "Source", "Read-Only", "Writes"):
+        assert col in output, (
+            f"Expected column '{col}' in Rich table output. Got: {output}"
+        )
+
+
@then("the tool CLI should show the tool details")
 def step_tool_show_details(context: Context) -> None:
    assert context.tool_result is not None
@@ -113,6 +113,26 @@ Feature: Tool and Validation CLI commands
    When I run tool CLI list with format "table"
    Then the tool CLI list should succeed with results

+  Scenario: Rich table list has exactly the five spec columns
+    Given there are mocked tools in the registry
+    When I run tool CLI list
+    Then the Rich tool list output contains all five spec columns
+
+  Scenario: Rich table list renders checkmark for read-only capability
+    Given there is a mocked read-only tool in the registry
+    When I run tool CLI list
+    Then the tool CLI output should contain "✓"
+
+  Scenario: Rich table list renders dash for non-read-only capability
+    Given there are mocked tools in the registry
+    When I run tool CLI list
+    Then the tool CLI output should contain "—"
+
+  Scenario: Rich table list shows a Summary panel
+    Given there are mocked tools in the registry
+    When I run tool CLI list
+    Then the tool CLI output should contain "Summary"
+
  # Tool show command tests
  Scenario: Show tool by name
    Given there is a mocked tool with name "local/test-tool"
@@ -93,3 +93,25 @@ All Six Formats Work With Version Command
    ${result}=    Run Process    ${PYTHON}    ${HELPER}    global-format-all-six    cwd=${WORKSPACE}
    Should Be Equal As Integers    ${result.rc}    0
    Should Contain    ${result.stdout}    cli-global-format-all-six-ok
+
+# ---------------------------------------------------------------------------
+# format_output() renderer routing tests (Issue #2921)
+# ---------------------------------------------------------------------------
+
+Format Output Rich Produces Styled Output Not JSON
+    [Documentation]    Verify that format_output(data, "rich") produces styled terminal output
+    ...    and NOT raw JSON (fix for issue #2921: rich format silently fell back to JSON)
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    format-output-rich    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-formats-format-output-rich-ok
+
+Format Output Color Produces ANSI Colored Output Not Plain Text
+    [Documentation]    Verify that format_output(data, "color") produces ANSI-colored output
+    ...    and NOT plain text (fix for issue #2921: color format used plain renderer)
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    format-output-color    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cli-formats-format-output-color-ok
@@ -49,13 +49,13 @@ def _mock_action(name: str = "local/fmt-smoke") -> Action:
        read_only=False,
        state=ActionState.AVAILABLE,
        created_by=None,
-        created_at=datetime.now(),
-        updated_at=datetime.now(),
+        created_at=datetime(2025, 1, 15, 10, 0, 0),
+        updated_at=datetime(2025, 1, 15, 10, 0, 0),
    )


 def _mock_plan(name: str = "local/fmt-smoke-plan") -> Plan:
-    now = datetime.now()
+    now = datetime(2025, 1, 15, 10, 0, 0)
    return Plan(
        identity=PlanIdentity(plan_id=_ULID),
        namespaced_name=NamespacedName.parse(name),
@@ -214,6 +214,69 @@ def global_format_all_six() -> None:
    print("cli-global-format-all-six-ok")


+def format_output_rich() -> None:
+    """Verify format_output(data, 'rich') produces styled output, not JSON.
+
+    Regression test for issue #2921: rich format silently fell back to JSON.
+    """
+    from cleveragents.cli.formatting import format_output
+
+    data = {"name": "smoke-test", "status": "active", "count": 42}
+    result = format_output(data, "rich")
+
+    # Must NOT be valid JSON (that was the bug)
+    try:
+        json.loads(result)
+        raise AssertionError(
+            f"format_output(data, 'rich') returned valid JSON — "
+            f"expected styled terminal output, got: {result!r}"
+        )
+    except (json.JSONDecodeError, ValueError):
+        pass  # Expected: rich output is not JSON
+
+    # Must contain the data keys
+    assert "name" in result, f"Rich output missing 'name': {result!r}"
+    assert "status" in result, f"Rich output missing 'status': {result!r}"
+
+    # Must contain ANSI codes or panel header (styled output)
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_panel = "Output" in result
+    assert has_ansi or has_panel, (
+        f"Rich output has no ANSI codes or panel header: {result!r}"
+    )
+    print("cli-formats-format-output-rich-ok")
+
+
+def format_output_color() -> None:
+    """Verify format_output(data, 'color') produces ANSI-colored output, not plain text.
+
+    Regression test for issue #2921: color format used plain renderer instead of color.
+    """
+    from cleveragents.cli.formatting import format_output
+
+    data = {"name": "smoke-test", "status": "active", "count": 42}
+    result = format_output(data, "color")
+
+    # Must contain the data keys
+    assert "name" in result, f"Color output missing 'name': {result!r}"
+    assert "status" in result, f"Color output missing 'status': {result!r}"
+
+    # Must NOT be plain text only (that was the bug)
+    plain_text = "name: smoke-test\nstatus: active\ncount: 42"
+    assert result != plain_text, (
+        f"format_output(data, 'color') returned plain text — "
+        f"expected ANSI-colored output, got: {result!r}"
+    )
+
+    # Must contain ANSI codes or panel header (styled output)
+    has_ansi = "\033[" in result or "\x1b[" in result
+    has_panel = "Output" in result
+    assert has_ansi or has_panel, (
+        f"Color output has no ANSI codes or panel header: {result!r}"
+    )
+    print("cli-formats-format-output-color-ok")
+
+
 _COMMANDS = {
    "action-list-json": action_list_json,
    "action-show-yaml": action_show_yaml,
@@ -226,6 +289,8 @@ _COMMANDS = {
    "global-format-json-diagnostics": global_format_json_diagnostics,
    "global-format-shorthand": global_format_shorthand,
    "global-format-all-six": global_format_all_six,
+    "format-output-rich": format_output_rich,
+    "format-output-color": format_output_color,
 }

 if __name__ == "__main__":
@@ -1,10 +1,10 @@
 """Helper script for skeleton compressor Robot Framework tests.

 Usage:
-    python helper_skeleton_compressor.py compress <ratio>
-    python helper_skeleton_compressor.py validate-ratio-bounds
-    python helper_skeleton_compressor.py metadata-fields
+    python helper_skeleton_compressor.py compress <budget>
+    python helper_skeleton_compressor.py validate-budget-bounds
    python helper_skeleton_compressor.py stable-ordering
+    python helper_skeleton_compressor.py protocol-check
 """

 from __future__ import annotations
@@ -17,6 +17,9 @@ _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

+from cleveragents.application.services.acms_service import (  # noqa: E402
+    SkeletonCompressor,
+)
 from cleveragents.application.services.skeleton_compressor import (  # noqa: E402
    SkeletonCompressorService,
 )
@@ -51,9 +54,9 @@ def _make_skel_fragment(
    )


-def _sample_fragments() -> list[ContextFragment]:
+def _sample_fragments() -> tuple[ContextFragment, ...]:
    """Build a repeatable set of sample fragments."""
-    return [
+    return (
        _make_skel_fragment(
            fragment_id="frag-001",
            content="High relevance content " * 20,
@@ -75,73 +78,44 @@ def _sample_fragments() -> list[ContextFragment]:
            relevance_score=0.3,
            source_decision_id="01HXDECISION00000000000003",
        ),
-    ]
+    )


-def cmd_compress(ratio_str: str) -> None:
+def cmd_compress(budget_str: str) -> None:
    """Compress sample fragments and print summary."""
-    ratio = float(ratio_str)
+    budget = int(budget_str)
    svc = SkeletonCompressorService()
-    result = svc.compress(_sample_fragments(), skeleton_ratio=ratio)
-    meta = result.metadata
-    print(f"skeleton-compress-ok ratio={meta.ratio}")
-    print(f"original_tokens={meta.original_tokens}")
-    print(f"compressed_tokens={meta.compressed_tokens}")
-    print(f"fragment_count={len(result.fragments)}")
-    print(f"decision_ids={len(meta.source_decision_ids)}")
+    result = svc.compress(_sample_fragments(), budget)
+    total_tokens = sum(f.token_count for f in result)
+    print(f"skeleton-compress-ok budget={budget}")
+    print(f"fragment_count={len(result)}")
+    print(f"total_tokens={total_tokens}")


-def cmd_validate_ratio_bounds() -> None:
-    """Verify that out-of-range ratios raise ValueError."""
+def cmd_validate_budget_bounds() -> None:
+    """Verify that negative budgets raise ValueError."""
    svc = SkeletonCompressorService()
    frags = _sample_fragments()

-    for bad in (-0.1, 1.5, 2.0):
+    for bad in (-1, -10, -100):
        try:
-            svc.compress(frags, skeleton_ratio=bad)
-            print(f"FAIL: ratio {bad} did not raise")
+            svc.compress(frags, bad)
+            print(f"FAIL: budget {bad} did not raise")
            sys.exit(1)
        except ValueError:
            pass

    # Valid bounds
-    for good in (0.0, 0.5, 1.0):
-        svc.compress(frags, skeleton_ratio=good)
+    for good in (0, 500, 1000):
+        svc.compress(frags, good)

-    print("skeleton-ratio-bounds-ok")
-
-
-def cmd_metadata_fields() -> None:
-    """Verify metadata fields are populated correctly."""
-    svc = SkeletonCompressorService()
-    result = svc.compress(_sample_fragments(), skeleton_ratio=0.5)
-    meta = result.metadata
-
-    checks_passed = True
-
-    if meta.ratio != 0.5:
-        print(f"FAIL: ratio={meta.ratio}")
-        checks_passed = False
-    if meta.original_tokens != 1000:
-        print(f"FAIL: original_tokens={meta.original_tokens}")
-        checks_passed = False
-    if meta.compressed_tokens > 500:
-        print(f"FAIL: compressed_tokens={meta.compressed_tokens} > 500")
-        checks_passed = False
-    if not meta.source_decision_ids:
-        print("FAIL: no decision IDs")
-        checks_passed = False
-
-    if checks_passed:
-        print("skeleton-metadata-ok")
-    else:
-        sys.exit(1)
+    print("skeleton-budget-bounds-ok")


 def cmd_stable_ordering() -> None:
    """Verify fragments come out in deterministic order."""
    svc = SkeletonCompressorService()
-    frags = [
+    frags = tuple(
        _make_skel_fragment(
            fragment_id=f"frag-{chr(ord('c') - i)}",
            content="x",
@@ -149,13 +123,13 @@ def cmd_stable_ordering() -> None:
            relevance_score=0.5,
        )
        for i in range(3)
-    ]
+    )

-    r1 = svc.compress(frags, skeleton_ratio=0.0)
-    r2 = svc.compress(list(reversed(frags)), skeleton_ratio=0.0)
+    r1 = svc.compress(frags, 10000)
+    r2 = svc.compress(tuple(reversed(frags)), 10000)

-    ids1 = [f.fragment_id for f in r1.fragments]
-    ids2 = [f.fragment_id for f in r2.fragments]
+    ids1 = [f.fragment_id for f in r1]
+    ids2 = [f.fragment_id for f in r2]

    if ids1 == ids2:
        print("skeleton-stable-ordering-ok")
@@ -164,6 +138,19 @@ def cmd_stable_ordering() -> None:
        sys.exit(1)


+def cmd_protocol_check() -> None:
+    """Verify SkeletonCompressorService satisfies the SkeletonCompressor protocol."""
+    svc = SkeletonCompressorService()
+    if isinstance(svc, SkeletonCompressor):
+        print("skeleton-protocol-check-ok")
+    else:
+        print(
+            "FAIL: SkeletonCompressorService does not satisfy"
+            " SkeletonCompressor protocol"
+        )
+        sys.exit(1)
+
+
 def main() -> None:
    """Dispatch subcommand."""
    if len(sys.argv) < 2:
@@ -173,15 +160,15 @@ def main() -> None:
    cmd = sys.argv[1]
    if cmd == "compress":
        if len(sys.argv) < 3:
-            print("Usage: compress <ratio>")
+            print("Usage: compress <budget>")
            sys.exit(1)
        cmd_compress(sys.argv[2])
-    elif cmd == "validate-ratio-bounds":
-        cmd_validate_ratio_bounds()
-    elif cmd == "metadata-fields":
-        cmd_metadata_fields()
+    elif cmd == "validate-budget-bounds":
+        cmd_validate_budget_bounds()
    elif cmd == "stable-ordering":
        cmd_stable_ordering()
+    elif cmd == "protocol-check":
+        cmd_protocol_check()
    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)
@@ -8,39 +8,39 @@ Suite Teardown   Cleanup Test Environment
 ${HELPER}    ${CURDIR}/helper_skeleton_compressor.py

 *** Test Cases ***
-Compress Fragments At Ratio 0.5
-    [Documentation]    Compress sample fragments at 50% ratio and verify output
-    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    0.5    cwd=${WORKSPACE}
+Compress Fragments At Budget 500
+    [Documentation]    Compress sample fragments with budget of 500 tokens and verify output
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    500    cwd=${WORKSPACE}
    Log    ${result.stdout}
    Log    ${result.stderr}
    Should Be Equal As Integers    ${result.rc}    0
-    Should Contain    ${result.stdout}    skeleton-compress-ok
+    Should Contain    ${result.stdout}    skeleton-compress-ok budget=500

-Compress Fragments At Ratio 0.0
-    [Documentation]    No compression — all fragments should survive
-    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    0.0    cwd=${WORKSPACE}
+Compress Fragments At Full Budget
+    [Documentation]    Full budget — all fragments should survive
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    1000    cwd=${WORKSPACE}
    Should Be Equal As Integers    ${result.rc}    0
-    Should Contain    ${result.stdout}    skeleton-compress-ok
+    Should Contain    ${result.stdout}    skeleton-compress-ok budget=1000
    Should Contain    ${result.stdout}    fragment_count=3

-Compress Fragments At Ratio 1.0
-    [Documentation]    Maximum compression — only top fragment should survive
-    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    1.0    cwd=${WORKSPACE}
+Compress Fragments At Budget Zero
+    [Documentation]    Zero budget — no fragments should survive
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    compress    0    cwd=${WORKSPACE}
    Should Be Equal As Integers    ${result.rc}    0
-    Should Contain    ${result.stdout}    skeleton-compress-ok
-    Should Contain    ${result.stdout}    fragment_count=1
+    Should Contain    ${result.stdout}    skeleton-compress-ok budget=0
+    Should Contain    ${result.stdout}    fragment_count=0

-Validate Ratio Bounds
-    [Documentation]    Out-of-range ratios must raise ValueError
-    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validate-ratio-bounds    cwd=${WORKSPACE}
+Validate Budget Bounds
+    [Documentation]    Out-of-range budgets must raise ValueError
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validate-budget-bounds    cwd=${WORKSPACE}
    Should Be Equal As Integers    ${result.rc}    0
-    Should Contain    ${result.stdout}    skeleton-ratio-bounds-ok
+    Should Contain    ${result.stdout}    skeleton-budget-bounds-ok

-Verify Metadata Fields
-    [Documentation]    Metadata should record ratio, tokens, and decision IDs
-    ${result}=    Run Process    ${PYTHON}    ${HELPER}    metadata-fields    cwd=${WORKSPACE}
+Verify Protocol Conformance
+    [Documentation]    SkeletonCompressorService must satisfy the SkeletonCompressor protocol
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    protocol-check    cwd=${WORKSPACE}
    Should Be Equal As Integers    ${result.rc}    0
-    Should Contain    ${result.stdout}    skeleton-metadata-ok
+    Should Contain    ${result.stdout}    skeleton-protocol-check-ok

 Verify Stable Fragment Ordering
    [Documentation]    Fragments with equal relevance must be ordered deterministically
@@ -49,3 +49,21 @@ Reject Invalid Skill YAML
    Should Be Equal As Integers    ${result.rc}    0
    Should Contain    ${result.stdout}    skill-schema-expected-fail

+Validate Spec-Compliant Skill YAML With skill: Wrapper Key
+    [Tags]    tdd_issue    tdd_issue_1472
+    [Documentation]    Parse a spec-compliant YAML with cleveragents: header and skill: wrapper key
+    ${wrapped_yaml}=    Set Variable    ${TEMPDIR}${/}wrapped_skill.yaml
+    ${yaml_content}=    Catenate    SEPARATOR=\n
+    ...    cleveragents:
+    ...    ${SPACE}${SPACE}version: "1.0"
+    ...    skill:
+    ...    ${SPACE}${SPACE}name: local/wrapped-skill
+    ...    ${SPACE}${SPACE}description: "A wrapped skill"
+    ...
+    Create File    ${wrapped_yaml}    ${yaml_content}
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validate    ${wrapped_yaml}    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-schema-ok
+
@@ -255,9 +255,6 @@ if TYPE_CHECKING:
    from cleveragents.application.services.session_service import (
        PersistentSessionService as PersistentSessionService,
    )
-    from cleveragents.application.services.skeleton_compressor import (
-        CompressionResult as CompressionResult,
-    )
    from cleveragents.application.services.skeleton_compressor import (
        SkeletonCompressorService as SkeletonCompressorService,
    )
@@ -516,7 +513,6 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
    "resolve_severity": ("semantic_validation_service", "resolve_severity"),
    "ServiceRetryWiring": ("service_retry_wiring", "ServiceRetryWiring"),
    "PersistentSessionService": ("session_service", "PersistentSessionService"),
-    "CompressionResult": ("skeleton_compressor", "CompressionResult"),
    "SkeletonCompressorService": ("skeleton_compressor", "SkeletonCompressorService"),
    "SkillRegistryService": ("skill_registry_service", "SkillRegistryService"),
    "CoordinationResult": ("strategy_coordinator", "CoordinationResult"),
@@ -493,6 +493,7 @@ class PreambleGenerator(Protocol):
    ) -> str | None: ...


+@runtime_checkable
 class SkeletonCompressor(Protocol):
    """Compress parent context into a skeleton for child plan inheritance.

@@ -3,105 +3,91 @@
 The ``SkeletonCompressorService`` takes a collection of context
 fragments from a parent plan and produces a compressed representation
 suitable for propagation to child plans.  The compression is governed
-by ``skeleton_ratio``:
+by ``skeleton_budget`` — an absolute token count:

- **0.0** — no compression; all fragments pass through unchanged.
- **1.0** — maximum compression; only the single highest-relevance
-  fragment is kept (with minimal content).
+- **0** — no fragments are kept (empty result).
+- **large value** — all fragments pass through if they fit.

 Fragments are sorted by ``relevance_score`` in **descending** order
 (highest first).  A stable secondary sort on ``fragment_id`` ensures
 deterministic output for equal-relevance fragments.

+The caller is responsible for converting a ``skeleton_ratio`` (fraction
+of the total context budget) to an absolute ``skeleton_budget`` before
+invoking ``compress()``.  For example::
+
+    skeleton_budget = int(total_tokens * skeleton_ratio)
+    compressed = service.compress(fragments, skeleton_budget)
+
 Based on ``docs/specification.md`` ACMS Skeleton section.
 """

 from __future__ import annotations

-from dataclasses import dataclass
-
+from cleveragents.application.services.acms_service import SkeletonCompressor
 from cleveragents.domain.models.core.context_fragment import ContextFragment
-from cleveragents.domain.models.core.skeleton_metadata import SkeletonMetadata
-
-# ---------------------------------------------------------------------------
-# Public data structures
-# ---------------------------------------------------------------------------
-
-
-@dataclass(frozen=True, slots=True)
-class CompressionResult:
-    """Output of a skeleton compression pass.
-
-    Attributes:
-        fragments: The compressed (filtered/truncated) fragments,
-            ordered by relevance descending then fragment_id ascending.
-        metadata: Auditable metadata for the compression pass.
-    """
-
-    fragments: tuple[ContextFragment, ...]
-    metadata: SkeletonMetadata
-

 # ---------------------------------------------------------------------------
 # Service
 # ---------------------------------------------------------------------------

-# Default skeleton_ratio when the caller does not specify one.
-DEFAULT_SKELETON_RATIO: float = 0.15
-

 class SkeletonCompressorService:
    """Compress context fragments for subplan context inheritance.

-    The service is stateless; all state lives in the arguments and
-    the returned ``CompressionResult``.
+    Implements the ``SkeletonCompressor`` protocol defined in
+    :mod:`cleveragents.application.services.acms_service`.  The service
+    is stateless; all state lives in the arguments and the returned tuple.
+
+    The caller is responsible for computing ``skeleton_budget`` from a
+    ratio and the total available token count before invoking
+    :meth:`compress`.
    """

    def compress(
        self,
-        fragments: list[ContextFragment],
-        skeleton_ratio: float | None = None,
-    ) -> CompressionResult:
-        """Compress *fragments* according to *skeleton_ratio*.
+        fragments: tuple[ContextFragment, ...],
+        skeleton_budget: int,
+    ) -> tuple[ContextFragment, ...]:
+        """Compress *fragments* to fit within *skeleton_budget* tokens.

        Args:
            fragments: Context fragments to compress.  Each must
                have ``token_count >= 0`` and ``relevance_score`` in
                ``[0.0, 1.0]``.
-            skeleton_ratio: Compression ratio in ``[0.0, 1.0]``.
-                ``None`` falls back to ``DEFAULT_SKELETON_RATIO``.
+            skeleton_budget: Maximum total token count for the returned
+                fragments.  Must be a non-negative integer.  When zero
+                an empty tuple is returned.

        Returns:
-            A ``CompressionResult`` containing the filtered fragments
-            and associated ``SkeletonMetadata``.
+            A tuple of ``ContextFragment`` objects whose combined
+            ``token_count`` does not exceed *skeleton_budget*, ordered
+            by relevance descending then ``fragment_id`` ascending.

        Raises:
-            ValueError: If *skeleton_ratio* is outside ``[0.0, 1.0]``
-                or any fragment has invalid fields.
-            TypeError: If *fragments* is not a list.
+            TypeError: If *fragments* is not a tuple or *skeleton_budget*
+                is not an integer.
+            ValueError: If *skeleton_budget* is negative or any fragment
+                has invalid fields.
        """
        # -- argument validation ------------------------------------------
-        if not isinstance(fragments, list):
-            raise TypeError(f"fragments must be a list, got {type(fragments).__name__}")
-
-        effective_ratio = (
-            skeleton_ratio if skeleton_ratio is not None else DEFAULT_SKELETON_RATIO
-        )
-
-        if not isinstance(effective_ratio, (int, float)):
+        if not isinstance(fragments, tuple):
            raise TypeError(
-                f"skeleton_ratio must be a float, got {type(effective_ratio).__name__}"
+                f"fragments must be a tuple, got {type(fragments).__name__}"
            )

-        if effective_ratio < 0.0 or effective_ratio > 1.0:
-            raise ValueError(
-                f"skeleton_ratio must be in [0.0, 1.0], got {effective_ratio}"
+        if not isinstance(skeleton_budget, int):
+            raise TypeError(
+                f"skeleton_budget must be an int, got {type(skeleton_budget).__name__}"
            )

+        if skeleton_budget < 0:
+            raise ValueError(f"skeleton_budget must be >= 0, got {skeleton_budget}")
+
        self._validate_fragments(fragments)

-        # -- compute totals -----------------------------------------------
-        original_tokens = sum(f.token_count for f in fragments)
+        if skeleton_budget == 0 or not fragments:
+            return ()

        # -- stable sort: relevance desc, fragment_id asc -----------------
        sorted_fragments = sorted(
@@ -110,35 +96,17 @@ class SkeletonCompressorService:
        )

        # -- select fragments within budget -------------------------------
-        kept = self._select_fragments(sorted_fragments, effective_ratio)
+        kept = self._select_fragments(sorted_fragments, skeleton_budget)

-        compressed_tokens = sum(f.token_count for f in kept)
-
-        source_ids = tuple(
-            f.metadata["source_decision_id"]
-            for f in kept
-            if "source_decision_id" in f.metadata
-        )
-
-        metadata = SkeletonMetadata(
-            ratio=effective_ratio,
-            original_tokens=original_tokens,
-            compressed_tokens=compressed_tokens,
-            source_decision_ids=source_ids,
-        )
-
-        return CompressionResult(
-            fragments=tuple(kept),
-            metadata=metadata,
-        )
+        return tuple(kept)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    @staticmethod
-    def _validate_fragments(fragments: list[ContextFragment]) -> None:
-        """Validate every fragment in the list.
+    def _validate_fragments(fragments: tuple[ContextFragment, ...]) -> None:
+        """Validate every fragment in the tuple.

        Raises:
            TypeError: If an element is not a ``ContextFragment``.
@@ -165,31 +133,15 @@ class SkeletonCompressorService:
    @staticmethod
    def _select_fragments(
        sorted_fragments: list[ContextFragment],
-        ratio: float,
+        budget: int,
    ) -> list[ContextFragment]:
-        """Select which fragments to keep given the compression ratio.
+        """Select which fragments to keep given the token *budget*.

-        When *ratio* is 0.0 every fragment is kept.  When *ratio* is
-        1.0 only the single highest-relevance fragment survives (or
-        none if the input is empty).  For intermediate values the
-        token budget is ``original_tokens * (1 - ratio)``; fragments
-        are added in relevance order until the budget is exhausted.
+        Fragments are added in relevance order (highest first) until the
+        budget is exhausted.  At least one fragment is always included
+        when the input is non-empty, even if it exceeds the budget, to
+        ensure callers always receive some context.
        """
-        if not sorted_fragments:
-            return []
-
-        if ratio == 0.0:
-            return list(sorted_fragments)
-
-        original_tokens = sum(f.token_count for f in sorted_fragments)
-
-        # Budget: fraction of tokens to *keep*
-        budget = int(original_tokens * (1.0 - ratio))
-
-        # At maximum compression keep at most one fragment
-        if ratio == 1.0:
-            budget = 0
-
        kept: list[ContextFragment] = []
        used = 0
        for frag in sorted_fragments:
@@ -198,11 +150,22 @@ class SkeletonCompressorService:
                break
            kept.append(frag)
            used += frag.token_count
-            if used >= budget and budget > 0:
+            if used >= budget:
                break

-        # At ratio 1.0, keep exactly the top fragment
-        if ratio == 1.0 and sorted_fragments:
-            return [sorted_fragments[0]]
-
        return kept
+
+
+# ---------------------------------------------------------------------------
+# Structural subtype assertion — prevents future protocol drift
+# ---------------------------------------------------------------------------
+# This assertion verifies at import time that SkeletonCompressorService
+# satisfies the SkeletonCompressor Protocol.  If the protocol signature
+# changes and the service is not updated, this will raise AssertionError
+# immediately rather than failing silently at runtime.
+
+assert isinstance(SkeletonCompressorService(), SkeletonCompressor), (
+    "SkeletonCompressorService does not satisfy the SkeletonCompressor protocol. "
+    "Ensure compress(fragments: tuple[ContextFragment, ...], skeleton_budget: int) "
+    "-> tuple[ContextFragment, ...] matches the protocol definition."
+)
@@ -38,7 +38,7 @@ code: |
 |----------------------------|------------------------------------------|
 | ``Config file error``      | File not found or not readable           |
 | ``Schema validation error``| YAML does not match Tool schema          |
-| ``Duplicate tool``         | Tool name already registered             |
+| ``Duplicate tool``         | Tool name already registered           |

 Based on implementation_plan.md -- Task C1.tool.cli.
 """
@@ -416,29 +416,75 @@ def list_tools(
            console.print(format_output(data, fmt))
            return

-        # Rich table
+        # Rich table — spec-compliant: Name, Type, Source, Read-Only, Writes
        table = Table(title=f"Tools ({len(tools)} total)")
        table.add_column("Name", style="cyan")
        table.add_column("Type", style="blue")
        table.add_column("Source", style="magenta")
-        table.add_column("Description", style="dim")
-        table.add_column("Timeout", justify="right")
+        table.add_column("Read-Only", justify="center", style="green")
+        table.add_column("Writes", justify="center", style="yellow")
+
+        read_only_count: int = 0
+        writes_count: int = 0
+        tool_count: int = 0
+        validation_count: int = 0
+        namespaces: set[str] = set()

        for tool in tools:
            spec = _tool_spec_dict(tool)
-            desc = str(spec.get("description", ""))
-            if len(desc) > 40:
-                desc = desc[:37] + "..."
+            name_val = str(spec.get("name", ""))
+            tool_type_val = str(spec.get("tool_type", "tool"))
+            source_val = str(spec.get("source", ""))
+
+            # Determine Read-Only / Writes from capability metadata
+            capability = spec.get("capability", {})
+            if isinstance(capability, dict):
+                is_read_only: bool = capability.get("read_only", False)
+                has_writes: bool = capability.get("writes", False)
+            else:
+                is_read_only = False
+                has_writes = False
+            read_only_str: str = "\u2713" if is_read_only else "\u2014"
+            writes_str: str = "\u2713" if has_writes else "\u2014"
+
            table.add_row(
-                str(spec.get("name", "")),
-                str(spec.get("tool_type", "tool")),
-                str(spec.get("source", "")),
-                desc,
-                str(spec.get("timeout", 300)),
+                name_val,
+                tool_type_val,
+                source_val,
+                read_only_str,
+                writes_str,
            )

+            # Accumulate summary statistics
+            namespaces.add(name_val.split("/", 1)[0] if "/" in name_val else name_val)
+            if tool_type_val == "validation":
+                validation_count += 1
+            else:
+                tool_count += 1
+            if is_read_only:
+                read_only_count += 1
+            if has_writes:
+                writes_count += 1
+
        console.print(table)

+        # Summary panel
+        summary_lines = [
+            f"  Total: {len(tools)}",
+            f"  Tools: {tool_count}",
+            f"  Validations: {validation_count}",
+            f"  Read-Only: {read_only_count}",
+            f"  Writes: {writes_count}",
+            f"  Namespaces: {len(namespaces)}",
+        ]
+        summary_panel = Panel(
+            "\n".join(summary_lines),
+            title="Summary",
+            expand=False,
+        )
+        console.print(summary_panel)
+        console.print(f"\n\u2713 OK {len(tools)} tools listed\n")
+
    except CleverAgentsError as exc:
        console.print(f"[red]Error:[/red] {exc.message}")
        raise typer.Abort() from exc
@@ -14,6 +14,7 @@ Based on v3_spec.md implementation plan Stage A4b / G5b.render.
 from __future__ import annotations

 import json
+import sys
 import time
 from datetime import datetime
 from enum import Enum, StrEnum
@@ -149,6 +150,58 @@ def _format_table(data: dict[str, Any] | list[dict[str, Any]]) -> str:
    return buf.getvalue().rstrip("\n")


+def _format_rich(data: dict[str, Any] | list[dict[str, Any]]) -> str:
+    """Render data using the Rich materializer (ANSI-styled terminal output).
+
+    Delegates to :func:`format_output_session` with the ``rich`` strategy so
+    that the output is consistent with the newer session-based rendering path.
+    The data passed here has already been redacted by the caller.
+    """
+    strategy = RichMaterializer()
+    with OutputSession(
+        format="rich", command="format_output", strategy=strategy
+    ) as session:
+        if isinstance(data, list):
+            for idx, item in enumerate(data):
+                if isinstance(item, dict):
+                    panel = session.panel(f"Item {idx + 1}")
+                    for key, val in item.items():
+                        panel.set_entry(key, str(_serialize_value(val)))
+                    panel.close()
+        elif isinstance(data, dict):
+            panel = session.panel("Output")
+            for key, val in data.items():
+                panel.set_entry(key, str(_serialize_value(val)))
+            panel.close()
+    return strategy.get_output()
+
+
+def _format_color(data: dict[str, Any] | list[dict[str, Any]]) -> str:
+    """Render data using the Color materializer (ANSI-colored terminal output).
+
+    Delegates to :func:`format_output_session` with the ``color`` strategy so
+    that the output is consistent with the newer session-based rendering path.
+    The data passed here has already been redacted by the caller.
+    """
+    strategy = ColorMaterializer()
+    with OutputSession(
+        format="color", command="format_output", strategy=strategy
+    ) as session:
+        if isinstance(data, list):
+            for idx, item in enumerate(data):
+                if isinstance(item, dict):
+                    panel = session.panel(f"Item {idx + 1}")
+                    for key, val in item.items():
+                        panel.set_entry(key, str(_serialize_value(val)))
+                    panel.close()
+        elif isinstance(data, dict):
+            panel = session.panel("Output")
+            for key, val in data.items():
+                panel.set_entry(key, str(_serialize_value(val)))
+            panel.close()
+    return strategy.get_output()
+
+
 def _redact_data(
    data: dict[str, Any] | list[dict[str, Any]],
 ) -> dict[str, Any] | list[dict[str, Any]]:
@@ -283,8 +336,6 @@ def format_output(
        The rendered string.  For machine-readable formats the output
        is written directly to stdout and an empty string is returned.
    """
-    import sys
-
    t_start = time.monotonic()
    safe_data = _redact_data(data)
    fmt = format_type.lower()
@@ -316,9 +367,11 @@ def format_output(

    if fmt == OutputFormat.TABLE.value:
        return _format_table(safe_data)
+    if fmt == OutputFormat.RICH.value:
+        return _format_rich(safe_data)
    if fmt == OutputFormat.COLOR.value:
-        return format_output_session(safe_data, fmt)
-    # ``rich`` and any unknown value fall back to JSON
+        return _format_color(safe_data)
+    # Unknown format: fall back to JSON
    return _format_json(safe_data)


@@ -110,7 +110,8 @@ class LegacyDataMigrator:
                            (p for p in all_plans if p.name == plan_name), None
                        )
                        if existing_plan:
-                            plan_id_map[plan_name] = existing_plan.id  # type: ignore
+                            assert existing_plan.id is not None
+                            plan_id_map[plan_name] = existing_plan.id
                            continue

                        # Create new plan
@@ -14,13 +14,17 @@ which defines a single ``resolve`` method returning a

 ## Built-in Handlers

-| Handler                  | Resource Type(s)          | Strategy             |
-|--------------------------|---------------------------|----------------------|
-| ``GitCheckoutHandler``   | ``git-checkout``          | ``git_worktree``     |
-| ``FsDirectoryHandler``   | ``fs-directory``          | ``copy_on_write``    |
-| ``DevcontainerHandler``  | ``devcontainer-instance`` | ``snapshot``         |
-| ``CloudResourceHandler`` | ``cloud-*``, ``aws-*``   | ``none``             |
-| ``DatabaseHandler``      | ``postgres``, ``mysql``   | ``transaction``      |
+- ``GitCheckoutHandler``      — ``git-checkout`` (git_worktree)
+- ``FsDirectoryHandler``      — ``fs-directory`` (copy_on_write)
+- ``DevcontainerHandler``     — ``devcontainer-instance`` (snapshot)
+- ``CloudResourceHandler``    — ``cloud-*``, ``aws-*`` (none)
+- ``DatabaseHandler``         — ``postgres``, ``mysql`` (transaction)
+- ``ContainerRuntimeHandler`` — ``container-runtime`` (none)
+- ``ContainerImageHandler``   — ``container-image`` (none)
+- ``ContainerChildHandler``   — ``container-mount``, ``container-exec-env``,
+  ``container-port`` (snapshot)
+- ``ContainerVolumeHandler``  — ``container-volume`` (snapshot)
+- ``ContainerNetworkHandler`` — ``container-network`` (none)

 ## Handler Resolution

@@ -30,6 +34,13 @@ dynamically imports the module and returns an instance.
 """

 from cleveragents.resource.handlers.cloud import CloudResourceHandler
+from cleveragents.resource.handlers.container import (
+    ContainerChildHandler,
+    ContainerImageHandler,
+    ContainerNetworkHandler,
+    ContainerRuntimeHandler,
+    ContainerVolumeHandler,
+)
 from cleveragents.resource.handlers.database import DatabaseResourceHandler
 from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
 from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
@@ -54,6 +65,11 @@ __all__ = [
    "AccessResult",
    "CheckpointResult",
    "CloudResourceHandler",
+    "ContainerChildHandler",
+    "ContainerImageHandler",
+    "ContainerNetworkHandler",
+    "ContainerRuntimeHandler",
+    "ContainerVolumeHandler",
    "Content",
    "DatabaseResourceHandler",
    "DeleteResult",
@@ -0,0 +1,331 @@
+"""Container infrastructure resource handlers for CleverAgents.
+
+Implements the five handler classes referenced in
+``_resource_registry_container.py`` for the seven built-in container
+infrastructure resource types:
+
+- ``ContainerRuntimeHandler`` -- for ``container-runtime``
+- ``ContainerImageHandler``   -- for ``container-image``
+- ``ContainerChildHandler``   -- for ``container-mount``, ``container-exec-env``,
+  ``container-port`` (shared child handler)
+- ``ContainerVolumeHandler``  -- for ``container-volume``
+- ``ContainerNetworkHandler`` -- for ``container-network``
+
+All handlers extend BaseResourceHandler and satisfy the ResourceHandler protocol.
+
+Container resources use ``sandbox_strategy = "none"`` (or ``"snapshot"``
+for child/volume types) because the container runtime itself provides
+isolation.  Actual container SDK operations are NOT implemented --
+calling :meth:`resolve` raises :exc:`NotImplementedError` after basic
+validation.  This mirrors the pattern used by CloudResourceHandler.
+
+Based on:
+    - src/cleveragents/application/services/_resource_registry_container.py
+    - docs/specification.md ~line 25096-25114 (container handler table)
+    - ADR-039: Container Resource Types
+    - Issue #2907: Container infrastructure resource types missing handler module
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+from typing import Any
+
+from cleveragents.domain.models.core.resource import Resource, SandboxStrategy
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
+from cleveragents.resource.handlers._base import EMPTY_CONTENT_HASH, BaseResourceHandler
+from cleveragents.resource.handlers.protocol import (
+    Content,
+    DeleteResult,
+    DiffResult,
+    WriteResult,
+)
+from cleveragents.tool.context import BoundResource
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "ContainerChildHandler",
+    "ContainerImageHandler",
+    "ContainerNetworkHandler",
+    "ContainerRuntimeHandler",
+    "ContainerVolumeHandler",
+]
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+
+def _identity_hash(resource: Resource, algorithm: str = "sha256") -> str:
+    """Compute an identity hash for a container resource.
+
+    The hash is based on the resource type name and location (e.g. socket
+    path, image reference, volume name, network name).  Returns
+    :data:`EMPTY_CONTENT_HASH` if no location is set.
+
+    Args:
+        resource: The container resource.
+        algorithm: Hash algorithm name (default ``sha256``).
+
+    Returns:
+        Hex-encoded hash digest, or :data:`EMPTY_CONTENT_HASH`.
+    """
+    if not resource.location:
+        return EMPTY_CONTENT_HASH
+    h = hashlib.new(algorithm)
+    h.update(resource.resource_type_name.encode("utf-8"))
+    h.update(b"\0")
+    h.update(resource.location.encode("utf-8"))
+    return h.hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Shared base for all container handlers
+# ---------------------------------------------------------------------------
+
+
+class _ContainerBaseHandler(BaseResourceHandler):
+    """Shared base for all container infrastructure handlers.
+
+    Provides:
+    - ``resolve`` -- raises NotImplementedError (container sandbox
+      provisioning is not yet implemented).
+    - ``content_hash`` -- identity hash based on resource type + location.
+    - All CRUD and lifecycle stubs -- raise NotImplementedError.
+
+    Subclasses set ``_default_strategy`` and ``_type_label``.
+    """
+
+    def resolve(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        slot_name: str,
+        sandbox_manager: SandboxManager,
+        access: str = "read_only",
+    ) -> BoundResource:
+        """Raise -- container sandbox provisioning is not yet implemented.
+
+        Args:
+            resource: A container resource instance.
+            plan_id: The plan requesting the sandbox.
+            slot_name: Name of the tool resource slot being filled.
+            sandbox_manager: The sandbox lifecycle manager (unused).
+            access: Access mode (unused).
+
+        Raises:
+            NotImplementedError: Always.
+        """
+        raise NotImplementedError(
+            f"Container resource sandbox provisioning is not yet implemented. "
+            f"Resource '{resource.resource_id}' "
+            f"(type={resource.resource_type_name}, "
+            f"plan={plan_id}, slot={slot_name})."
+        )
+
+    # -- CRUD stubs --------------------------------------------------------
+
+    def read(self, *, resource: Resource, path: str = "") -> Content:
+        """Not supported for container resources."""
+        raise NotImplementedError(f"{self._type_label} handler does not support read()")
+
+    def write(self, *, resource: Resource, path: str, data: bytes) -> WriteResult:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support write()"
+        )
+
+    def delete(self, *, resource: Resource, path: str = "") -> DeleteResult:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support delete()"
+        )
+
+    def list_children(self, *, resource: Resource) -> list[str]:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support list_children()"
+        )
+
+    def diff(self, *, resource: Resource, other_location: str) -> DiffResult:
+        """Not supported for container resources."""
+        raise NotImplementedError(f"{self._type_label} handler does not support diff()")
+
+    def discover_children(self, *, resource: Resource) -> list[Resource]:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support discover_children()"
+        )
+
+    # -- Lifecycle stubs ---------------------------------------------------
+
+    def create_sandbox(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> Any:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support create_sandbox()"
+        )
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> Any:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support create_checkpoint()"
+        )
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> Any:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support rollback_to()"
+        )
+
+    def project_access(
+        self,
+        *,
+        resource: Resource,
+        principal: str,
+        action: str = "read",
+        project_id: str = "",
+    ) -> Any:
+        """Not supported for container resources."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support project_access()"
+        )
+
+    # -- Content hashing ---------------------------------------------------
+
+    def content_hash(
+        self,
+        resource: Resource,
+        *,
+        algorithm: str = "sha256",
+    ) -> str:
+        """Compute an identity hash for a container resource.
+
+        The hash is based on the resource type name and location.
+        Returns :data:`EMPTY_CONTENT_HASH` if no location is set.
+
+        Args:
+            resource: The container resource.
+            algorithm: Hash algorithm name (default ``sha256``).
+
+        Returns:
+            Hex-encoded hash digest, or :data:`EMPTY_CONTENT_HASH`.
+        """
+        return _identity_hash(resource, algorithm)
+
+
+# ---------------------------------------------------------------------------
+# Concrete handler classes
+# ---------------------------------------------------------------------------
+
+
+class ContainerRuntimeHandler(_ContainerBaseHandler):
+    """Handler for ``container-runtime`` resource types.
+
+    A container runtime resource represents a container engine
+    (Docker, Podman, containerd, etc.) identified by its socket path
+    or remote endpoint URL.
+
+    Sandbox provisioning is NOT implemented -- calling :meth:`resolve`
+    raises :exc:`NotImplementedError`.
+
+    This handler satisfies the ResourceHandler protocol.
+    """
+
+    _default_strategy = SandboxStrategy.NONE
+    _type_label = "container-runtime"
+
+
+class ContainerImageHandler(_ContainerBaseHandler):
+    """Handler for ``container-image`` resource types.
+
+    A container image resource represents an immutable container image
+    identified by tag and/or digest (e.g. ``ubuntu:22.04``,
+    ``ghcr.io/org/app@sha256:...``).
+
+    Sandbox provisioning is NOT implemented -- calling :meth:`resolve`
+    raises :exc:`NotImplementedError`.
+
+    This handler satisfies the ResourceHandler protocol.
+    """
+
+    _default_strategy = SandboxStrategy.NONE
+    _type_label = "container-image"
+
+
+class ContainerChildHandler(_ContainerBaseHandler):
+    """Shared handler for container child resource types.
+
+    Used by three resource types that represent child components of a
+    running container instance:
+
+    - ``container-mount`` -- bind mount or volume mount inside a container
+    - ``container-exec-env`` -- execution environment variables and config
+    - ``container-port`` -- port mapping between host and container
+
+    These resources are not user-addable (``user_addable = False``) and
+    are discovered automatically from a running container instance.
+
+    Sandbox provisioning is NOT implemented -- calling :meth:`resolve`
+    raises :exc:`NotImplementedError`.
+
+    This handler satisfies the ResourceHandler protocol.
+    """
+
+    _default_strategy = SandboxStrategy.SNAPSHOT
+    _type_label = "container-child"
+
+
+class ContainerVolumeHandler(_ContainerBaseHandler):
+    """Handler for ``container-volume`` resource types.
+
+    A container volume resource represents a named container volume
+    managed by the runtime engine (Docker, Podman, etc.).
+
+    Sandbox provisioning is NOT implemented -- calling :meth:`resolve`
+    raises :exc:`NotImplementedError`.
+
+    This handler satisfies the ResourceHandler protocol.
+    """
+
+    _default_strategy = SandboxStrategy.SNAPSHOT
+    _type_label = "container-volume"
+
+
+class ContainerNetworkHandler(_ContainerBaseHandler):
+    """Handler for ``container-network`` resource types.
+
+    A container network resource represents a container network
+    (bridge, host, overlay, macvlan, or none).
+
+    Sandbox provisioning is NOT implemented -- calling :meth:`resolve`
+    raises :exc:`NotImplementedError`.
+
+    This handler satisfies the ResourceHandler protocol.
+    """
+
+    _default_strategy = SandboxStrategy.NONE
+    _type_label = "container-network"
@@ -6,6 +6,8 @@ Provides :class:`SkillConfigSchema`, a Pydantic model that:
 * Normalizes camelCase keys to snake_case before validation.
 * Interpolates ``${ENV_VAR}`` placeholders from environment variables.
 * Produces clear, actionable error messages for every validation failure.
+* Handles the spec-required ``skill:`` wrapper key and ``cleveragents:``
+  metadata block for forward-compatible YAML parsing.

 Schema definition lives in ``docs/schema/skill.schema.yaml``.
 Example configs live under ``examples/skills/``.
@@ -392,6 +394,30 @@ class SkillConfigSchema(BaseModel):
    def from_yaml(cls, yaml_string: str) -> SkillConfigSchema:
        """Parse and validate a skill YAML string.

+        Handles both the spec-required YAML format with a top-level
+        ``skill:`` wrapper key and optional ``cleveragents:`` metadata
+        block, as well as the legacy flat format for backward
+        compatibility.
+
+        Spec-compliant YAML:
+
+        .. code-block:: yaml
+
+            cleveragents:
+              version: "1.0"
+            skill:
+              name: local/my-skill
+              tools:
+                - name: builtin/shell_execute
+
+        Flat (legacy) YAML, also supported:
+
+        .. code-block:: yaml
+
+            name: local/my-skill
+            tools:
+              - name: builtin/shell_execute
+
        Args:
            yaml_string: Raw YAML content.

@@ -399,7 +425,8 @@ class SkillConfigSchema(BaseModel):
            Validated ``SkillConfigSchema`` instance.

        Raises:
-            ValueError: If the YAML is not a mapping or is empty.
+            ValueError: If the YAML is not a mapping or is empty,
+                or if ``skill:`` has an invalid value.
            pydantic.ValidationError: If schema validation fails.
        """
        if yaml_string is None:
@@ -415,7 +442,35 @@ class SkillConfigSchema(BaseModel):
                f"Skill YAML must be a mapping (key: value), got {type(raw).__name__}."
            )

+        # ── Normalize camelCase keys ────────────────────────────
        normalized = _normalize_keys(raw)
+
+        # ── Strip optional cleveragents metadata header ─────────
+        normalized.pop("cleveragents", None)
+
+        # ── Unwrap skill: wrapper key if present ────────────────
+        if "skill" in normalized:
+            wrapper = normalized.pop("skill")
+            if wrapper is None:
+                raise ValueError(
+                    "skill: key is present but empty. "
+                    "Provide a skill mapping with at least a 'name' field."
+                )
+            if not isinstance(wrapper, dict):
+                raise ValueError(
+                    f"skill: key must contain a mapping (dict), "
+                    f"got {type(wrapper).__name__}. "
+                    "Wrap the skill configuration as: skill:\\n  <your config here>"
+                )
+            # Merge wrapper content with normalized keys.
+            # Wrapper values take precedence so the spec-compliant
+            # format works reliably even if both formats overlap.
+            wrapper_dict: dict[str, Any] = wrapper
+            for key, value in wrapper_dict.items():
+                snake_key = _CAMEL_TO_SNAKE.get(key, key)
+                normalized[snake_key] = value
+
+        # ── Interpolate environment variables ───────────────────
        interpolated = _interpolate_env_vars(normalized)
        return cls.model_validate(interpolated)

@@ -574,8 +574,6 @@ validate_spawn  # noqa: B018, F821
 SkeletonMetadata  # noqa: B018, F821
 SkeletonCompressorService  # noqa: B018, F821
 ContextFragment  # noqa: B018, F821
-CompressionResult  # noqa: B018, F821
-DEFAULT_SKELETON_RATIO  # noqa: B018, F821
 skeleton_compressor_service  # noqa: B018, F821
 skeleton_metadata  # noqa: B018, F821
 source_decision_ids  # noqa: B018, F821