feat(skill): add skill registry persistence

alembic/versions/c0_001_skill_registry_tables.py

@@ -0,0 +1,70 @@
+"""Add skill registry tables (skills, skill_items).
+
+This migration creates the two core skill registry tables needed for
+the Skill Registry Persistence layer:
+
+- ``skills``: Registered skill definitions with namespaced names as
+  primary keys, description, version, metadata JSON, and timestamps.
+- ``skill_items``: Individual skill items (tool refs, includes, inline
+  tools, MCP sources, agent sources) linked to skills with stable
+  ordering and JSON config.
+
+Revision ID: c0_001_skill_registry
+Revises: 71cd40eb661f
+Create Date: 2026-02-19 12:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "c0_001_skill_registry"
+down_revision: str | Sequence[str] | None = "71cd40eb661f"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # -- skills table --
+    op.create_table(
+        "skills",
+        sa.Column("name", sa.String(255), primary_key=True),
+        sa.Column("namespace", sa.String(100), nullable=False),
+        sa.Column("short_name", sa.String(150), nullable=False),
+        sa.Column("description", sa.Text(), nullable=False),
+        sa.Column("version", sa.String(50), nullable=True),
+        sa.Column("metadata_json", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.Column("updated_at", sa.String(30), nullable=False),
+    )
+    op.create_index("ix_skills_namespace", "skills", ["namespace"])
+
+    # -- skill_items table --
+    op.create_table(
+        "skill_items",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column(
+            "skill_name",
+            sa.String(255),
+            sa.ForeignKey("skills.name", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("item_type", sa.String(30), nullable=False),
+        sa.Column("item_name", sa.String(500), nullable=False),
+        sa.Column("item_config", sa.Text(), nullable=True),
+        sa.Column("item_order", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.String(30), nullable=False),
+    )
+    op.create_index("ix_skill_items_skill_name", "skill_items", ["skill_name"])
+    op.create_index("ix_skill_items_item_type", "skill_items", ["item_type"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_skill_items_item_type", table_name="skill_items")
+    op.drop_index("ix_skill_items_skill_name", table_name="skill_items")
+    op.drop_table("skill_items")
+    op.drop_index("ix_skills_namespace", table_name="skills")
+    op.drop_table("skills")

alembic/versions/c0_002_merge_skill_registry_head.py

@@ -0,0 +1,33 @@
+"""Merge skill registry and actor/automation heads.
+
+After rebasing feature/m3-skill-registry onto the session-persistence
+chain, c0_001_skill_registry still pointed at 71cd40eb661f while the
+main chain had advanced through a7_002 -> c3_001 -> a6_002.  This
+no-op merge migration resolves the two heads into one.
+
+Revision ID: c0_002_merge_skill_registry
+Revises: a6_002_drop_automation_level, c0_001_skill_registry
+Create Date: 2026-02-21 12:00:00
+
+"""
+
+from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "c0_002_merge_skill_registry"
+down_revision: str | Sequence[str] | None = (
+    "a6_002_drop_automation_level",
+    "c0_001_skill_registry",
+)
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    pass
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    pass

benchmarks/skill_registry_bench.py

@@ -0,0 +1,127 @@
+"""ASV benchmarks for Skill Registry persistence operations.
+
+Measures the performance of:
+- SkillModel.from_domain() construction
+- SkillModel.to_domain() reconstruction
+- SkillRepository CRUD operations (in-memory SQLite)
+- SkillRegistryService list operations
+"""
+
+from __future__ import annotations
+
+import sys
+import uuid
+from pathlib import Path
+
+try:
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    from cleveragents.domain.models.core.skill import Skill
+    from cleveragents.infrastructure.database.models import Base, SkillModel
+    from cleveragents.infrastructure.database.repositories import (
+        SkillRepository,
+    )
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    from cleveragents.domain.models.core.skill import Skill
+    from cleveragents.infrastructure.database.models import Base, SkillModel
+    from cleveragents.infrastructure.database.repositories import (
+        SkillRepository,
+    )
+
+
+def _make_skill(name: str = "bench/skill-test") -> Skill:
+    """Create a minimal Skill domain object for benchmarks."""
+    return Skill(
+        name=name,
+        description=f"Benchmark skill {name}",
+        tool_refs=["local/tool-a", "local/tool-b"],
+    )
+
+
+class SkillModelConstruction:
+    """Benchmark SkillModel.from_domain() construction."""
+
+    def setup(self) -> None:
+        self.skill = _make_skill()
+
+    def time_from_domain(self) -> None:
+        SkillModel.from_domain(self.skill)
+
+
+class SkillModelReconstruction:
+    """Benchmark SkillModel.to_domain() reconstruction."""
+
+    def setup(self) -> None:
+        self.model = SkillModel.from_domain(_make_skill())
+
+    def time_to_domain(self) -> None:
+        self.model.to_domain()
+
+
+class SkillRepositoryCRUD:
+    """Benchmark SkillRepository CRUD with in-memory SQLite."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", echo=False)
+        Base.metadata.create_all(self.engine)
+        self.factory = sessionmaker(bind=self.engine)
+        self.repo = SkillRepository(session_factory=self.factory)
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_create_skill(self) -> None:
+        skill = _make_skill(f"bench/{uuid.uuid4().hex[:8]}")
+        self.repo.create(skill)
+        # Commit via the same session-factory the repo uses internally
+        session = self.factory()
+        try:
+            session.commit()
+        finally:
+            session.close()
+
+    def time_get_skill(self) -> None:
+        self.repo.get_by_name("bench/nonexistent")
+
+    def time_list_all(self) -> None:
+        self.repo.list_all()
+
+    def time_list_namespace(self) -> None:
+        self.repo.list_all(namespace="bench")
+
+
+class SkillRegistryListPerformance:
+    """Benchmark listing performance with many skills."""
+
+    timeout = 120.0
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", echo=False)
+        Base.metadata.create_all(self.engine)
+        self.factory = sessionmaker(bind=self.engine)
+        self.repo = SkillRepository(session_factory=self.factory)
+        # Insert 100 skills and commit via the same session-factory
+        for i in range(100):
+            skill = _make_skill(f"bench/skill-{i:04d}")
+            self.repo.create(skill)
+        session = self.factory()
+        try:
+            session.commit()
+        finally:
+            session.close()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_list_100_skills(self) -> None:
+        self.repo.list_all()
+
+    def time_list_100_skills_filtered(self) -> None:
+        self.repo.list_all(namespace="bench")

docs/reference/skill_registry.md

@@ -0,0 +1,91 @@
+# Skill Registry
+
+The Skill Registry provides persistent storage for skill definitions in
+CleverAgents. Skills are namespaced, reusable collections of tools
+registered via YAML config and persisted to the `skills` and
+`skill_items` database tables.
+
+## Registration Behavior
+
+Skills are registered with a namespaced name (`namespace/short_name`)
+as the unique identifier. The registry enforces:
+
+- **Name uniqueness**: Duplicate names are rejected with
+  `DuplicateSkillError`.
+- **Name validation**: Names must match `^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$`.
+- **Cascading deletes**: Removing a skill cascades to all child
+  `skill_items` rows.
+
+## Skill Items
+
+Each skill stores its components as ordered items in the `skill_items`
+table. The `item_type` discriminator distinguishes:
+
+| item_type      | Description                              |
+|----------------|------------------------------------------|
+| `tool_ref`     | Reference to a named tool in the registry |
+| `include`      | Recursive inclusion of another skill      |
+| `inline_tool`  | Anonymous tool defined inline             |
+| `mcp_source`   | MCP server tool source                    |
+| `agent_source` | Agent Skills Standard folder source       |
+
+Items are stored with a stable `item_order` to preserve definition
+ordering across round-trips.
+
+## Filters
+
+The `SkillRegistryService.list_skills()` method supports:
+
+- **Namespace filter**: Pass `namespace="local"` to list only skills
+  in the `local` namespace.
+
+## Service API
+
+```python
+from cleveragents.application.services import SkillRegistryService
+from cleveragents.infrastructure.database.repositories import SkillRepository
+
+svc = SkillRegistryService(skill_repo=SkillRepository(session_factory))
+
+# Register
+svc.add_skill(skill)
+
+# Retrieve
+skill = svc.get_skill("local/code-tools")
+
+# List with filter
+skills = svc.list_skills(namespace="local")
+
+# Update
+svc.update_skill(updated_skill)
+
+# Remove
+svc.remove_skill("local/code-tools")
+```
+
+## Database Schema
+
+### skills
+
+| Column         | Type         | Description                     |
+|----------------|--------------|---------------------------------|
+| name           | String(255)  | PK, namespaced name             |
+| namespace      | String(100)  | Extracted namespace              |
+| short_name     | String(150)  | Extracted short name             |
+| description    | Text         | Skill description                |
+| version        | String(50)   | Optional version string          |
+| metadata_json  | Text         | JSON metadata (overrides, etc.)  |
+| created_at     | String(30)   | ISO-8601 creation timestamp      |
+| updated_at     | String(30)   | ISO-8601 update timestamp        |
+
+### skill_items
+
+| Column      | Type         | Description                          |
+|-------------|--------------|--------------------------------------|
+| id          | Integer      | Auto-increment PK                    |
+| skill_name  | String(255)  | FK to skills.name (CASCADE)          |
+| item_type   | String(30)   | Discriminator (tool_ref, etc.)       |
+| item_name   | String(500)  | Item identifier or name              |
+| item_config | Text         | Optional JSON configuration          |
+| item_order  | Integer      | Stable ordering index                |
+| created_at  | String(30)   | ISO-8601 creation timestamp          |

features/skill_registry.feature

@@ -0,0 +1,214 @@
+@phase1 @domain @repository @skill_registry
+Feature: Skill Registry Persistence
+  As a system operator managing skills
+  I want the skill registry to reliably persist, retrieve, and filter skills
+  So that the execution layer can discover and resolve skill tool sets
+
+  Background:
+    Given a clean in-memory database with the skill registry schema
+    And a skill registry repository backed by the database
+    And a skill registry service backed by the repository
+
+  # ---------------------------------------------------------------------------
+  # Registering skills
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_create
+  Scenario: A new skill is registered and retrievable
+    Given a valid skill named "local/code-tools" with description "Code tools"
+    When the skill is registered through the skill registry service
+    Then the skill registry service should not raise an error
+    And the persisted skill should have the name "local/code-tools"
+
+  @skill_reg_create
+  Scenario: A skill with tool refs is registered
+    Given a valid skill named "local/editor" with description "Editor tools"
+    And the skill has tool refs "local/edit-file,local/read-file"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 2 items of type "tool_ref"
+
+  @skill_reg_create
+  Scenario: A skill with includes is registered
+    Given a valid skill named "local/full-stack" with description "Full stack"
+    And the skill includes "local/frontend,local/backend"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 2 items of type "include"
+
+  @skill_reg_create
+  Scenario: A skill with inline tools is registered
+    Given a valid skill named "local/custom" with description "Custom tools"
+    And the skill has an inline tool with description "Custom lint"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 1 items of type "inline_tool"
+
+  @skill_reg_create
+  Scenario: A skill with MCP sources is registered
+    Given a valid skill named "local/mcp-skill" with description "MCP tools"
+    And the skill has an MCP source "github-server"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 1 items of type "mcp_source"
+
+  @skill_reg_create
+  Scenario: A skill with agent sources is registered
+    Given a valid skill named "local/agent-skill" with description "Agent tools"
+    And the skill has an agent source "/skills/my-agent"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 1 items of type "agent_source"
+
+  @skill_reg_create @error_handling
+  Scenario: Registering a skill with a duplicate name is rejected
+    Given a valid skill named "local/code-tools" with description "Code tools"
+    And the skill has already been registered once in the skill registry
+    When a second skill with the same name "local/code-tools" is registered
+    Then a DuplicateSkillError should be raised mentioning "local/code-tools"
+
+  # ---------------------------------------------------------------------------
+  # Retrieving skills
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_get
+  Scenario: Retrieving a non-existent skill returns None
+    When I retrieve skill "local/nonexistent" from the skill registry
+    Then the skill registry result should be None
+
+  @skill_reg_get
+  Scenario: Retrieving a registered skill returns correct data
+    Given a valid skill named "local/dev-tools" with description "Dev tools"
+    And the skill has tool refs "local/lint"
+    When the skill is registered through the skill registry service
+    And I retrieve skill "local/dev-tools" from the skill registry
+    Then the persisted skill should have the name "local/dev-tools"
+    And the persisted skill description should be "Dev tools"
+
+  # ---------------------------------------------------------------------------
+  # Listing skills with filters
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_list
+  Scenario: List skills returns all skills
+    Given the following skills are registered:
+      | name              | description     |
+      | local/code-tools  | Code tools      |
+      | local/dev-tools   | Dev tools       |
+      | devops/deploy     | Deploy tools    |
+    When I list all skills from the skill registry
+    Then the skill list should contain 3 skills
+
+  @skill_reg_list
+  Scenario: List skills with namespace filter
+    Given the following skills are registered:
+      | name              | description     |
+      | local/code-tools  | Code tools      |
+      | local/dev-tools   | Dev tools       |
+      | devops/deploy     | Deploy tools    |
+    When I list skills from the skill registry with namespace "local"
+    Then the skill list should contain 2 skills
+
+  # ---------------------------------------------------------------------------
+  # Updating skills
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_update
+  Scenario: Updating a skill changes its description
+    Given a valid skill named "local/code-tools" with description "Code tools"
+    And the skill is registered through the skill registry service
+    When I update skill "local/code-tools" with description "Updated tools"
+    Then the persisted skill description should be "Updated tools"
+
+  @skill_reg_update @error_handling
+  Scenario: Updating a non-existent skill raises error
+    When I try to update a non-existent skill "local/ghost"
+    Then a SkillNotFoundError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Removing skills
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_delete
+  Scenario: Removing a registered skill succeeds
+    Given a valid skill named "local/temp-skill" with description "Temp"
+    And the skill is registered through the skill registry service
+    When I remove skill "local/temp-skill" from the skill registry
+    Then the skill removal should return True
+    And I retrieve skill "local/temp-skill" from the skill registry
+    And the skill registry result should be None
+
+  @skill_reg_delete
+  Scenario: Removing a non-existent skill returns False
+    When I remove skill "local/nonexistent" from the skill registry
+    Then the skill removal should return False
+
+  # ---------------------------------------------------------------------------
+  # Invalid include detection
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_include @error_handling
+  Scenario: A skill with invalid include name persists but include is stored
+    Given a valid skill named "local/with-includes" with description "Has includes"
+    And the skill includes "local/nonexistent-skill"
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 1 items of type "include"
+
+  # ---------------------------------------------------------------------------
+  # Overrides round-trip (TEST-1)
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_overrides
+  Scenario: Skill overrides survive persistence round-trip
+    Given a valid skill named "local/override-test" with overrides for "local/tool-a"
+    When the skill is registered and retrieved with overrides check
+    Then the skill registry service should not raise an error
+    And the retrieved skill should have overrides for "local/tool-a"
+
+  # ---------------------------------------------------------------------------
+  # Item ordering stability (TEST-1)
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_ordering
+  Scenario: Skill items maintain stable ordering across round-trip
+    Given a valid skill named "local/ordered-test" with mixed item types
+    And the skill is registered through the skill registry service
+    When I retrieve the skill and check item ordering
+    Then the retrieved skill should have items in stable order
+
+  # ---------------------------------------------------------------------------
+  # Name validation at persistence layer (SEC-1 / TEST-1)
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_name_validation @error_handling
+  Scenario: Empty name is rejected at persistence layer
+    When I register a skill with empty name via dict bypass
+    Then a ValueError should be raised about invalid skill name
+
+  @skill_reg_name_validation @error_handling
+  Scenario: Name without slash is rejected at persistence layer
+    When I register a skill with invalid name "noslash" via dict bypass
+    Then a ValueError should be raised about invalid skill name
+
+  @skill_reg_name_validation @error_handling
+  Scenario: Name with special characters is rejected at persistence layer
+    When I register a skill with invalid name "bad/na me!" via dict bypass
+    Then a ValueError should be raised about invalid skill name
+
+  # ---------------------------------------------------------------------------
+  # Large payload (TEST-1)
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_large
+  Scenario: Skill with 50 tool refs persists correctly
+    Given a valid skill named "local/large-skill" with 50 tool refs
+    When the skill is registered through the skill registry service
+    Then the persisted skill should have 50 tool refs
+
+  # ---------------------------------------------------------------------------
+  # Domain object return type (DESIGN-1)
+  # ---------------------------------------------------------------------------
+
+  @skill_reg_domain
+  Scenario: Listed skills are proper Skill domain objects
+    Given the following skills are registered:
+      | name              | description     |
+      | local/code-tools  | Code tools      |
+      | local/dev-tools   | Dev tools       |
+    When I list all skills from the skill registry
+    Then the skill list entries should all be Skill domain objects

features/skill_repo_coverage.feature

@@ -0,0 +1,185 @@
+@phase1 @domain @repository @skill_repo_coverage
+Feature: Skill Repository Coverage – Uncovered Lines
+  As a developer maintaining the skill registry
+  I want all repository error handlers and conditional branches fully tested
+  So that database failures and edge-case data paths are exercised
+
+  # ===========================================================================
+  # SkillInUseError constructor (lines 4027-4031)
+  # ===========================================================================
+
+  @srcover_error_class
+  Scenario: SkillInUseError stores name and message without detail
+    When a SkillInUseError is created for skill "local/my-skill" without detail
+    Then the SkillInUseError message should contain "local/my-skill"
+    And the SkillInUseError skill_name should be "local/my-skill"
+
+  @srcover_error_class
+  Scenario: SkillInUseError stores name and detail in message
+    When a SkillInUseError is created for skill "local/busy" with detail "3 plans active"
+    Then the SkillInUseError message should contain "local/busy"
+    And the SkillInUseError message should contain "3 plans active"
+    And the SkillInUseError skill_name should be "local/busy"
+
+  # ===========================================================================
+  # ToolRepository.__init__ TypeError guard (line 3387)
+  # ===========================================================================
+
+  @srcover_tool_init
+  Scenario: ToolRepository without session factory raises TypeError
+    When a ToolRepository is created without any session factory
+    Then a TypeError should be raised about session_factory requirement
+
+  # ===========================================================================
+  # ToolRepository._extract_value enum unwrapping (line 3410)
+  # ===========================================================================
+
+  @srcover_extract_value
+  Scenario: ToolRepository extracts value from enum-like object
+    When _extract_value is called with an enum-like attribute having value "active"
+    Then the extracted value should be "active"
+
+  # ===========================================================================
+  # ToolRepository.remove success return path (line 3514)
+  # ===========================================================================
+
+  @srcover_tool_remove
+  Scenario: ToolRepository.remove returns True for existing tool
+    Given a tool repository with a real in-memory database for srcover
+    And a tool "test/removable" exists in the srcover tool repository
+    When the tool "test/removable" is removed via srcover repository
+    Then the srcover tool removal result should be True
+
+  # ===========================================================================
+  # ValidationAttachmentRepository.create with non-None args (line 3589)
+  # ===========================================================================
+
+  @srcover_val_args
+  Scenario: ValidationAttachment created with args serialises JSON
+    Given a validation attachment repository with real database for srcover
+    When a validation is attached with args {"retries": 3, "timeout": 120} for srcover
+    Then the srcover attachment should have args_json containing "retries"
+
+  # ===========================================================================
+  # ValidationAttachmentRepository.get_by_id returns None (line 3714)
+  # ===========================================================================
+
+  @srcover_val_get_none
+  Scenario: ValidationAttachmentRepository.get_by_id returns None for missing ID
+    Given a validation attachment repository with real database for srcover
+    When get_by_id is called with a non-existent attachment ID for srcover
+    Then the srcover get_by_id result should be None
+
+  # ===========================================================================
+  # ValidationAttachmentRepository.get_by_id OperationalError (lines 3725-3726)
+  # ===========================================================================
+
+  @srcover_val_get_error
+  Scenario: ValidationAttachmentRepository.get_by_id raises DatabaseError on OperationalError
+    Given a validation attachment repository with mock session raising OperationalError for srcover
+    When get_by_id is called and an OperationalError occurs for srcover
+    Then a srcover DatabaseError should be raised containing "Failed to get attachment"
+
+  # ===========================================================================
+  # ResourceRepository.auto_discover_children – empty child_type_name (line 2555)
+  # ===========================================================================
+
+  @srcover_auto_discover
+  Scenario: auto_discover_children skips rules with empty type name
+    Given a resource repository with in-memory database for srcover auto-discover
+    And a resource type with auto_discover rules containing an empty type for srcover
+    And a resource of that type exists for srcover auto-discover
+    When auto_discover_children is called for the srcover resource
+    Then the srcover auto-discover result should be an empty list
+
+  # ===========================================================================
+  # SkillRepository.create – OperationalError handler (lines 4093-4095)
+  # ===========================================================================
+
+  @srcover_skill_create_error
+  Scenario: SkillRepository.create raises DatabaseError on OperationalError
+    Given a skill repository with mock session raising OperationalError on flush for srcover
+    When a skill is created through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to create skill"
+
+  # ===========================================================================
+  # SkillRepository.create – non-UNIQUE IntegrityError (line 4090)
+  # ===========================================================================
+
+  @srcover_skill_create_error
+  Scenario: SkillRepository.create raises DatabaseError on non-UNIQUE IntegrityError
+    Given a skill repository with mock session raising non-UNIQUE IntegrityError for srcover
+    When a skill is created through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to create skill"
+
+  # ===========================================================================
+  # SkillRepository.create – DuplicateSkillError re-raise (lines 4091-4092)
+  # ===========================================================================
+
+  @srcover_skill_create_error
+  Scenario: SkillRepository.create re-raises DuplicateSkillError from within try block
+    Given a skill repository with mock session raising DuplicateSkillError on flush for srcover
+    When a skill is created through the srcover repository and a duplicate error is expected
+    Then a srcover DuplicateSkillError should be raised
+
+  # ===========================================================================
+  # SkillRepository.get_by_name – OperationalError handler (lines 4110-4111)
+  # ===========================================================================
+
+  @srcover_skill_get_error
+  Scenario: SkillRepository.get_by_name raises DatabaseError on OperationalError
+    Given a skill repository with mock session raising OperationalError on query for srcover
+    When a skill is fetched by name through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to get skill"
+
+  # ===========================================================================
+  # SkillRepository.list_all – OperationalError handler (lines 4133-4134)
+  # ===========================================================================
+
+  @srcover_skill_list_error
+  Scenario: SkillRepository.list_all raises DatabaseError on OperationalError
+    Given a skill repository with mock session raising OperationalError on query for srcover
+    When skills are listed through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to list skills"
+
+  # ===========================================================================
+  # SkillRepository.update – OperationalError handler (lines 4192-4194)
+  # ===========================================================================
+
+  @srcover_skill_update_error
+  Scenario: SkillRepository.update raises DatabaseError on OperationalError
+    Given a skill repository with mock session raising OperationalError on query for srcover
+    When a skill is updated through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to update skill"
+
+  # ===========================================================================
+  # SkillRepository.update – overrides truthy + child item loop (lines 4168, 4178)
+  # ===========================================================================
+
+  @srcover_skill_update_overrides
+  Scenario: SkillRepository.update persists overrides and replaces child items
+    Given a clean in-memory skill repository for srcover update tests
+    And a skill "local/upd-test" is registered with tool refs for srcover
+    When the skill "local/upd-test" is updated with overrides and new items for srcover
+    Then the srcover updated skill should have overrides metadata
+    And the srcover updated skill should have the new items
+
+  # ===========================================================================
+  # SkillRepository.delete – OperationalError handler (lines 4219-4221)
+  # ===========================================================================
+
+  @srcover_skill_delete_error
+  Scenario: SkillRepository.delete raises DatabaseError on OperationalError
+    Given a skill repository with mock session raising OperationalError on query for srcover
+    When a skill is deleted through the srcover repository and an error is expected
+    Then a srcover DatabaseError should be raised containing "Failed to delete skill"
+
+  # ===========================================================================
+  # ResourceTypeRepository.create – DuplicateResourceTypeError (line 1711)
+  # ===========================================================================
+
+  @srcover_rt_duplicate
+  Scenario: ResourceTypeRepository.create raises DuplicateResourceTypeError on UNIQUE constraint
+    Given a resource type repository with mock session raising UNIQUE IntegrityError for srcover
+    When a resource type is created through the srcover repository and an error is expected
+    Then a srcover DuplicateResourceTypeError should be raised

features/steps/skill_registry_steps.py

@@ -0,0 +1,552 @@
+"""Step definitions for Skill Registry persistence coverage.
+
+Covers skills, skill_items tables and the SkillRegistryService
+integration layer.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event, text
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.skill_registry_service import (
+    SkillRegistryService,
+)
+from cleveragents.domain.models.core.skill import (
+    Skill,
+    SkillAgentSource,
+    SkillInclude,
+    SkillInlineTool,
+    SkillMcpSource,
+)
+from cleveragents.domain.models.core.tool import ToolSource
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    DuplicateSkillError,
+    SkillNotFoundError,
+    SkillRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_session_factory(context: Context) -> sessionmaker[Session]:
+    return context.skill_session_factory  # type: ignore[no-any-return]
+
+
+def _commit_pending(context: Context) -> None:
+    """Commit pending transaction on the shared SQLite :memory: connection.
+
+    The repository creates its own session internally, so the flushed
+    data lives on the underlying connection's pending transaction.  We
+    must bind a session to that connection and commit to make the data
+    durable across subsequent sessions.
+    """
+    session = _get_session_factory(context)()
+    try:
+        session.execute(text("SELECT 1"))
+        session.commit()
+    finally:
+        session.close()
+
+
+def _make_skill(
+    name: str = "local/test-skill",
+    description: str = "Test skill",
+    tool_refs: list[str] | None = None,
+    includes: list[SkillInclude] | None = None,
+    anonymous_tools: list[SkillInlineTool] | None = None,
+    mcp_servers: list[SkillMcpSource] | None = None,
+    agent_skills: list[SkillAgentSource] | None = None,
+    overrides: dict[str, dict[str, Any]] | None = None,
+) -> Skill:
+    """Create a minimal valid Skill domain object for testing."""
+    return Skill(
+        name=name,
+        description=description,
+        tool_refs=tool_refs or [],
+        includes=includes or [],
+        anonymous_tools=anonymous_tools or [],
+        mcp_servers=mcp_servers or [],
+        agent_skills=agent_skills or [],
+        overrides=overrides or {},
+    )
+
+
+def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
+    """Enable SQLite foreign key enforcement per connection (SEC-3)."""
+    cursor = dbapi_conn.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+# ---------------------------------------------------------------------------
+# Background / setup steps
+# ---------------------------------------------------------------------------
+
+
+@given("a clean in-memory database with the skill registry schema")
+def step_clean_db(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    # SEC-3: Enable FK enforcement so CASCADE works at the DB level
+    event.listen(engine, "connect", _enable_fk_pragma)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    context.skill_engine = engine
+    context.skill_session_factory = factory
+    context.skill_error = None
+    context.skill_result = None
+    context.skill_removal_result = None
+
+
+@given("a skill registry repository backed by the database")
+def step_repo(context: Context) -> None:
+    factory = _get_session_factory(context)
+    context.skill_repo = SkillRepository(session_factory=factory)
+
+
+@given("a skill registry service backed by the repository")
+def step_service(context: Context) -> None:
+    context.skill_service = SkillRegistryService(
+        skill_repo=context.skill_repo,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given: Skill construction
+# ---------------------------------------------------------------------------
+
+
+@given('a valid skill named "{name}" with description "{description}"')
+def step_valid_skill(context: Context, name: str, description: str) -> None:
+    context.current_skill = _make_skill(name=name, description=description)
+
+
+@given('the skill has tool refs "{refs_csv}"')
+def step_skill_tool_refs(context: Context, refs_csv: str) -> None:
+    refs = [r.strip() for r in refs_csv.split(",")]
+    skill: Skill = context.current_skill
+    context.current_skill = skill.model_copy(update={"tool_refs": refs})
+
+
+@given('the skill includes "{includes_csv}"')
+def step_skill_includes(context: Context, includes_csv: str) -> None:
+    includes = [SkillInclude(name=n.strip()) for n in includes_csv.split(",")]
+    skill: Skill = context.current_skill
+    context.current_skill = skill.model_copy(update={"includes": includes})
+
+
+@given('the skill has an inline tool with description "{desc}"')
+def step_skill_inline_tool(context: Context, desc: str) -> None:
+    tool = SkillInlineTool(description=desc, source=ToolSource.CUSTOM, timeout=300)
+    skill: Skill = context.current_skill
+    context.current_skill = skill.model_copy(update={"anonymous_tools": [tool]})
+
+
+@given('the skill has an MCP source "{server}"')
+def step_skill_mcp(context: Context, server: str) -> None:
+    mcp = SkillMcpSource(server=server)
+    skill: Skill = context.current_skill
+    context.current_skill = skill.model_copy(update={"mcp_servers": [mcp]})
+
+
+@given('the skill has an agent source "{path}"')
+def step_skill_agent(context: Context, path: str) -> None:
+    agent = SkillAgentSource(path=path)
+    skill: Skill = context.current_skill
+    context.current_skill = skill.model_copy(update={"agent_skills": [agent]})
+
+
+@given("the skill has already been registered once in the skill registry")
+def step_skill_already_registered(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    svc.add_skill(context.current_skill)
+    _commit_pending(context)
+
+
+@given("the following skills are registered:")
+def step_register_skills_table(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    for row in context.table:
+        skill = _make_skill(
+            name=row["name"],
+            description=row["description"],
+        )
+        svc.add_skill(skill)
+    _commit_pending(context)
+
+
+@given('a valid skill named "{name}" with overrides for "{tool_name}"')
+def step_skill_with_overrides(context: Context, name: str, tool_name: str) -> None:
+    context.current_skill = _make_skill(
+        name=name,
+        description="Skill with overrides",
+        tool_refs=[tool_name],
+        overrides={tool_name: {"timeout": 60, "retries": 3}},
+    )
+
+
+@given('a valid skill named "{name}" with version "{version}"')
+def step_skill_with_version(context: Context, name: str, version: str) -> None:
+    # Skill domain model has no version field; store in context for DB check
+    context.current_skill = _make_skill(name=name, description="Versioned skill")
+    context.skill_expected_version = version
+
+
+@given('a valid skill named "{name}" with mixed item types')
+def step_skill_mixed_items(context: Context, name: str) -> None:
+    context.current_skill = _make_skill(
+        name=name,
+        description="Mixed items skill",
+        tool_refs=["local/tool-a", "local/tool-b"],
+        includes=[SkillInclude(name="local/base-skill")],
+        agent_skills=[SkillAgentSource(path="/skills/my-agent")],
+    )
+
+
+@given('a valid skill named "{name}" with {count:d} tool refs')
+def step_skill_many_refs(context: Context, name: str, count: int) -> None:
+    refs = [f"local/tool-{i:04d}" for i in range(count)]
+    context.current_skill = _make_skill(
+        name=name,
+        description=f"Skill with {count} tool refs",
+        tool_refs=refs,
+    )
+
+
+# ---------------------------------------------------------------------------
+# When: Actions
+# ---------------------------------------------------------------------------
+
+
+@when("the skill is registered through the skill registry service")
+@given("the skill is registered through the skill registry service")
+def step_register_skill(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    try:
+        context.skill_result = svc.add_skill(context.current_skill)
+        _commit_pending(context)
+        context.skill_error = None
+    except Exception as exc:
+        context.skill_error = exc
+
+
+@when('a second skill with the same name "{name}" is registered')
+def step_register_duplicate(context: Context, name: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    dup_skill = _make_skill(name=name, description="Duplicate")
+    try:
+        svc.add_skill(dup_skill)
+        _commit_pending(context)
+        context.skill_error = None
+    except DuplicateSkillError as exc:
+        context.skill_error = exc
+    except Exception as exc:
+        # tenacity.RetryError may wrap the real exception; unwrap.
+        cause = exc.__cause__ or exc
+        context.skill_error = cause
+
+
+@when('I retrieve skill "{name}" from the skill registry')
+@then('I retrieve skill "{name}" from the skill registry')
+def step_retrieve_skill(context: Context, name: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    context.skill_result = svc.get_skill(name)
+
+
+@when("I list all skills from the skill registry")
+def step_list_all_skills(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    context.skill_list_result = svc.list_skills()
+
+
+@when('I list skills from the skill registry with namespace "{ns}"')
+def step_list_skills_ns(context: Context, ns: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    context.skill_list_result = svc.list_skills(namespace=ns)
+
+
+@when('I update skill "{name}" with description "{desc}"')
+def step_update_skill(context: Context, name: str, desc: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    updated_skill = _make_skill(name=name, description=desc)
+    try:
+        svc.update_skill(updated_skill)
+        _commit_pending(context)
+        context.skill_result = svc.get_skill(name)
+        context.skill_error = None
+    except Exception as exc:
+        context.skill_error = exc
+
+
+@when('I try to update a non-existent skill "{name}"')
+def step_update_nonexistent(context: Context, name: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    fake_skill = _make_skill(name=name, description="Ghost")
+    try:
+        svc.update_skill(fake_skill)
+        _commit_pending(context)
+        context.skill_error = None
+    except SkillNotFoundError as exc:
+        context.skill_error = exc
+    except Exception as exc:
+        # tenacity.RetryError may wrap the real exception; unwrap.
+        cause = exc.__cause__ or exc
+        context.skill_error = cause
+
+
+@when('I remove skill "{name}" from the skill registry')
+def step_remove_skill(context: Context, name: str) -> None:
+    svc: SkillRegistryService = context.skill_service
+    try:
+        context.skill_removal_result = svc.remove_skill(name)
+        _commit_pending(context)
+    except Exception as exc:
+        context.skill_error = exc
+
+
+@when('I register a skill with invalid name "{name}" via dict bypass')
+def step_register_invalid_name(context: Context, name: str) -> None:
+    """Attempt to register a skill with an invalid name through from_domain."""
+    from cleveragents.infrastructure.database.models import SkillModel
+
+    try:
+        SkillModel.from_domain({"name": name, "description": "Invalid"})
+        context.skill_error = None
+    except ValueError as exc:
+        context.skill_error = exc
+
+
+@when("I register a skill with empty name via dict bypass")
+def step_register_empty_name(context: Context) -> None:
+    """Attempt to register a skill with an empty name through from_domain."""
+    from cleveragents.infrastructure.database.models import SkillModel
+
+    try:
+        SkillModel.from_domain({"name": "", "description": "Invalid"})
+        context.skill_error = None
+    except ValueError as exc:
+        context.skill_error = exc
+
+
+@when("the skill is registered and retrieved with overrides check")
+def step_register_and_check_overrides(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    try:
+        svc.add_skill(context.current_skill)
+        _commit_pending(context)
+        context.skill_result = svc.get_skill(context.current_skill.name)
+        context.skill_error = None
+    except Exception as exc:
+        context.skill_error = exc
+
+
+@when("I retrieve the skill and check item ordering")
+def step_check_item_ordering(context: Context) -> None:
+    svc: SkillRegistryService = context.skill_service
+    context.skill_result = svc.get_skill(context.current_skill.name)
+
+
+@when("the skill is registered with a transient DB error then succeeds")
+def step_register_with_retry(context: Context) -> None:
+    """Simulate a transient DB error on first attempt, success on second."""
+    from sqlalchemy.exc import OperationalError as SqlaOpError
+
+    svc: SkillRegistryService = context.skill_service
+    repo: SkillRepository = context.skill_repo
+
+    call_count = 0
+    original_session = repo._session
+
+    def flaky_session() -> Session:
+        nonlocal call_count
+        call_count += 1
+        session = original_session()
+        if call_count == 1:
+            # Simulate a transient DB error on first call
+            raise SqlaOpError(
+                "database is locked", params=None, orig=Exception("locked")
+            )
+        return session
+
+    repo._session = flaky_session  # type: ignore[assignment]
+    try:
+        context.skill_result = svc.add_skill(context.current_skill)
+        _commit_pending(context)
+        context.skill_error = None
+    except Exception as exc:
+        context.skill_error = exc
+    finally:
+        repo._session = original_session  # type: ignore[assignment]
+    context.retry_call_count = call_count
+
+
+# ---------------------------------------------------------------------------
+# Then: Assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the skill registry service should not raise an error")
+def step_no_error(context: Context) -> None:
+    assert context.skill_error is None, f"Unexpected error: {context.skill_error}"
+
+
+@then('the persisted skill should have the name "{name}"')
+def step_skill_has_name(context: Context, name: str) -> None:
+    result = context.skill_result
+    # If result is None, re-fetch from service
+    if result is None:
+        svc: SkillRegistryService = context.skill_service
+        result = svc.get_skill(name)
+        context.skill_result = result
+    assert result is not None, f"Skill '{name}' not found"
+    assert result.name == name, f"Expected name {name}, got {result.name}"
+
+
+@then('the persisted skill description should be "{desc}"')
+def step_skill_description(context: Context, desc: str) -> None:
+    result = context.skill_result
+    assert result is not None, "No skill result available"
+    assert result.description == desc, (
+        f"Expected description '{desc}', got '{result.description}'"
+    )
+
+
+@then('the persisted skill should have {count:d} items of type "{item_type}"')
+def step_skill_items_count(context: Context, count: int, item_type: str) -> None:
+    # Re-fetch to get the full reconstructed Skill
+    skill = context.current_skill
+    svc: SkillRegistryService = context.skill_service
+    result = svc.get_skill(skill.name)
+    assert result is not None, f"Skill '{skill.name}' not found"
+
+    # Map item_type to the corresponding Skill field
+    type_to_field: dict[str, int] = {
+        "tool_ref": len(result.tool_refs),
+        "include": len(result.includes),
+        "inline_tool": len(result.anonymous_tools),
+        "mcp_source": len(result.mcp_servers),
+        "agent_source": len(result.agent_skills),
+    }
+    actual = type_to_field.get(item_type, 0)
+    assert actual == count, (
+        f"Expected {count} items of type '{item_type}', got {actual}"
+    )
+
+
+@then('a DuplicateSkillError should be raised mentioning "{name}"')
+def step_duplicate_error(context: Context, name: str) -> None:
+    assert context.skill_error is not None, "Expected DuplicateSkillError"
+    assert isinstance(context.skill_error, DuplicateSkillError), (
+        f"Expected DuplicateSkillError, got {type(context.skill_error)}"
+    )
+    assert name in str(context.skill_error), (
+        f"Expected '{name}' in error message: {context.skill_error}"
+    )
+
+
+@then("the skill registry result should be None")
+def step_result_none(context: Context) -> None:
+    assert context.skill_result is None, f"Expected None, got {context.skill_result}"
+
+
+@then("the skill list should contain {count:d} skills")
+def step_list_count(context: Context, count: int) -> None:
+    result_list: list[Skill] = context.skill_list_result
+    assert len(result_list) == count, f"Expected {count} skills, got {len(result_list)}"
+
+
+@then("a SkillNotFoundError should be raised")
+def step_not_found_error(context: Context) -> None:
+    assert context.skill_error is not None, "Expected SkillNotFoundError"
+    assert isinstance(context.skill_error, SkillNotFoundError), (
+        f"Expected SkillNotFoundError, got {type(context.skill_error)}"
+    )
+
+
+@then("the skill removal should return True")
+def step_removal_true(context: Context) -> None:
+    assert context.skill_removal_result is True, (
+        f"Expected True, got {context.skill_removal_result}"
+    )
+
+
+@then("the skill removal should return False")
+def step_removal_false(context: Context) -> None:
+    assert context.skill_removal_result is False, (
+        f"Expected False, got {context.skill_removal_result}"
+    )
+
+
+@then("a ValueError should be raised about invalid skill name")
+def step_invalid_name_error(context: Context) -> None:
+    assert context.skill_error is not None, "Expected ValueError"
+    assert isinstance(context.skill_error, ValueError), (
+        f"Expected ValueError, got {type(context.skill_error)}"
+    )
+    assert (
+        "namespace/short_name" in str(context.skill_error).lower()
+        or "pattern" in str(context.skill_error).lower()
+    ), f"Error should mention name format: {context.skill_error}"
+
+
+@then('the retrieved skill should have overrides for "{tool_name}"')
+def step_check_overrides(context: Context, tool_name: str) -> None:
+    result: Skill = context.skill_result
+    assert result is not None, "No skill result available"
+    assert tool_name in result.overrides, (
+        f"Expected overrides for '{tool_name}', got keys: {list(result.overrides.keys())}"
+    )
+    assert result.overrides[tool_name]["timeout"] == 60
+    assert result.overrides[tool_name]["retries"] == 3
+
+
+@then("the retrieved skill should have items in stable order")
+def step_check_item_order(context: Context) -> None:
+    result: Skill = context.skill_result
+    assert result is not None, "No skill result available"
+    # Original was: 2 tool_refs, 1 include, 1 agent_source
+    assert len(result.tool_refs) == 2, (
+        f"Expected 2 tool_refs, got {len(result.tool_refs)}"
+    )
+    assert result.tool_refs[0] == "local/tool-a"
+    assert result.tool_refs[1] == "local/tool-b"
+    assert len(result.includes) == 1, f"Expected 1 include, got {len(result.includes)}"
+    assert result.includes[0].name == "local/base-skill"
+    assert len(result.agent_skills) == 1, (
+        f"Expected 1 agent_skill, got {len(result.agent_skills)}"
+    )
+    assert result.agent_skills[0].path == "/skills/my-agent"
+
+
+@then("the skill list entries should all be Skill domain objects")
+def step_list_domain_objects(context: Context) -> None:
+    result_list: list[Skill] = context.skill_list_result
+    for item in result_list:
+        assert isinstance(item, Skill), f"Expected Skill instance, got {type(item)}"
+
+
+@then("the persisted skill should have {count:d} tool refs")
+def step_skill_tool_ref_count(context: Context, count: int) -> None:
+    skill = context.current_skill
+    svc: SkillRegistryService = context.skill_service
+    result = svc.get_skill(skill.name)
+    assert result is not None, f"Skill '{skill.name}' not found"
+    assert len(result.tool_refs) == count, (
+        f"Expected {count} tool refs, got {len(result.tool_refs)}"
+    )
+
+
+@then("the retry should have been attempted")
+def step_retry_attempted(context: Context) -> None:
+    # The @database_retry decorator retries on OperationalError,
+    # so the call count should be > 1 (first attempt raised, second succeeded)
+    assert context.retry_call_count >= 1, (
+        f"Expected retry, but call_count was {context.retry_call_count}"
+    )

features/steps/skill_repo_coverage_steps.py

@@ -0,0 +1,713 @@
+"""Step definitions for skill_repo_coverage.feature.
+
+Targets the remaining ~32 uncovered lines in repositories.py, focusing on
+SkillRepository error handlers, SkillInUseError, ToolRepository edge cases,
+ValidationAttachmentRepository branches, and ResourceRepository auto-discover.
+
+All step patterns are prefixed with "srcover" to avoid AmbiguousStep collisions.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event, text
+from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.core.exceptions import DatabaseError
+from cleveragents.domain.models.core.skill import Skill
+from cleveragents.infrastructure.database.models import (
+    Base,
+    ResourceModel,
+    ResourceTypeModel,
+)
+from cleveragents.infrastructure.database.repositories import (
+    DuplicateResourceTypeError,
+    DuplicateSkillError,
+    ResourceRepository,
+    ResourceTypeRepository,
+    SkillInUseError,
+    SkillRepository,
+    ToolRepository,
+    ValidationAttachmentRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
+    """Enable SQLite foreign key enforcement."""
+    cursor = dbapi_conn.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+def _capture_srcover_error(
+    context: Context, fn: Any, *args: Any, **kwargs: Any
+) -> None:
+    """Call *fn* and capture any error to ``context.srcover_error``."""
+    try:
+        context.srcover_result = fn(*args, **kwargs)
+    except (DatabaseError, DuplicateSkillError, DuplicateResourceTypeError) as exc:
+        context.srcover_error = exc
+    except Exception as exc:
+        # tenacity RetryError wraps the real cause
+        cause = getattr(exc, "__cause__", None) or exc
+        context.srcover_error = cause
+
+
+def _make_skill(
+    name: str = "local/test-skill",
+    description: str = "Test skill",
+    tool_refs: list[str] | None = None,
+    overrides: dict[str, dict[str, Any]] | None = None,
+) -> Skill:
+    """Create a minimal Skill domain object."""
+
+    return Skill(
+        name=name,
+        description=description,
+        tool_refs=tool_refs or [],
+        includes=[],
+        anonymous_tools=[],
+        mcp_servers=[],
+        agent_skills=[],
+        overrides=overrides or {},
+    )
+
+
+def _mock_session_op_error_on_flush() -> MagicMock:
+    """Session whose ``.flush()`` raises OperationalError."""
+    mock = MagicMock()
+    mock.add.return_value = None
+    mock.flush.side_effect = OperationalError("disk I/O error", {}, None)
+    mock.rollback.return_value = None
+    query_mock = MagicMock()
+    query_mock.filter_by.return_value.first.return_value = None
+    query_mock.filter.return_value.first.return_value = None
+    mock.query.return_value = query_mock
+    return mock
+
+
+def _mock_session_op_error_on_query() -> MagicMock:
+    """Session whose ``.query()`` always raises OperationalError."""
+    mock = MagicMock()
+    mock.query.side_effect = OperationalError("connection lost", {}, None)
+    mock.rollback.return_value = None
+    return mock
+
+
+def _now_iso() -> str:
+    from datetime import UTC, datetime
+
+    return datetime.now(tz=UTC).isoformat()
+
+
+# ===========================================================================
+# SkillInUseError (lines 4027-4031)
+# ===========================================================================
+
+
+@when('a SkillInUseError is created for skill "{name}" without detail')
+def step_skill_in_use_no_detail(context: Context, name: str) -> None:
+    context.srcover_error = SkillInUseError(name)
+
+
+@when('a SkillInUseError is created for skill "{name}" with detail "{detail}"')
+def step_skill_in_use_with_detail(context: Context, name: str, detail: str) -> None:
+    context.srcover_error = SkillInUseError(name, detail=detail)
+
+
+@then('the SkillInUseError message should contain "{fragment}"')
+def step_skill_in_use_message(context: Context, fragment: str) -> None:
+    msg = str(context.srcover_error)
+    assert fragment in msg, f"Expected '{fragment}' in message: {msg}"
+
+
+@then('the SkillInUseError skill_name should be "{name}"')
+def step_skill_in_use_skill_name(context: Context, name: str) -> None:
+    assert context.srcover_error.skill_name == name, (
+        f"Expected skill_name '{name}', got '{context.srcover_error.skill_name}'"
+    )
+
+
+# ===========================================================================
+# ToolRepository.__init__ TypeError (line 3387)
+# ===========================================================================
+
+
+@when("a ToolRepository is created without any session factory")
+def step_tool_repo_no_factory(context: Context) -> None:
+    context.srcover_error = None
+    try:
+        ToolRepository()  # type: ignore[call-arg]
+    except TypeError as exc:
+        context.srcover_error = exc
+
+
+@then("a TypeError should be raised about session_factory requirement")
+def step_tool_repo_type_error(context: Context) -> None:
+    assert context.srcover_error is not None, "Expected TypeError but none raised"
+    assert isinstance(context.srcover_error, TypeError), (
+        f"Expected TypeError, got {type(context.srcover_error).__name__}"
+    )
+    assert (
+        "session_factory" in str(context.srcover_error).lower()
+        or "factory" in str(context.srcover_error).lower()
+    ), f"Expected 'session_factory' in message: {context.srcover_error}"
+
+
+# ===========================================================================
+# ToolRepository._extract_value enum unwrapping (line 3410)
+# ===========================================================================
+
+
+@when('_extract_value is called with an enum-like attribute having value "{val}"')
+def step_extract_value_enum(context: Context, val: str) -> None:
+    # Line 3409 checks ``"value" in raw.__class__.__dict__``.
+    # Python stdlib Enum puts ``value`` in the *base* Enum class, not the
+    # concrete subclass, so we need a custom class whose ``__class__.__dict__``
+    # contains ``"value"`` directly (a class attribute, not instance attribute).
+    class _EnumLike:
+        value: str = val  # class-level attribute -> in __class__.__dict__
+
+    obj = SimpleNamespace(status=_EnumLike())
+    context.srcover_result = ToolRepository._extract_value(obj, "status")
+
+
+@then('the extracted value should be "{val}"')
+def step_extracted_value(context: Context, val: str) -> None:
+    assert context.srcover_result == val, (
+        f"Expected '{val}', got '{context.srcover_result}'"
+    )
+
+
+# ===========================================================================
+# ToolRepository.remove success (line 3514)
+# ===========================================================================
+
+
+@given("a tool repository with a real in-memory database for srcover")
+def step_tool_repo_real_db(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    event.listen(engine, "connect", _enable_fk_pragma)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    context.srcover_tool_repo = ToolRepository(session_factory=factory)
+    context.srcover_tool_factory = factory
+    context.srcover_error = None
+
+
+@given('a tool "{name}" exists in the srcover tool repository')
+def step_tool_exists(context: Context, name: str) -> None:
+    tool = SimpleNamespace(
+        name=name,
+        description="A removable tool",
+        tool_type=SimpleNamespace(value="tool"),
+        source=SimpleNamespace(value="builtin"),
+        input_schema=None,
+        output_schema=None,
+        capability=SimpleNamespace(
+            read_only=False, writes=False, checkpointable=False, side_effects=[]
+        ),
+        config_yaml=None,
+        resource_slots=[],
+        tool_id=None,
+    )
+    context.srcover_tool_repo.add(tool)
+
+
+@when('the tool "{name}" is removed via srcover repository')
+def step_tool_remove(context: Context, name: str) -> None:
+    context.srcover_removal_result = context.srcover_tool_repo.remove(name)
+
+
+@then("the srcover tool removal result should be True")
+def step_tool_removal_true(context: Context) -> None:
+    assert context.srcover_removal_result is True, (
+        f"Expected True, got {context.srcover_removal_result}"
+    )
+
+
+# ===========================================================================
+# ValidationAttachmentRepository.create with args (line 3589)
+# ===========================================================================
+
+
+def _ensure_va_resource_srcover(factory: sessionmaker[Session]) -> None:
+    """Create a resource type + resource for VA tests."""
+    session = factory()
+    now = _now_iso()
+    existing = (
+        session.query(ResourceTypeModel).filter_by(name="test/va-srcover").first()
+    )
+    if existing is None:
+        rt = ResourceTypeModel(
+            name="test/va-srcover",
+            namespace="test",
+            description="for srcover VA tests",
+            resource_kind="physical",
+            sandbox_strategy="none",
+            user_addable=True,
+            handler_ref=None,
+            args_schema_json=None,
+            allowed_parent_types_json=None,
+            allowed_child_types_json="[]",
+            auto_discover_json=None,
+            equivalence_json=None,
+            capabilities_json='{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
+            source=None,
+            created_at=now,
+            updated_at=now,
+        )
+        session.add(rt)
+        session.flush()
+    existing_res = (
+        session.query(ResourceModel).filter_by(resource_id="srcover-res-1").first()
+    )
+    if existing_res is None:
+        res = ResourceModel(
+            resource_id="srcover-res-1",
+            namespaced_name=None,
+            namespace=None,
+            type_name="test/va-srcover",
+            resource_kind="physical",
+            location=None,
+            description="srcover test resource",
+            read_only=False,
+            auto_discovered=False,
+            sandbox_strategy=None,
+            content_hash=None,
+            properties_json=None,
+            metadata_json=None,
+            created_at=now,
+            updated_at=now,
+        )
+        session.add(res)
+    session.commit()
+
+
+@given("a validation attachment repository with real database for srcover")
+def step_va_repo_real(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    _ensure_va_resource_srcover(factory)
+    context.srcover_va_repo = ValidationAttachmentRepository(factory)
+    context.srcover_va_factory = factory
+    context.srcover_error = None
+
+
+@when("a validation is attached with args {args_json} for srcover")
+def step_va_attach_with_args(context: Context, args_json: str) -> None:
+    args = json.loads(args_json)
+    context.srcover_va_result = context.srcover_va_repo.attach(
+        "local/test-validation",
+        "srcover-res-1",
+        "required",
+        args=args,
+    )
+
+
+@then('the srcover attachment should have args_json containing "{fragment}"')
+def step_va_args_json(context: Context, fragment: str) -> None:
+    result = context.srcover_va_result
+    assert result is not None, "Attachment result was None"
+    args_json_val = (
+        result.get("args_json")
+        if isinstance(result, dict)
+        else getattr(result, "args_json", None)
+    )
+    assert args_json_val is not None, "args_json should not be None"
+    assert fragment in args_json_val, (
+        f"Expected '{fragment}' in args_json: {args_json_val}"
+    )
+
+
+# ===========================================================================
+# ValidationAttachmentRepository.get_by_id returns None (line 3714)
+# ===========================================================================
+
+
+@when("get_by_id is called with a non-existent attachment ID for srcover")
+def step_va_get_by_id_none(context: Context) -> None:
+    context.srcover_va_get_result = context.srcover_va_repo.get_by_id(
+        "01NONEXISTENT00000000000000"
+    )
+
+
+@then("the srcover get_by_id result should be None")
+def step_va_get_by_id_is_none(context: Context) -> None:
+    assert context.srcover_va_get_result is None, (
+        f"Expected None, got {context.srcover_va_get_result}"
+    )
+
+
+# ===========================================================================
+# ValidationAttachmentRepository.get_by_id OperationalError (lines 3725-3726)
+# ===========================================================================
+
+
+@given(
+    "a validation attachment repository with mock session raising OperationalError for srcover"
+)
+def step_va_repo_mock_op_error(context: Context) -> None:
+    mock = _mock_session_op_error_on_query()
+    context.srcover_va_repo = ValidationAttachmentRepository(
+        session_factory=lambda: mock
+    )
+    context.srcover_error = None
+
+
+@when("get_by_id is called and an OperationalError occurs for srcover")
+def step_va_get_by_id_error(context: Context) -> None:
+    _capture_srcover_error(
+        context, context.srcover_va_repo.get_by_id, "01FAKE0000000000000000000000"
+    )
+
+
+# ===========================================================================
+# ResourceRepository.auto_discover_children - empty type (line 2555)
+# ===========================================================================
+
+
+@given("a resource repository with in-memory database for srcover auto-discover")
+def step_res_repo_real(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    context.srcover_res_factory = factory
+    context.srcover_res_repo = ResourceRepository(session_factory=factory)
+    context.srcover_error = None
+
+
+@given("a resource type with auto_discover rules containing an empty type for srcover")
+def step_res_type_empty_child(context: Context) -> None:
+    session = context.srcover_res_factory()
+    now = _now_iso()
+    rt = ResourceTypeModel(
+        name="test/empty-child-rule",
+        namespace="test",
+        description="has rule with empty type",
+        resource_kind="physical",
+        sandbox_strategy="none",
+        user_addable=True,
+        handler_ref=None,
+        args_schema_json=None,
+        allowed_parent_types_json=None,
+        allowed_child_types_json="[]",
+        auto_discover_json=json.dumps({"enabled": True, "rules": [{"type": ""}, {}]}),
+        equivalence_json=None,
+        capabilities_json='{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
+        source=None,
+        created_at=now,
+        updated_at=now,
+    )
+    session.add(rt)
+    session.commit()
+
+
+@given("a resource of that type exists for srcover auto-discover")
+def step_res_for_auto_discover(context: Context) -> None:
+    session = context.srcover_res_factory()
+    now = _now_iso()
+    res = ResourceModel(
+        resource_id="srcover-auto-disc-001",
+        namespaced_name=None,
+        namespace=None,
+        type_name="test/empty-child-rule",
+        resource_kind="physical",
+        location=None,
+        description="srcover auto-disc test",
+        read_only=False,
+        auto_discovered=False,
+        sandbox_strategy=None,
+        content_hash=None,
+        properties_json=None,
+        metadata_json=None,
+        created_at=now,
+        updated_at=now,
+    )
+    session.add(res)
+    session.commit()
+
+
+@when("auto_discover_children is called for the srcover resource")
+def step_auto_discover_empty_type(context: Context) -> None:
+    context.srcover_auto_disc_result = context.srcover_res_repo.auto_discover_children(
+        "srcover-auto-disc-001"
+    )
+
+
+@then("the srcover auto-discover result should be an empty list")
+def step_auto_discover_empty(context: Context) -> None:
+    assert context.srcover_auto_disc_result == [], (
+        f"Expected empty list, got {context.srcover_auto_disc_result}"
+    )
+
+
+# ===========================================================================
+# SkillRepository.create - OperationalError (lines 4093-4095)
+# ===========================================================================
+
+
+@given(
+    "a skill repository with mock session raising OperationalError on flush for srcover"
+)
+def step_skill_repo_op_error_flush(context: Context) -> None:
+    mock = _mock_session_op_error_on_flush()
+    context.srcover_skill_repo = SkillRepository(session_factory=lambda: mock)
+    context.srcover_error = None
+
+
+@when("a skill is created through the srcover repository and an error is expected")
+def step_skill_create_error(context: Context) -> None:
+    skill = _make_skill(name="local/error-skill")
+    _capture_srcover_error(context, context.srcover_skill_repo.create, skill)
+
+
+# ===========================================================================
+# SkillRepository.create - non-UNIQUE IntegrityError (line 4090)
+# ===========================================================================
+
+
+@given(
+    "a skill repository with mock session raising non-UNIQUE IntegrityError for srcover"
+)
+def step_skill_repo_non_unique_integrity(context: Context) -> None:
+    mock = MagicMock()
+    mock.add.return_value = None
+    mock.flush.side_effect = IntegrityError(
+        "CHECK constraint failed: some_check", {}, None
+    )
+    mock.rollback.return_value = None
+    context.srcover_skill_repo = SkillRepository(session_factory=lambda: mock)
+    context.srcover_error = None
+
+
+# ===========================================================================
+# SkillRepository.create - DuplicateSkillError re-raise (lines 4091-4092)
+# ===========================================================================
+
+
+@given(
+    "a skill repository with mock session raising DuplicateSkillError on flush for srcover"
+)
+def step_skill_repo_dup_on_flush(context: Context) -> None:
+    mock = MagicMock()
+    mock.add.return_value = None
+    mock.flush.side_effect = DuplicateSkillError("local/dup-skill")
+    mock.rollback.return_value = None
+    context.srcover_skill_repo = SkillRepository(session_factory=lambda: mock)
+    context.srcover_error = None
+
+
+@when(
+    "a skill is created through the srcover repository and a duplicate error is expected"
+)
+def step_skill_create_dup_error(context: Context) -> None:
+    skill = _make_skill(name="local/dup-skill")
+    _capture_srcover_error(context, context.srcover_skill_repo.create, skill)
+
+
+@then("a srcover DuplicateSkillError should be raised")
+def step_check_dup_skill_error(context: Context) -> None:
+    assert context.srcover_error is not None, (
+        "Expected DuplicateSkillError but none raised"
+    )
+    assert isinstance(context.srcover_error, DuplicateSkillError), (
+        f"Expected DuplicateSkillError, got {type(context.srcover_error).__name__}: "
+        f"{context.srcover_error}"
+    )
+
+
+# ===========================================================================
+# SkillRepository.get_by_name / list_all / update / delete - OperationalError
+# (lines 4110-4111, 4133-4134, 4192-4194, 4219-4221)
+# ===========================================================================
+
+
+@given(
+    "a skill repository with mock session raising OperationalError on query for srcover"
+)
+def step_skill_repo_op_error_query(context: Context) -> None:
+    mock = _mock_session_op_error_on_query()
+    context.srcover_skill_repo = SkillRepository(session_factory=lambda: mock)
+    context.srcover_error = None
+
+
+@when(
+    "a skill is fetched by name through the srcover repository and an error is expected"
+)
+def step_skill_get_error(context: Context) -> None:
+    _capture_srcover_error(context, context.srcover_skill_repo.get_by_name, "local/any")
+
+
+@when("skills are listed through the srcover repository and an error is expected")
+def step_skill_list_error(context: Context) -> None:
+    _capture_srcover_error(context, context.srcover_skill_repo.list_all)
+
+
+@when("a skill is updated through the srcover repository and an error is expected")
+def step_skill_update_error(context: Context) -> None:
+    skill = _make_skill(name="local/update-err")
+    _capture_srcover_error(context, context.srcover_skill_repo.update, skill)
+
+
+@when("a skill is deleted through the srcover repository and an error is expected")
+def step_skill_delete_error(context: Context) -> None:
+    _capture_srcover_error(context, context.srcover_skill_repo.delete, "local/del-err")
+
+
+# ===========================================================================
+# SkillRepository.update - overrides + child items (lines 4168, 4178)
+# ===========================================================================
+
+
+@given("a clean in-memory skill repository for srcover update tests")
+def step_clean_skill_repo(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    event.listen(engine, "connect", _enable_fk_pragma)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    context.srcover_skill_repo = SkillRepository(session_factory=factory)
+    context.srcover_skill_factory = factory
+    context.srcover_error = None
+
+
+def _commit_srcover(context: Context) -> None:
+    """Commit pending transaction on the shared in-memory connection."""
+    session = context.srcover_skill_factory()
+    try:
+        session.execute(text("SELECT 1"))
+        session.commit()
+    finally:
+        session.close()
+
+
+@given('a skill "{name}" is registered with tool refs for srcover')
+def step_register_skill_with_refs(context: Context, name: str) -> None:
+    skill = _make_skill(
+        name=name,
+        description="Original skill",
+        tool_refs=["local/tool-a", "local/tool-b"],
+    )
+    context.srcover_skill_repo.create(skill)
+    _commit_srcover(context)
+
+
+@when('the skill "{name}" is updated with overrides and new items for srcover')
+def step_update_with_overrides(context: Context, name: str) -> None:
+    updated_skill = _make_skill(
+        name=name,
+        description="Updated skill",
+        tool_refs=["local/tool-c", "local/tool-d", "local/tool-e"],
+        overrides={"local/tool-c": {"timeout": 30}},
+    )
+    context.srcover_skill_repo.update(updated_skill)
+    _commit_srcover(context)
+    # Re-fetch to verify
+    context.srcover_updated_skill = context.srcover_skill_repo.get_by_name(name)
+
+
+@then("the srcover updated skill should have overrides metadata")
+def step_check_overrides_metadata(context: Context) -> None:
+    skill = context.srcover_updated_skill
+    assert skill is not None, "Updated skill was None"
+    # The overrides are stored in metadata_json and reconstructed
+    assert skill.overrides is not None, "Overrides should not be None"
+    assert "local/tool-c" in skill.overrides, (
+        f"Expected 'local/tool-c' in overrides, got: {skill.overrides}"
+    )
+
+
+@then("the srcover updated skill should have the new items")
+def step_check_new_items(context: Context) -> None:
+    skill = context.srcover_updated_skill
+    assert skill is not None, "Updated skill was None"
+    assert len(skill.tool_refs) == 3, (
+        f"Expected 3 tool refs, got {len(skill.tool_refs)}"
+    )
+    assert "local/tool-c" in skill.tool_refs
+
+
+# ===========================================================================
+# Common assertion steps
+# ===========================================================================
+
+
+@then('a srcover DatabaseError should be raised containing "{fragment}"')
+def step_check_db_error(context: Context, fragment: str) -> None:
+    assert context.srcover_error is not None, (
+        f"Expected a DatabaseError containing '{fragment}' but no error was raised"
+    )
+    assert isinstance(context.srcover_error, DatabaseError), (
+        f"Expected DatabaseError, got {type(context.srcover_error).__name__}: "
+        f"{context.srcover_error}"
+    )
+    assert fragment in str(context.srcover_error), (
+        f"Expected '{fragment}' in error message, got: {context.srcover_error}"
+    )
+
+
+# ===========================================================================
+# ResourceTypeRepository.create - DuplicateResourceTypeError (line 1711)
+# ===========================================================================
+
+
+@given(
+    "a resource type repository with mock session raising UNIQUE IntegrityError for srcover"
+)
+def step_rt_repo_unique_integrity(context: Context) -> None:
+    mock = MagicMock()
+    mock.add.return_value = None
+    mock.flush.side_effect = IntegrityError(
+        "UNIQUE constraint failed: resource_types.name", {}, None
+    )
+    mock.rollback.return_value = None
+    query_mock = MagicMock()
+    query_mock.filter_by.return_value.first.return_value = None
+    mock.query.return_value = query_mock
+    context.srcover_rt_repo = ResourceTypeRepository(session_factory=lambda: mock)
+    context.srcover_error = None
+
+
+@when(
+    "a resource type is created through the srcover repository and an error is expected"
+)
+def step_rt_create_error(context: Context) -> None:
+    fake_rt = SimpleNamespace(
+        name="test/dup-type",
+        description="A duplicate type",
+        resource_kind=SimpleNamespace(value="physical"),
+        sandbox_strategy=SimpleNamespace(value="none"),
+        user_addable=True,
+        handler=None,
+        cli_args=[],
+        parent_types=[],
+        child_types=[],
+        auto_discovery=None,
+        equivalence=None,
+        capabilities=None,
+        source=None,
+    )
+    _capture_srcover_error(context, context.srcover_rt_repo.create, fake_rt)
+
+
+@then("a srcover DuplicateResourceTypeError should be raised")
+def step_check_dup_rt_error(context: Context) -> None:
+    assert context.srcover_error is not None, (
+        "Expected DuplicateResourceTypeError but none raised"
+    )
+    assert isinstance(context.srcover_error, DuplicateResourceTypeError), (
+        f"Expected DuplicateResourceTypeError, got "
+        f"{type(context.srcover_error).__name__}: {context.srcover_error}"
+    )

implementation_plan.md

@@ -3363,7 +3363,7 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [X] Git [Brent]: `git push -u origin feature/m3-skill-domain-robot`
   - [X] Forgejo PR [Brent]: Open PR from `feature/m3-skill-domain-robot` to `master` with a suitable and thorough description. — merged to master via develop-brent-2 PR #132 on 2026-02-20
   - [X] Quality [Brent]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
-- [X] **COMMIT (Owner: Luis | Group: C0.skill.registry | Branch: feature/m3-skill-registry | Done: commit not found) - Commit message: "feat(skill): add skill registry persistence"**
+- [X] **COMMIT (Owner: Luis | Group: C0.skill.registry | Branch: feature/m3-skill-registry | Done: Day 20, February 19, 2026) - Commit message: "feat(skill): add skill registry persistence"**
   - [X] Git [Luis]: `git checkout master`
   - [X] Git [Luis]: `git pull origin master`
   - [X] Git [Luis]: `git checkout -b feature/m3-skill-registry`

robot/helper_skill_registry.py

@@ -0,0 +1,162 @@
+"""Helper script for Robot Framework skill registry smoke tests.
+
+Usage:
+    python helper_skill_registry.py <command>
+
+Commands:
+    register-and-get     Register a skill and retrieve it
+    list-filter          Register skills and list with namespace filter
+    update-skill         Register and update a skill
+    duplicate-reject     Verify duplicate skill names are rejected
+    remove-skill         Register and remove a skill
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure src is on the path
+src_dir = str(Path(__file__).resolve().parents[1] / "src")
+if src_dir not in sys.path:
+    sys.path.insert(0, src_dir)
+
+from sqlalchemy import create_engine, event, text  # noqa: E402
+from sqlalchemy.orm import Session, sessionmaker  # noqa: E402
+
+from cleveragents.application.services.skill_registry_service import (  # noqa: E402
+    SkillRegistryService,
+)
+from cleveragents.domain.models.core.skill import Skill  # noqa: E402
+from cleveragents.infrastructure.database.models import Base  # noqa: E402
+from cleveragents.infrastructure.database.repositories import (  # noqa: E402
+    DuplicateSkillError,
+    SkillRepository,
+)
+
+
+def _enable_fk_pragma(dbapi_conn, _connection_record):  # type: ignore[no-untyped-def]
+    """Enable SQLite foreign key enforcement per connection."""
+    cursor = dbapi_conn.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+def _setup() -> tuple[SkillRegistryService, sessionmaker[Session]]:
+    """Create an in-memory DB and return the service + session factory."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    event.listen(engine, "connect", _enable_fk_pragma)
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    repo = SkillRepository(session_factory=factory)
+    svc = SkillRegistryService(skill_repo=repo)
+    return svc, factory
+
+
+def _commit(factory: sessionmaker[Session]) -> None:
+    """Commit the pending transaction on the shared connection."""
+    session = factory()
+    try:
+        session.execute(text("SELECT 1"))
+        session.commit()
+    finally:
+        session.close()
+
+
+def _make_skill(name: str, description: str = "Test skill") -> Skill:
+    return Skill(name=name, description=description)
+
+
+def cmd_register_and_get() -> None:
+    svc, factory = _setup()
+    skill = _make_skill("local/test-tool", "A test skill")
+    svc.add_skill(skill)
+    _commit(factory)
+
+    result = svc.get_skill("local/test-tool")
+    assert result is not None, "Skill not found"
+    assert result.name == "local/test-tool"
+    print("skill-registry-ok")
+
+
+def cmd_list_filter() -> None:
+    svc, factory = _setup()
+    svc.add_skill(_make_skill("local/tool-a", "Tool A"))
+    svc.add_skill(_make_skill("local/tool-b", "Tool B"))
+    svc.add_skill(_make_skill("devops/deploy", "Deploy"))
+    _commit(factory)
+
+    all_skills = svc.list_skills()
+    assert len(all_skills) == 3
+
+    local_skills = svc.list_skills(namespace="local")
+    assert len(local_skills) == 2
+
+    print("skill-list-ok")
+
+
+def cmd_update_skill() -> None:
+    svc, factory = _setup()
+    svc.add_skill(_make_skill("local/updatable", "Original"))
+    _commit(factory)
+
+    updated = _make_skill("local/updatable", "Updated desc")
+    svc.update_skill(updated)
+    _commit(factory)
+
+    result = svc.get_skill("local/updatable")
+    assert result is not None
+    assert result.description == "Updated desc"
+    print("skill-update-ok")
+
+
+def cmd_duplicate_reject() -> None:
+    svc, factory = _setup()
+    svc.add_skill(_make_skill("local/dup-test", "First"))
+    _commit(factory)
+
+    try:
+        svc.add_skill(_make_skill("local/dup-test", "Second"))
+        _commit(factory)
+        print("FAIL: should have raised DuplicateSkillError")
+        sys.exit(1)
+    except DuplicateSkillError:
+        print("skill-duplicate-reject-ok")
+    except Exception as exc:
+        # tenacity may wrap the error; unwrap if needed
+        cause = exc.__cause__ or exc
+        if isinstance(cause, DuplicateSkillError):
+            print("skill-duplicate-reject-ok")
+        else:
+            print(f"FAIL: unexpected {type(exc).__name__}: {exc}")
+            sys.exit(1)
+
+
+def cmd_remove_skill() -> None:
+    svc, factory = _setup()
+    svc.add_skill(_make_skill("local/removable", "Removable"))
+    _commit(factory)
+
+    result = svc.remove_skill("local/removable")
+    _commit(factory)
+    assert result is True
+
+    check = svc.get_skill("local/removable")
+    assert check is None
+    print("skill-remove-ok")
+
+
+COMMANDS = {
+    "register-and-get": cmd_register_and_get,
+    "list-filter": cmd_list_filter,
+    "update-skill": cmd_update_skill,
+    "duplicate-reject": cmd_duplicate_reject,
+    "remove-skill": cmd_remove_skill,
+}
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
+        print(f"Usage: {sys.argv[0]} <{'|'.join(COMMANDS)}>", file=sys.stderr)
+        sys.exit(1)
+    COMMANDS[sys.argv[1]]()

robot/skill_registry.robot

@@ -0,0 +1,49 @@
+*** Settings ***
+Documentation    Smoke tests for Skill Registry persistence layer
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_skill_registry.py
+
+*** Test Cases ***
+Register And Retrieve A Skill
+    [Documentation]    Register a skill and retrieve it by name
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    register-and-get    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-registry-ok
+
+List Skills With Namespace Filter
+    [Documentation]    Register skills and list with namespace filter
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    list-filter    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-list-ok
+
+Update A Skill
+    [Documentation]    Register and update a skill description
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    update-skill    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-update-ok
+
+Reject Duplicate Skill Name
+    [Documentation]    Verify duplicate skill names are rejected
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    duplicate-reject    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-duplicate-reject-ok
+
+Remove A Skill
+    [Documentation]    Register and remove a skill
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    remove-skill    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    skill-remove-ok

src/cleveragents/application/services/__init__.py

@@ -6,11 +6,15 @@ Contains service classes that orchestrate business operations.
 from cleveragents.application.services.session_service import (
     PersistentSessionService,
 )
+from cleveragents.application.services.skill_registry_service import (
+    SkillRegistryService,
+)
 from cleveragents.application.services.tool_registry_service import (
     ToolRegistryService,
 )
 
 __all__ = [
     "PersistentSessionService",
+    "SkillRegistryService",
     "ToolRegistryService",
 ]

src/cleveragents/application/services/skill_registry_service.py

@@ -0,0 +1,91 @@
+"""Service layer for managing skill registrations.
+
+Orchestrates persistence through ``SkillRepository``, providing a clean
+API for CLI and higher-level orchestrators.
+"""
+
+from __future__ import annotations
+
+from cleveragents.domain.models.core.skill import Skill
+from cleveragents.infrastructure.database.repositories import (
+    SkillRepository,
+)
+
+
+class SkillRegistryService:
+    """Coordinate skill registration and lifecycle management.
+
+    Delegates all persistence to the repository layer.  Designed to be
+    injected with a ``SkillRepository`` (using session-factory pattern).
+    """
+
+    def __init__(self, skill_repo: SkillRepository) -> None:
+        self._skill_repo = skill_repo
+
+    # -- Skill CRUD --------------------------------------------------------
+
+    def add_skill(self, skill: Skill) -> Skill:
+        """Register a new skill.
+
+        Args:
+            skill: A ``Skill`` domain model instance.
+
+        Returns:
+            The registered skill.
+
+        Raises:
+            DuplicateSkillError: If the name already exists.
+            DatabaseError: On persistence failure.
+        """
+        return self._skill_repo.create(skill)
+
+    def get_skill(self, name: str) -> Skill | None:
+        """Retrieve a single skill by name.
+
+        Args:
+            name: The namespaced name.
+
+        Returns:
+            The ``Skill`` domain object, or ``None`` if not found.
+        """
+        return self._skill_repo.get_by_name(name)
+
+    def list_skills(
+        self,
+        namespace: str | None = None,
+    ) -> list[Skill]:
+        """List skills with optional filters.
+
+        Args:
+            namespace: Optional namespace filter.
+
+        Returns:
+            List of ``Skill`` domain objects.
+        """
+        return self._skill_repo.list_all(namespace=namespace)
+
+    def update_skill(self, skill: Skill) -> Skill:
+        """Update an existing skill definition.
+
+        Args:
+            skill: A ``Skill`` domain object with updated fields.
+
+        Returns:
+            The updated skill.
+
+        Raises:
+            SkillNotFoundError: If the skill is not found.
+            DatabaseError: On persistence failure.
+        """
+        return self._skill_repo.update(skill)
+
+    def remove_skill(self, name: str) -> bool:
+        """Remove a skill from the registry.
+
+        Args:
+            name: The namespaced name of the skill.
+
+        Returns:
+            ``True`` if the skill was removed, ``False`` if not found.
+        """
+        return self._skill_repo.delete(name)

src/cleveragents/infrastructure/database/__init__.py

@@ -24,6 +24,8 @@ from .models import (
     ResourceTypeModel,
     SessionMessageModel,
     SessionModel,
+    SkillItemModel,
+    SkillModel,
     ToolBindingModel,
     ToolModel,
     ToolResourceBindingModel,
@@ -43,6 +45,7 @@ from .repositories import (
     DuplicateResourceError,
     DuplicateResourceLinkError,
     DuplicateResourceTypeError,
+    DuplicateSkillError,
     DuplicateToolError,
     DuplicateValidationAttachmentError,
     InvalidToolTypeError,
@@ -62,6 +65,9 @@ from .repositories import (
     ResourceTypeRepository,
     SessionMessageRepository,
     SessionRepository,
+    SkillInUseError,
+    SkillNotFoundError,
+    SkillRepository,
     ToolInUseError,
     ToolNotFoundError,
     ToolRegistryRepository,
@@ -88,6 +94,7 @@ __all__ = [
     "DuplicateResourceError",
     "DuplicateResourceLinkError",
     "DuplicateResourceTypeError",
+    "DuplicateSkillError",
     "DuplicateToolError",
     "DuplicateValidationAttachmentError",
     "InvalidToolTypeError",
@@ -122,6 +129,11 @@ __all__ = [
     "SessionMessageRepository",
     "SessionModel",
     "SessionRepository",
+    "SkillInUseError",
+    "SkillItemModel",
+    "SkillModel",
+    "SkillNotFoundError",
+    "SkillRepository",
     "ToolBindingModel",
     "ToolInUseError",
     "ToolModel",

src/cleveragents/infrastructure/database/models.py

@@ -28,7 +28,8 @@ Includes spec-aligned lifecycle models per Stage A5
 from __future__ import annotations
 
 import json
-from datetime import datetime
+import re
+from datetime import UTC, datetime
 from typing import Any, cast
 
 from sqlalchemy import (
@@ -2104,6 +2105,296 @@ class AuditLogModel(Base):  # type: ignore[misc]
     )
 
 
+# ---------------------------------------------------------------------------
+# Skill Registry Models (Stage C0 - migration c0_001_skill_registry_tables)
+# ---------------------------------------------------------------------------
+
+_SKILL_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$")
+
+
+class SkillModel(Base):  # type: ignore[misc]
+    """Database model for registered skills.
+
+    Skills are namespaced, reusable collections of tools. The
+    ``name`` column (``namespace/short_name``) is the primary key.
+
+    Table: ``skills``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "skills"
+
+    # PK: namespaced name (e.g. "local/code-tools")
+    name = Column(String(255), primary_key=True)
+    namespace = Column(String(100), nullable=False)
+    short_name = Column(String(150), nullable=False)
+
+    # Core fields
+    description = Column(Text, nullable=False)
+    version = Column(String(50), nullable=True)
+    metadata_json = Column(Text, nullable=True)
+
+    # Timestamps (ISO-8601 strings)
+    created_at = Column(String(30), nullable=False)
+    updated_at = Column(String(30), nullable=False)
+
+    # Relationships (lazy="joined" prevents N+1 queries on list_all)
+    items_rel = relationship(
+        "SkillItemModel",
+        back_populates="skill",
+        cascade="all, delete-orphan",
+        order_by="SkillItemModel.item_order",
+        lazy="joined",
+    )
+
+    __table_args__ = (Index("ix_skills_namespace", "namespace"),)
+
+    # -- Domain conversion helpers ------------------------------------------
+
+    def to_domain(self) -> Any:
+        """Reconstruct a ``Skill`` domain object from the persisted model.
+
+        Reverse-maps persisted ``skill_items`` rows back to the appropriate
+        domain sub-objects (tool_refs, includes, anonymous_tools, etc.) and
+        reconstructs ``overrides`` from ``metadata_json``.
+
+        Returns:
+            A ``Skill`` domain model instance.
+        """
+        from cleveragents.domain.models.core.skill import (
+            Skill,
+            SkillAgentSource,
+            SkillInclude,
+            SkillInlineTool,
+            SkillMcpSource,
+        )
+
+        tool_refs: list[str] = []
+        includes: list[SkillInclude] = []
+        anonymous_tools: list[SkillInlineTool] = []
+        mcp_servers: list[SkillMcpSource] = []
+        agent_skills: list[SkillAgentSource] = []
+
+        for item in self.items_rel or []:
+            item_type = cast(str, item.item_type)
+            item_name = cast(str, item.item_name)
+            item_config_str = cast("str | None", item.item_config)
+
+            if item_type == "tool_ref":
+                tool_refs.append(item_name)
+            elif item_type == "include":
+                inc_overrides: dict[str, Any] | None = None
+                if item_config_str:
+                    config_data = json.loads(item_config_str)
+                    inc_overrides = config_data.get("overrides")
+                includes.append(SkillInclude(name=item_name, overrides=inc_overrides))
+            elif item_type == "inline_tool":
+                if item_config_str:
+                    config_data = json.loads(item_config_str)
+                    anonymous_tools.append(SkillInlineTool.model_validate(config_data))
+            elif item_type == "mcp_source":
+                if item_config_str:
+                    config_data = json.loads(item_config_str)
+                    mcp_servers.append(SkillMcpSource.model_validate(config_data))
+            elif item_type == "agent_source":
+                agent_skills.append(SkillAgentSource(path=item_name))
+
+        overrides: dict[str, dict[str, Any]] = {}
+        metadata_str = cast("str | None", self.metadata_json)
+        if metadata_str:
+            meta = json.loads(metadata_str)
+            overrides = meta.get("overrides", {})
+
+        return Skill(
+            name=cast(str, self.name),
+            description=cast(str, self.description),
+            tool_refs=tool_refs,
+            includes=includes,
+            anonymous_tools=anonymous_tools,
+            mcp_servers=mcp_servers,
+            agent_skills=agent_skills,
+            overrides=overrides,
+        )
+
+    @classmethod
+    def from_domain(cls, skill: Any) -> SkillModel:
+        """Create from a ``Skill`` domain model or dict.
+
+        Args:
+            skill: A ``Skill`` domain instance or dict-like object.
+
+        Returns:
+            A ``SkillModel`` ready for persistence.
+
+        Raises:
+            ValueError: If the skill name is empty or does not match
+                the required ``namespace/short_name`` pattern.
+        """
+
+        def _get(obj: Any, key: str, default: Any = None) -> Any:
+            if isinstance(obj, dict):
+                return obj.get(key, default)
+            return getattr(obj, key, default)
+
+        now_iso = datetime.now(tz=UTC).isoformat()
+        name_str: str = _get(skill, "name", "")
+
+        # SEC-1: Validate name format regardless of input type
+        if not _SKILL_NAME_RE.match(name_str):
+            raise ValueError(
+                f"Skill name must match 'namespace/short_name' pattern "
+                f"(alphanumeric, hyphens, underscores): got '{name_str}'"
+            )
+
+        # Build metadata JSON from overrides
+        overrides = _get(skill, "overrides", {})
+        metadata_json_str: str | None = None
+        if overrides:
+            metadata_json_str = json.dumps({"overrides": overrides})
+
+        model = cls(
+            name=name_str,
+            namespace=name_str.split("/", 1)[0] if "/" in name_str else "",
+            short_name=name_str.split("/", 1)[1] if "/" in name_str else name_str,
+            description=_get(skill, "description", ""),
+            version=_get(skill, "version"),
+            metadata_json=metadata_json_str,
+            created_at=_get(skill, "created_at", now_iso),
+            updated_at=_get(skill, "updated_at", now_iso),
+        )
+
+        # Populate items from domain model
+        idx = 0
+
+        # tool_refs
+        for ref in _get(skill, "tool_refs", []) or []:
+            model.items_rel.append(
+                SkillItemModel(
+                    item_type="tool_ref",
+                    item_name=ref,
+                    item_config=None,
+                    item_order=idx,
+                    created_at=now_iso,
+                )
+            )
+            idx += 1
+
+        # includes
+        for inc in _get(skill, "includes", []) or []:
+            inc_name = _get(inc, "name", "") if not isinstance(inc, str) else inc
+            inc_overrides = _get(inc, "overrides") if not isinstance(inc, str) else None
+            config: str | None = None
+            if inc_overrides:
+                config = json.dumps({"overrides": inc_overrides})
+            model.items_rel.append(
+                SkillItemModel(
+                    item_type="include",
+                    item_name=inc_name,
+                    item_config=config,
+                    item_order=idx,
+                    created_at=now_iso,
+                )
+            )
+            idx += 1
+
+        # anonymous (inline) tools
+        for anon in _get(skill, "anonymous_tools", []) or []:
+            anon_dict: dict[str, Any] = {}
+            if hasattr(anon, "model_dump"):
+                anon_dict = anon.model_dump(mode="json")
+            elif isinstance(anon, dict):
+                anon_dict = anon
+            else:
+                anon_dict = {
+                    "description": _get(anon, "description", ""),
+                    "source": _get(anon, "source", "custom"),
+                }
+            model.items_rel.append(
+                SkillItemModel(
+                    item_type="inline_tool",
+                    item_name=_get(anon, "description", f"_anon_{idx}"),
+                    item_config=json.dumps(anon_dict),
+                    item_order=idx,
+                    created_at=now_iso,
+                )
+            )
+            idx += 1
+
+        # mcp_servers
+        for mcp in _get(skill, "mcp_servers", []) or []:
+            mcp_dict: dict[str, Any] = {}
+            if hasattr(mcp, "model_dump"):
+                mcp_dict = mcp.model_dump(mode="json")
+            elif isinstance(mcp, dict):
+                mcp_dict = mcp
+            else:
+                mcp_dict = {"server": _get(mcp, "server", "")}
+            model.items_rel.append(
+                SkillItemModel(
+                    item_type="mcp_source",
+                    item_name=_get(mcp, "server", ""),
+                    item_config=json.dumps(mcp_dict),
+                    item_order=idx,
+                    created_at=now_iso,
+                )
+            )
+            idx += 1
+
+        # agent_skills
+        for agent in _get(skill, "agent_skills", []) or []:
+            agent_path = (
+                _get(agent, "path", "") if not isinstance(agent, str) else agent
+            )
+            model.items_rel.append(
+                SkillItemModel(
+                    item_type="agent_source",
+                    item_name=agent_path,
+                    item_config=None,
+                    item_order=idx,
+                    created_at=now_iso,
+                )
+            )
+            idx += 1
+
+        return model
+
+
+class SkillItemModel(Base):  # type: ignore[misc]
+    """Database model for skill items (child of skills).
+
+    Stores individual skill components: tool refs, includes, inline tools,
+    MCP sources, and agent sources with stable ordering.
+
+    Table: ``skill_items``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "skill_items"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    skill_name = Column(
+        String(255),
+        ForeignKey("skills.name", ondelete="CASCADE"),
+        nullable=False,
+    )
+    item_type = Column(String(30), nullable=False)
+    item_name = Column(String(500), nullable=False)
+    item_config = Column(Text, nullable=True)
+    item_order = Column(Integer, nullable=False)
+    created_at = Column(String(30), nullable=False)
+
+    # Relationships
+    skill = relationship(
+        "SkillModel",
+        back_populates="items_rel",
+    )
+
+    __table_args__ = (
+        Index("ix_skill_items_skill_name", "skill_name"),
+        Index("ix_skill_items_item_type", "item_type"),
+    )
+
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.

src/cleveragents/infrastructure/database/repositories.py

@@ -74,6 +74,7 @@ from cleveragents.domain.models.core import (
     Project,
     ProjectSettings,
 )
+from cleveragents.domain.models.core.skill import Skill
 from cleveragents.infrastructure.database.models import (
     ActorModel,
     AutomationProfileModel,
@@ -95,6 +96,8 @@ from cleveragents.infrastructure.database.models import (
     ResourceTypeModel,
     SessionMessageModel,
     SessionModel,
+    SkillItemModel,
+    SkillModel,
     ToolModel,
     ToolResourceBindingModel,
     ValidationAttachmentModel,
@@ -4353,3 +4356,217 @@ class AutomationProfileRepository:
             profile.allow_unsafe_tools
         )
         row.updated_at = now_iso  # type: ignore[assignment]
+
+
+# ---------------------------------------------------------------------------
+# Skill Registry Errors
+# ---------------------------------------------------------------------------
+
+
+class DuplicateSkillError(DatabaseError):
+    """Raised when creating a skill with a name that already exists."""
+
+    def __init__(self, name: str):
+        super().__init__(f"Skill with name '{name}' already exists")
+        self.skill_name = name
+
+
+class SkillInUseError(BusinessRuleViolation):
+    """Raised when deleting a skill that is still referenced."""
+
+    def __init__(self, skill_name: str, detail: str = ""):
+        msg = f"Cannot delete skill '{skill_name}'"
+        if detail:
+            msg = f"{msg}: {detail}"
+        super().__init__(msg)
+        self.skill_name = skill_name
+
+
+class SkillNotFoundError(DatabaseError):
+    """Raised when a skill cannot be found for update or retrieval."""
+
+    def __init__(self, name: str):
+        super().__init__(f"Skill '{name}' not found")
+        self.skill_name = name
+
+
+# ---------------------------------------------------------------------------
+# Skill Registry Repository
+# ---------------------------------------------------------------------------
+
+
+class SkillRepository:
+    """Repository for skill persistence.
+
+    Uses a session-factory pattern: each public method obtains its own
+    session from the factory, ensuring proper session lifecycle management.
+
+    All mutating methods flush (but do NOT commit); the caller or a
+    ``UnitOfWork`` wrapper is responsible for committing the transaction.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """Initialise with a callable that returns a new SQLAlchemy Session."""
+        self._session_factory = session_factory
+
+    def _session(self) -> Session:
+        """Convenience helper to obtain a session."""
+        return self._session_factory()
+
+    @database_retry
+    def create(self, skill: Skill) -> Skill:
+        """Persist a new skill domain object.
+
+        Args:
+            skill: A ``Skill`` domain model instance.
+
+        Returns:
+            The skill after persistence.
+
+        Raises:
+            DuplicateSkillError: If a skill with the same name already exists.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            db_model = SkillModel.from_domain(skill)
+            session.add(db_model)
+            session.flush()
+            return skill
+        except IntegrityError as exc:
+            session.rollback()
+            name_str = getattr(skill, "name", "")
+            if "UNIQUE" in str(exc).upper() or "unique" in str(exc).lower():
+                raise DuplicateSkillError(name_str) from exc
+            raise DatabaseError(f"Failed to create skill: {exc}") from exc
+        except DuplicateSkillError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to create skill: {exc}") from exc
+
+    @database_retry
+    def get_by_name(self, name: str) -> Skill | None:
+        """Retrieve a skill by its namespaced name (PK).
+
+        Returns:
+            The skill as a ``Skill`` domain object, or ``None`` if not found.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                return None
+            return row.to_domain()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to get skill {name}: {exc}") from exc
+
+    @database_retry
+    def list_all(
+        self,
+        namespace: str | None = None,
+    ) -> list[Skill]:
+        """List skills with optional namespace filter.
+
+        Args:
+            namespace: Optional namespace filter.
+
+        Returns:
+            List of ``Skill`` domain objects.
+        """
+        session = self._session()
+        try:
+            query = session.query(SkillModel)
+            if namespace is not None:
+                query = query.filter(SkillModel.namespace == namespace)
+            rows = query.order_by(SkillModel.name).all()
+            return [row.to_domain() for row in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to list skills: {exc}") from exc
+
+    @database_retry
+    def update(self, skill: Skill) -> Skill:
+        """Update all mutable fields of an existing skill.
+
+        Replaces child items entirely.
+
+        Args:
+            skill: The ``Skill`` domain object with updated fields.
+
+        Returns:
+            The skill after persistence.
+
+        Raises:
+            SkillNotFoundError: If the skill is not found.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+
+        name_str: str = skill.name
+        try:
+            row = session.query(SkillModel).filter_by(name=name_str).first()
+            if row is None:
+                raise SkillNotFoundError(name_str)
+
+            row.namespace = name_str.split("/", 1)[0]  # type: ignore[assignment]
+            row.short_name = name_str.split("/", 1)[1]  # type: ignore[assignment]
+            row.description = skill.description  # type: ignore[assignment]
+            row.version = getattr(skill, "version", None)  # type: ignore[assignment]
+
+            # Update metadata from overrides
+            overrides = skill.overrides
+            if overrides:
+                row.metadata_json = json.dumps({"overrides": overrides})  # type: ignore[assignment]
+            else:
+                row.metadata_json = None  # type: ignore[assignment]
+
+            row.updated_at = datetime.now(tz=UTC).isoformat()  # type: ignore[assignment]
+
+            # Replace child items via a fresh from_domain and swap items
+            row.items_rel.clear()  # type: ignore[union-attr]
+            new_model = SkillModel.from_domain(skill)
+            for item in new_model.items_rel:
+                row.items_rel.append(  # type: ignore[union-attr]
+                    SkillItemModel(
+                        item_type=item.item_type,
+                        item_name=item.item_name,
+                        item_config=item.item_config,
+                        item_order=item.item_order,
+                        created_at=item.created_at,
+                    )
+                )
+
+            session.flush()
+            return skill
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to update skill {name_str}: {exc}") from exc
+
+    @database_retry
+    def delete(self, name: str) -> bool:
+        """Delete a skill by namespaced name.
+
+        Cascades to child skill_items rows.
+
+        Args:
+            name: The namespaced name of the skill to delete.
+
+        Returns:
+            ``True`` if the skill was deleted, ``False`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                return False
+            session.delete(row)
+            session.flush()
+            return True
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to delete skill {name}: {exc}") from exc