alembic/versions/m4_002_skill_flattened_tools.py

@@ -0,0 +1,60 @@
+"""Add flattened tool persistence columns to skills table.
+
+Adds columns for caching flattened tool sets, includes, capability
+summaries, original YAML text, and a SHA-256 hash for refresh
+invalidation.  Also adds a uniqueness constraint on the ``name``
+column (defence-in-depth — already PK-unique) and ensures the
+``ix_skills_namespace`` index exists for filtered listing.
+
+Revision ID: m4_002_skill_flattened_tools
+Revises: m4_001_decision_tables
+Create Date: 2026-02-26 12:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m4_002_skill_flattened_tools"
+down_revision: str | Sequence[str] | None = "m4_001_decision_tables"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add flattened tool caching columns to skills table."""
+    with op.batch_alter_table("skills") as batch_op:
+        batch_op.add_column(
+            sa.Column("flattened_tools_json", sa.Text(), nullable=True),
+        )
+        batch_op.add_column(
+            sa.Column("includes_json", sa.Text(), nullable=True),
+        )
+        batch_op.add_column(
+            sa.Column("capability_summary_json", sa.Text(), nullable=True),
+        )
+        batch_op.add_column(
+            sa.Column("yaml_text", sa.Text(), nullable=True),
+        )
+        batch_op.add_column(
+            sa.Column("flattening_hash", sa.String(64), nullable=True),
+        )
+        # Defence-in-depth: explicit uniqueness on name (already PK).
+        batch_op.create_unique_constraint(
+            "uq_skills_name",
+            ["name"],
+        )
+
+
+def downgrade() -> None:
+    """Remove flattened tool caching columns from skills table."""
+    with op.batch_alter_table("skills") as batch_op:
+        batch_op.drop_constraint("uq_skills_name", type_="unique")
+        batch_op.drop_column("flattening_hash")
+        batch_op.drop_column("yaml_text")
+        batch_op.drop_column("capability_summary_json")
+        batch_op.drop_column("includes_json")
+        batch_op.drop_column("flattened_tools_json")

benchmarks/skill_registry_persist_bench.py

@@ -0,0 +1,130 @@
+"""ASV benchmarks for Skill Registry flattened tool persistence.
+
+Measures performance of:
+- Persisting flattened tools with ``update_flattened_tools``
+- Refresh-check via ``needs_refresh``
+- Namespace-filtered listing via ``list_all``
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+import uuid
+from pathlib import Path
+
+try:
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import Session, sessionmaker
+
+    from cleveragents.domain.models.core.skill import Skill
+    from cleveragents.infrastructure.database.models import Base
+    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 Session, sessionmaker
+
+    from cleveragents.domain.models.core.skill import Skill
+    from cleveragents.infrastructure.database.models import Base
+    from cleveragents.infrastructure.database.repositories import (
+        SkillRepository,
+    )
+
+
+def _make_skill(name: str) -> Skill:
+    return Skill(name=name, description=f"Benchmark skill {name}")
+
+
+class SkillPersistWithTools:
+    """Benchmark persisting flattened tools on an existing skill."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", echo=False)
+        Base.metadata.create_all(self.engine)
+        self.factory: sessionmaker[Session] = sessionmaker(bind=self.engine)
+        self.repo = SkillRepository(session_factory=self.factory)
+
+        self.skill = _make_skill("bench/persist-tools")
+        self.repo.create(self.skill)
+        self.factory().commit()
+
+        self.tools_json = json.dumps([f"tool-{i}" for i in range(20)])
+        self.yaml_text = "name: bench/persist-tools\ntools: [...]"
+        self.hash_val = hashlib.sha256(self.yaml_text.encode()).hexdigest()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_skill_persist_with_tools(self) -> None:
+        self.repo.update_flattened_tools(
+            name="bench/persist-tools",
+            flattened_tools_json=self.tools_json,
+            includes_json="[]",
+            capability_summary_json="{}",
+            yaml_text=self.yaml_text,
+            flattening_hash=self.hash_val,
+        )
+        self.factory().commit()
+
+
+class SkillRefreshCheck:
+    """Benchmark refresh-check (hash comparison)."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", echo=False)
+        Base.metadata.create_all(self.engine)
+        self.factory: sessionmaker[Session] = sessionmaker(bind=self.engine)
+        self.repo = SkillRepository(session_factory=self.factory)
+
+        self.skill = _make_skill("bench/refresh-check")
+        self.repo.create(self.skill)
+        self.factory().commit()
+
+        self.repo.update_flattened_tools(
+            name="bench/refresh-check",
+            flattened_tools_json="[]",
+            includes_json="[]",
+            capability_summary_json="{}",
+            yaml_text="name: bench/refresh-check",
+            flattening_hash="match_hash",
+        )
+        self.factory().commit()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_skill_refresh_check(self) -> None:
+        self.repo.needs_refresh("bench/refresh-check", "match_hash")
+
+
+class SkillListByNamespace:
+    """Benchmark listing skills filtered by namespace."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", echo=False)
+        Base.metadata.create_all(self.engine)
+        self.factory: sessionmaker[Session] = sessionmaker(bind=self.engine)
+        self.repo = SkillRepository(session_factory=self.factory)
+
+        # Seed 50 skills across 5 namespaces
+        for ns in ("local", "devops", "data", "ml", "infra"):
+            for i in range(10):
+                uid = uuid.uuid4().hex[:8]
+                self.repo.create(_make_skill(f"{ns}/tool-{i}-{uid}"))
+        self.factory().commit()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_skill_list_by_namespace(self) -> None:
+        self.repo.list_all(namespace="local")

docs/reference/database_schema.md

@@ -230,6 +230,75 @@ tracking, and optional compiled metadata.
 
 ---
 
+## Skill Registry Tables
+
+The skill registry stores namespaced skill definitions with cached
+flattened tool sets for fast resolution.
+
+| Table          | Model              | Description                        |
+|----------------|--------------------|------------------------------------|
+| `skills`       | `SkillModel`       | Registered skill definitions       |
+| `skill_items`  | `SkillItemModel`   | Child items (tools, includes, etc) |
+
+### `skills`
+
+Primary key: `name` (namespaced, e.g. `local/code-tools`).
+
+| Column                   | Type         | Constraints          | Description                                    |
+|--------------------------|--------------|----------------------|------------------------------------------------|
+| `name`                   | String(255)  | PK, UNIQUE           | Namespaced name (`namespace/short_name`)       |
+| `namespace`              | String(100)  | NOT NULL             | Namespace extracted from name                  |
+| `short_name`             | String(150)  | NOT NULL             | Short name extracted from name                 |
+| `description`            | Text         | NOT NULL             | Human-readable description                     |
+| `version`                | String(50)   |                      | Optional version string                        |
+| `metadata_json`          | Text         |                      | JSON overrides metadata                        |
+| `flattened_tools_json`   | Text         |                      | JSON-serialized flattened tool descriptors      |
+| `includes_json`          | Text         |                      | JSON-serialized includes list                  |
+| `capability_summary_json`| Text         |                      | JSON-serialized capability summary             |
+| `yaml_text`              | Text         |                      | Original YAML text of the skill definition     |
+| `flattening_hash`        | String(64)   |                      | SHA-256 hash for refresh invalidation          |
+| `created_at`             | String(30)   | NOT NULL             | ISO-8601 timestamp                             |
+| `updated_at`             | String(30)   | NOT NULL             | ISO-8601 timestamp                             |
+
+**Unique constraints:**
+
+- `uq_skills_name`: `UNIQUE(name)` (defence-in-depth — PK already unique)
+
+**Indexes:**
+
+- `ix_skills_namespace` on `(namespace)` — used for filtered skill listing
+
+### `skill_items`
+
+Child table linking skill components to their parent skill.
+
+| Column       | Type         | Constraints                              | Description                    |
+|--------------|--------------|------------------------------------------|--------------------------------|
+| `id`         | Integer      | PK, AUTOINCREMENT                        | Auto-generated row ID          |
+| `skill_name` | String(255)  | NOT NULL, FK -> skills.name CASCADE      | Parent skill reference         |
+| `item_type`  | String(30)   | NOT NULL                                 | Item kind (tool_ref, include, etc) |
+| `item_name`  | String(500)  | NOT NULL                                 | Item identifier                |
+| `item_config`| Text         |                                          | JSON configuration blob        |
+| `item_order` | Integer      | NOT NULL                                 | Stable ordering index          |
+| `created_at` | String(30)   | NOT NULL                                 | ISO-8601 timestamp             |
+
+**Indexes:**
+
+- `ix_skill_items_skill_name` on `(skill_name)`
+- `ix_skill_items_item_type` on `(item_type)`
+
+### Persistence field mapping
+
+| Database Column            | Domain Model Property                    |
+|----------------------------|------------------------------------------|
+| `flattened_tools_json`     | `ResolvedToolEntry` list (JSON)          |
+| `includes_json`            | `SkillInclude` list (JSON)               |
+| `capability_summary_json`  | `SkillCapabilitySummary` (JSON)          |
+| `yaml_text`                | Original YAML source text                |
+| `flattening_hash`          | SHA-256 digest of `yaml_text`            |
+
+---
+
 ## Decision Tree Tables
 
 The decision tree subsystem consists of two tables introduced in Stage M3.2:

features/skill_registry_persistence.feature

@@ -0,0 +1,84 @@
+@phase1 @domain @repository @skill_registry @m4_persist
+Feature: Skill Registry Flattened Tool Persistence
+  As a system operator managing skills
+  I want flattened tool sets persisted alongside skill definitions
+  So that the execution layer can skip redundant resolution and detect stale caches
+
+  Background:
+    Given a clean in-memory database with the skill persistence schema
+    And a skill persistence repository
+
+  # ---------------------------------------------------------------------------
+  # Persisting flattened tools
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_create
+  Scenario: Skill creation persists flattened tool set
+    Given a skill "local/code-tools" with description "Code helper tools"
+    When the skill is created via the persistence repository
+    And the flattened tools are updated with tools json '["tool-a","tool-b"]' and hash "abc123"
+    Then the stored flattened tools json should be '["tool-a","tool-b"]'
+    And the stored flattening hash should be "abc123"
+
+  # ---------------------------------------------------------------------------
+  # Update invalidation
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_invalidate
+  Scenario: Skill update invalidates cached capability summary
+    Given a skill "local/editor" with description "Editor tools"
+    And the skill is created via the persistence repository
+    And the flattened tools are updated with tools json '["edit"]' and hash "hash1"
+    When the skill "local/editor" is updated with description "Updated editor"
+    Then the stored flattened tools json for "local/editor" should be null
+    And the stored capability summary json for "local/editor" should be null
+    And the stored flattening hash for "local/editor" should be null
+
+  # ---------------------------------------------------------------------------
+  # Hash-based staleness detection
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_hash
+  Scenario: Flattening hash detects stale cached data
+    Given a skill "local/lint" with description "Lint tools"
+    And the skill is created via the persistence repository
+    And the flattened tools are updated with tools json '["lint"]' and hash "hash_v1"
+    Then the skill "local/lint" should not need refresh for hash "hash_v1"
+    And the skill "local/lint" should need refresh for hash "hash_v2"
+
+  # ---------------------------------------------------------------------------
+  # Refresh recomputation
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_refresh
+  Scenario: Skill refresh recomputes flattened tools
+    Given a skill "local/refresh-me" with description "Refreshable"
+    And the skill is created via the persistence repository
+    When the flattening hash for "local/refresh-me" is recomputed from yaml "name: local/refresh-me"
+    Then the stored flattening hash for "local/refresh-me" should not be null
+    And the skill "local/refresh-me" should not need refresh for the recomputed hash
+
+  # ---------------------------------------------------------------------------
+  # Uniqueness constraint
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_unique
+  Scenario: Uniqueness constraint prevents duplicate skill names
+    Given a skill "local/unique-test" with description "First"
+    And the skill is created via the persistence repository
+    When a second skill "local/unique-test" with description "Second" is created
+    Then the second creation should raise a duplicate error
+
+  # ---------------------------------------------------------------------------
+  # Namespace index
+  # ---------------------------------------------------------------------------
+
+  @skill_persist_namespace
+  Scenario: Namespace index supports filtered listing
+    Given a skill "local/tool-a" with description "Tool A"
+    And the skill is created via the persistence repository
+    And a skill "devops/deploy" with description "Deploy"
+    And the skill is created via the persistence repository
+    When skills are listed with namespace "local"
+    Then the listed skills should contain "local/tool-a"
+    And the listed skills should not contain "devops/deploy"

features/steps/skill_registry_persistence_steps.py

@@ -0,0 +1,275 @@
+"""Step definitions for Skill Registry flattened tool persistence.
+
+Covers the m4_002 columns: ``flattened_tools_json``, ``includes_json``,
+``capability_summary_json``, ``yaml_text``, ``flattening_hash`` and
+the related ``SkillRepository`` helper methods.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.domain.models.core.skill import Skill
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    DuplicateSkillError,
+    SkillRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
+    """Enable SQLite foreign key enforcement per connection."""
+    cursor = dbapi_conn.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+def _factory(context: Context) -> Callable[[], Session]:
+    return context.sp_session_factory  # type: ignore[no-any-return]
+
+
+def _commit(context: Context) -> None:
+    _factory(context)().commit()
+
+
+def _make_skill(name: str, description: str = "Test skill") -> Skill:
+    return Skill(name=name, description=description)
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a clean in-memory database with the skill persistence schema")
+def step_clean_db(context: Context) -> None:
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    event.listen(engine, "connect", _enable_fk_pragma)
+    Base.metadata.create_all(engine)
+
+    shared_session: Session = sessionmaker(bind=engine)()
+
+    def _sf() -> Session:
+        return shared_session
+
+    context.sp_engine = engine
+    context.sp_session_factory = _sf
+    context.sp_error: BaseException | None = None
+    context.sp_recomputed_hash: str | None = None
+    context.sp_listed: list[Skill] = []
+
+
+@given("a skill persistence repository")
+def step_repo(context: Context) -> None:
+    context.sp_repo = SkillRepository(session_factory=_factory(context))
+
+
+# ---------------------------------------------------------------------------
+# Skill creation
+# ---------------------------------------------------------------------------
+
+
+@given('a skill "{name}" with description "{desc}"')
+def step_skill(context: Context, name: str, desc: str) -> None:
+    context.sp_current_skill = _make_skill(name, desc)
+
+
+@when("the skill is created via the persistence repository")
+def step_create(context: Context) -> None:
+    context.sp_repo.create(context.sp_current_skill)
+    _commit(context)
+
+
+@given("the skill is created via the persistence repository")
+def step_create_given(context: Context) -> None:
+    context.sp_repo.create(context.sp_current_skill)
+    _commit(context)
+
+
+# ---------------------------------------------------------------------------
+# Flattened tools update
+# ---------------------------------------------------------------------------
+
+
+@when("the flattened tools are updated with tools json '{tools}' and hash \"{h}\"")
+def step_update_flat(context: Context, tools: str, h: str) -> None:
+    name = context.sp_current_skill.name
+    context.sp_repo.update_flattened_tools(
+        name=name,
+        flattened_tools_json=tools,
+        includes_json="[]",
+        capability_summary_json="{}",
+        yaml_text="name: " + name,
+        flattening_hash=h,
+    )
+    _commit(context)
+
+
+@given("the flattened tools are updated with tools json '{tools}' and hash \"{h}\"")
+def step_update_flat_given(context: Context, tools: str, h: str) -> None:
+    name = context.sp_current_skill.name
+    context.sp_repo.update_flattened_tools(
+        name=name,
+        flattened_tools_json=tools,
+        includes_json="[]",
+        capability_summary_json="{}",
+        yaml_text="name: " + name,
+        flattening_hash=h,
+    )
+    _commit(context)
+
+
+# ---------------------------------------------------------------------------
+# Assertions on stored fields
+# ---------------------------------------------------------------------------
+
+
+@then("the stored flattened tools json should be '{expected}'")
+def step_check_tools(context: Context, expected: str) -> None:
+    data = context.sp_repo.get_flattened_tools(context.sp_current_skill.name)
+    assert data["flattened_tools_json"] == expected, (
+        f"Expected {expected!r}, got {data['flattened_tools_json']!r}"
+    )
+
+
+@then('the stored flattening hash should be "{expected}"')
+def step_check_hash(context: Context, expected: str) -> None:
+    data = context.sp_repo.get_flattened_tools(context.sp_current_skill.name)
+    assert data["flattening_hash"] == expected, (
+        f"Expected {expected!r}, got {data['flattening_hash']!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Update & invalidation
+# ---------------------------------------------------------------------------
+
+
+@when('the skill "{name}" is updated with description "{desc}"')
+def step_update_skill(context: Context, name: str, desc: str) -> None:
+    updated = _make_skill(name, desc)
+    context.sp_repo.update(updated)
+    _commit(context)
+
+
+@then('the stored flattened tools json for "{name}" should be null')
+def step_check_tools_null(context: Context, name: str) -> None:
+    data = context.sp_repo.get_flattened_tools(name)
+    assert data["flattened_tools_json"] is None, (
+        f"Expected None, got {data['flattened_tools_json']!r}"
+    )
+
+
+@then('the stored capability summary json for "{name}" should be null')
+def step_check_cap_null(context: Context, name: str) -> None:
+    data = context.sp_repo.get_flattened_tools(name)
+    assert data["capability_summary_json"] is None, (
+        f"Expected None, got {data['capability_summary_json']!r}"
+    )
+
+
+@then('the stored flattening hash for "{name}" should be null')
+def step_check_hash_null(context: Context, name: str) -> None:
+    data = context.sp_repo.get_flattened_tools(name)
+    assert data["flattening_hash"] is None, (
+        f"Expected None, got {data['flattening_hash']!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Hash-based staleness
+# ---------------------------------------------------------------------------
+
+
+@then('the skill "{name}" should not need refresh for hash "{h}"')
+def step_no_refresh(context: Context, name: str, h: str) -> None:
+    assert not context.sp_repo.needs_refresh(name, h)
+
+
+@then('the skill "{name}" should need refresh for hash "{h}"')
+def step_needs_refresh(context: Context, name: str, h: str) -> None:
+    assert context.sp_repo.needs_refresh(name, h)
+
+
+# ---------------------------------------------------------------------------
+# Recompute
+# ---------------------------------------------------------------------------
+
+
+@when('the flattening hash for "{name}" is recomputed from yaml "{yaml}"')
+def step_recompute(context: Context, name: str, yaml: str) -> None:
+    context.sp_recomputed_hash = context.sp_repo.recompute_flattening_hash(name, yaml)
+    _commit(context)
+
+
+@then('the stored flattening hash for "{name}" should not be null')
+def step_hash_not_null(context: Context, name: str) -> None:
+    data = context.sp_repo.get_flattened_tools(name)
+    assert data["flattening_hash"] is not None
+
+
+@then('the skill "{name}" should not need refresh for the recomputed hash')
+def step_no_refresh_recomputed(context: Context, name: str) -> None:
+    h = context.sp_recomputed_hash
+    assert h is not None, "Recomputed hash is None"
+    assert not context.sp_repo.needs_refresh(name, h)
+
+
+# ---------------------------------------------------------------------------
+# Uniqueness constraint
+# ---------------------------------------------------------------------------
+
+
+@when('a second skill "{name}" with description "{desc}" is created')
+def step_create_duplicate(context: Context, name: str, desc: str) -> None:
+    try:
+        context.sp_repo.create(_make_skill(name, desc))
+        _commit(context)
+        context.sp_error = None
+    except DuplicateSkillError as exc:
+        context.sp_error = exc
+    except Exception as exc:
+        cause = exc.__cause__ or exc
+        if isinstance(cause, DuplicateSkillError):
+            context.sp_error = cause
+        else:
+            context.sp_error = exc
+
+
+@then("the second creation should raise a duplicate error")
+def step_check_dup(context: Context) -> None:
+    assert isinstance(context.sp_error, DuplicateSkillError), (
+        f"Expected DuplicateSkillError, got {type(context.sp_error)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Namespace filtering
+# ---------------------------------------------------------------------------
+
+
+@when('skills are listed with namespace "{ns}"')
+def step_list_ns(context: Context, ns: str) -> None:
+    context.sp_listed = context.sp_repo.list_all(namespace=ns)
+
+
+@then('the listed skills should contain "{name}"')
+def step_list_contains(context: Context, name: str) -> None:
+    names = [s.name for s in context.sp_listed]
+    assert name in names, f"{name} not in {names}"
+
+
+@then('the listed skills should not contain "{name}"')
+def step_list_not_contains(context: Context, name: str) -> None:
+    names = [s.name for s in context.sp_listed]
+    assert name not in names, f"{name} unexpectedly in {names}"

robot/helper_skill_registry_persistence.py

@@ -0,0 +1,134 @@
+"""Helper script for Robot Framework skill registry persistence smoke tests.
+
+Usage:
+    python helper_skill_registry_persistence.py <command>
+
+Commands:
+    persist-round-trip    Create a skill, persist flattened tools, retrieve
+    update-invalidation   Update a skill and verify cache invalidation
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+# 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.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
+    SkillRepository,
+)
+
+
+def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
+    """Enable SQLite foreign key enforcement per connection."""
+    cursor = dbapi_conn.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()
+
+
+def _setup() -> tuple[SkillRepository, sessionmaker[Session]]:
+    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)
+    return repo, factory
+
+
+def _commit(factory: sessionmaker[Session]) -> None:
+    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_persist_round_trip() -> None:
+    """Create skill, persist flattened tools, retrieve and verify."""
+    repo, factory = _setup()
+
+    skill = _make_skill("local/persist-test", "Persistence test")
+    repo.create(skill)
+    _commit(factory)
+
+    repo.update_flattened_tools(
+        name="local/persist-test",
+        flattened_tools_json='["tool-a","tool-b"]',
+        includes_json="[]",
+        capability_summary_json='{"total_tools": 2}',
+        yaml_text="name: local/persist-test",
+        flattening_hash="deadbeef",
+    )
+    _commit(factory)
+
+    data = repo.get_flattened_tools("local/persist-test")
+    assert data["flattened_tools_json"] == '["tool-a","tool-b"]'
+    assert data["flattening_hash"] == "deadbeef"
+    assert data["capability_summary_json"] == '{"total_tools": 2}'
+
+    print("persist-round-trip-ok")
+
+
+def cmd_update_invalidation() -> None:
+    """Update skill and verify cached fields are nulled."""
+    repo, factory = _setup()
+
+    skill = _make_skill("local/invalidate-test", "Before update")
+    repo.create(skill)
+    _commit(factory)
+
+    repo.update_flattened_tools(
+        name="local/invalidate-test",
+        flattened_tools_json='["x"]',
+        includes_json="[]",
+        capability_summary_json="{}",
+        yaml_text="name: local/invalidate-test",
+        flattening_hash="hash1",
+    )
+    _commit(factory)
+
+    # Now update the skill — should null the cache
+    updated = _make_skill("local/invalidate-test", "After update")
+    repo.update(updated)
+    _commit(factory)
+
+    data = repo.get_flattened_tools("local/invalidate-test")
+    assert data["flattened_tools_json"] is None, (
+        f"Expected None, got {data['flattened_tools_json']}"
+    )
+    assert data["flattening_hash"] is None, (
+        f"Expected None, got {data['flattening_hash']}"
+    )
+
+    print("update-invalidation-ok")
+
+
+COMMANDS: dict[str, Any] = {
+    "persist-round-trip": cmd_persist_round_trip,
+    "update-invalidation": cmd_update_invalidation,
+}
+
+
+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_persistence_smoke.robot

@@ -0,0 +1,25 @@
+*** Settings ***
+Documentation    Smoke tests for Skill Registry flattened tool persistence
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_skill_registry_persistence.py
+
+*** Test Cases ***
+Verify Skill Persistence Round Trip
+    [Documentation]    Create a skill, persist flattened tools, and retrieve them
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    persist-round-trip    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    persist-round-trip-ok
+
+Verify Skill Update Invalidation
+    [Documentation]    Update a skill and verify cached fields are nulled
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    update-invalidation    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    update-invalidation-ok

src/cleveragents/infrastructure/database/models.py

@@ -2147,6 +2147,13 @@ class SkillModel(Base):  # type: ignore[misc]
     created_at = Column(String(30), nullable=False)
     updated_at = Column(String(30), nullable=False)
 
+    # Flattened tool cache columns (m4_002)
+    flattened_tools_json = Column(Text, nullable=True)
+    includes_json = Column(Text, nullable=True)
+    capability_summary_json = Column(Text, nullable=True)
+    yaml_text = Column(Text, nullable=True)
+    flattening_hash = Column(String(64), nullable=True)
+
     # Relationships (lazy="joined" prevents N+1 queries on list_all)
     items_rel = relationship(
         "SkillItemModel",
@@ -2156,7 +2163,10 @@ class SkillModel(Base):  # type: ignore[misc]
         lazy="joined",
     )
 
-    __table_args__ = (Index("ix_skills_namespace", "namespace"),)
+    __table_args__ = (
+        Index("ix_skills_namespace", "namespace"),
+        UniqueConstraint("name", name="uq_skills_name"),
+    )
 
     # -- Domain conversion helpers ------------------------------------------
 

src/cleveragents/infrastructure/database/repositories.py

@@ -52,6 +52,7 @@ Based on ADR-007 (Repository Pattern) and ADR-033 (Retry Patterns).
 
 from __future__ import annotations
 
+import hashlib
 import json
 import warnings
 from collections import deque
@@ -60,6 +61,7 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, cast
 
+import structlog
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.orm import Session
@@ -111,6 +113,8 @@ from cleveragents.infrastructure.database.models import (
 if TYPE_CHECKING:
     from cleveragents.domain.models.core.decision import Decision
 
+_log = structlog.get_logger(__name__)
+
 
 class ProjectRepository:
     """Repository for project persistence."""
@@ -4530,6 +4534,12 @@ class SkillRepository:
 
             row.updated_at = datetime.now(tz=UTC).isoformat()  # type: ignore[assignment]
 
+            # Invalidate cached flattened tool data on update
+            row.flattened_tools_json = None  # type: ignore[assignment]
+            row.includes_json = None  # type: ignore[assignment]
+            row.capability_summary_json = None  # type: ignore[assignment]
+            row.flattening_hash = None  # 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)
@@ -4579,6 +4589,203 @@ class SkillRepository:
             session.rollback()
             raise DatabaseError(f"Failed to delete skill {name}: {exc}") from exc
 
+    # -- Flattened tool cache methods (m4_002) ------------------------------
+
+    @database_retry
+    def update_flattened_tools(
+        self,
+        name: str,
+        flattened_tools_json: str,
+        includes_json: str,
+        capability_summary_json: str,
+        yaml_text: str,
+        flattening_hash: str,
+    ) -> None:
+        """Persist a flattened tool set and associated cache data.
+
+        Args:
+            name: Namespaced skill name (PK).
+            flattened_tools_json: JSON-serialized flattened tool descriptors.
+            includes_json: JSON-serialized includes list.
+            capability_summary_json: JSON-serialized capability summary.
+            yaml_text: Original YAML text of the skill definition.
+            flattening_hash: SHA-256 hash for refresh invalidation.
+
+        Raises:
+            SkillNotFoundError: If the skill does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                raise SkillNotFoundError(name)
+            row.flattened_tools_json = flattened_tools_json  # type: ignore[assignment]
+            row.includes_json = includes_json  # type: ignore[assignment]
+            row.capability_summary_json = capability_summary_json  # type: ignore[assignment]
+            row.yaml_text = yaml_text  # type: ignore[assignment]
+            row.flattening_hash = flattening_hash  # type: ignore[assignment]
+            row.updated_at = datetime.now(tz=UTC).isoformat()  # type: ignore[assignment]
+            session.flush()
+            _log.info(
+                "skill_flattened_tools_updated",
+                skill_name=name,
+                hash=flattening_hash,
+            )
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to update flattened tools for {name}: {exc}",
+            ) from exc
+
+    @database_retry
+    def get_flattened_tools(
+        self,
+        name: str,
+    ) -> dict[str, str | None]:
+        """Retrieve the stored flattened tool cache fields for a skill.
+
+        Args:
+            name: Namespaced skill name.
+
+        Returns:
+            Dict with keys ``flattened_tools_json``, ``includes_json``,
+            ``capability_summary_json``, ``yaml_text``, ``flattening_hash``.
+            Values are ``None`` when the cache has been invalidated.
+
+        Raises:
+            SkillNotFoundError: If the skill does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                raise SkillNotFoundError(name)
+            return {
+                "flattened_tools_json": cast("str | None", row.flattened_tools_json),
+                "includes_json": cast("str | None", row.includes_json),
+                "capability_summary_json": cast(
+                    "str | None", row.capability_summary_json
+                ),
+                "yaml_text": cast("str | None", row.yaml_text),
+                "flattening_hash": cast("str | None", row.flattening_hash),
+            }
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get flattened tools for {name}: {exc}",
+            ) from exc
+
+    @database_retry
+    def needs_refresh(self, name: str, current_hash: str) -> bool:
+        """Check whether a skill's cached flattened data is stale.
+
+        Compares the stored ``flattening_hash`` against *current_hash*.
+
+        Args:
+            name: Namespaced skill name.
+            current_hash: SHA-256 hash of the current YAML text.
+
+        Returns:
+            ``True`` if the stored hash is missing or differs from
+            *current_hash*; ``False`` if they match.
+
+        Raises:
+            SkillNotFoundError: If the skill does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                raise SkillNotFoundError(name)
+            stored_hash = cast("str | None", row.flattening_hash)
+            if stored_hash is None:
+                return True
+            return stored_hash != current_hash
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to check refresh for {name}: {exc}",
+            ) from exc
+
+    @database_retry
+    def recompute_flattening_hash(self, name: str, yaml_text: str) -> str:
+        """Compute SHA-256 of *yaml_text* and store it on the skill row.
+
+        Args:
+            name: Namespaced skill name.
+            yaml_text: The YAML text to hash.
+
+        Returns:
+            The computed SHA-256 hex digest.
+
+        Raises:
+            SkillNotFoundError: If the skill does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                raise SkillNotFoundError(name)
+            new_hash = hashlib.sha256(yaml_text.encode("utf-8")).hexdigest()
+            row.flattening_hash = new_hash  # type: ignore[assignment]
+            row.updated_at = datetime.now(tz=UTC).isoformat()  # type: ignore[assignment]
+            session.flush()
+            _log.info(
+                "skill_flattening_hash_recomputed",
+                skill_name=name,
+                hash=new_hash,
+            )
+            return new_hash
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to recompute hash for {name}: {exc}",
+            ) from exc
+
+    @database_retry
+    def invalidate_cached_summaries(self, name: str) -> None:
+        """Null the cached flattened fields for a skill.
+
+        Called when the skill definition changes and the cached
+        flattened tool set is no longer valid.
+
+        Args:
+            name: Namespaced skill name.
+
+        Raises:
+            SkillNotFoundError: If the skill does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(SkillModel).filter_by(name=name).first()
+            if row is None:
+                raise SkillNotFoundError(name)
+            row.flattened_tools_json = None  # type: ignore[assignment]
+            row.includes_json = None  # type: ignore[assignment]
+            row.capability_summary_json = None  # type: ignore[assignment]
+            row.flattening_hash = None  # type: ignore[assignment]
+            row.updated_at = datetime.now(tz=UTC).isoformat()  # type: ignore[assignment]
+            session.flush()
+            _log.info("skill_cached_summaries_invalidated", skill_name=name)
+        except SkillNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to invalidate cached summaries for {name}: {exc}",
+            ) from exc
+
 
 # ---------------------------------------------------------------------------
 # Decision repository (Stage M3.2)