COG-6311 feat: Add SKILLS search type and skill gate

CLAUDE.md

@@ -255,6 +255,7 @@ Available search types (from `cognee/modules/search/types/SearchType.py`), passe
 - **TEMPORAL** - Time-aware graph search
 - **FEELING_LUCKY** - Automatic search type selection
 - **CODING_RULES** - Code-specific search rules
+- **SKILLS** - Semantic discovery of skill playbooks (metadata-only, no LLM; requires exactly one dataset)
 
 `recall()` picks one of these automatically when `query_type` is omitted. The CLI is narrower: `cognee-cli recall --query-type` accepts only the choices in `cognee/cli/config.py:SEARCH_TYPE_CHOICES` and defaults to `HYBRID_COMPLETION`; the rest are SDK-only.
 
@@ -704,6 +705,15 @@ Opt-in LLM check that runs as the last `cognify()` task (default **off**). After
 - **Applies to `remember()` too** — and to session memory bridged back by `improve()` — since those build their graphs through `cognify()`. The exception is `remember(content_type="code")`, which runs the separate code-graph pipeline.
 - **Scope / limitations**: only the 1-hop neighbourhood of the touched entities is compared; structural edges (`contains`, `is_part_of`, `made_from`, `exists_in`, `contradicts`) and edges with an unnamed endpoint are skipped; the temporal cognify path is not covered.
 
+### Skills (Procedural Memory)
+Dataset-scoped `SKILL.md` playbooks agents can discover, load on demand, execute, and improve from run history.
+
+- **Ingest**: `remember(content_type="skills", dataset_name=...)` (folder, file, or inline via `skills_text`/`skill_name`) — requires an explicit dataset; re-ingest upserts (deterministic ids). HTTP: `POST /skills`.
+- **Discover**: `SearchType.SKILLS` — one vector search over the `Skill_search_text` collection, no LLM, **metadata-only results** (never the procedure body; progressive disclosure). Requires exactly one dataset; skills outside that dataset's scope, inactive skills, and empty-scope legacy skills are filtered out. Missing collection returns `[]`, not an error.
+- **Skill gate**: `recall()` runs a deterministic regex gate (`cognee/api/v1/recall/skill_gate.py`); procedural-sounding queries trigger a concurrent SKILLS lookup whose hits are appended tagged `source="skills"`. Additive and fail-safe; only fires when exactly one dataset is targeted. Disable with `SKILL_GATE_ENABLED=false`.
+- **Execute**: `SearchType.AGENTIC_COMPLETION` with `skills=[...]` — LLM sees name+description, loads bodies via the `load_skill` tool (12k char cap).
+- **Improve**: `SkillRun` records (via `remember()` skill-run entries) feed LLM-drafted `SkillImprovementProposal`s; preview then apply by proposal id (`/proposals` router).
+
 ### Code Files (cognify CODE route)
 Supported code files (`.py`, `.go`, `.ts`, `.java`, `.rs`, … — the extension list lives on `code_loader`) are recognized at add time through the loader system: the code loader claims the file, stores it under its real extension, and `ingest_data` tags the record with `system_metadata = {"source": "code"}`. Cognify then routes such items down the CODE route, which runs the deterministic enola code graph pipeline per file — typed `CodeSymbol`/`CodeModule`/… nodes with `calls`/`imports`/`has_method` edges, **no LLM calls**.
 

README.md

@@ -198,6 +198,31 @@ A third flag, `DATASET_QUEUE_ENABLED=false`, removes the per-process concurrency
 on datasets; it saves a little latency but risks file-lock leaks and resource
 exhaustion when multiple datasets run in parallel — leave it on for servers.
 
+### Procedural memory (skills)
+
+Beyond facts, cognee stores **skills** — dataset-scoped `SKILL.md` playbooks that
+agents can discover semantically, load on demand, execute, and improve from run
+history:
+
+```python
+# Ingest a folder of SKILL.md files into a dataset
+await cognee.remember("./skills", content_type="skills", dataset_name="ops")
+
+# Discover by meaning — returns name/description/metadata, never the full procedure
+results = await cognee.search("how do I deploy to staging",
+                              query_type=SearchType.SKILLS, datasets=["ops"])
+
+# recall() also runs a deterministic (no-LLM) skill gate: procedural questions
+# automatically get matching skills appended, tagged source="skills"
+results = await cognee.recall("how do I deploy to staging", datasets=["ops"])
+```
+
+Skills are authored as `SKILL.md` files (one directory per skill, frontmatter for
+description/tools/maintainer), scoped to exactly one dataset at ingest time, and
+executed through `SearchType.AGENTIC_COMPLETION`, where the agent loads full
+procedure bodies on demand via the `load_skill` tool. Disable the recall gate with
+`SKILL_GATE_ENABLED=false`.
+
 ## Run with Docker
 
 Prefer containers? Cognee publishes prebuilt images to Docker Hub on every push to `main`:

cognee/api/v1/recall/recall.py

@@ -1,3 +1,4 @@
+import asyncio
 import re
 from typing import Annotated, Literal
 from uuid import UUID
@@ -34,6 +35,7 @@
     ResponseMarkerEntry,
     ResponseQAEntry,
     ResponseSessionContextEntry,
+    ResponseSkillEntry,
     ResponseToolEntry,
 )
 from cognee.modules.recall.types.SearchResultItem import SearchResultItem
@@ -900,6 +902,60 @@ async def _run_code() -> list[RecallResponse]:
                     )
                 return tagged
 
+            async def _run_skill_gate(gate_top_k: int) -> list[RecallResponse]:
+                """Metadata-only SKILLS lookup for the deterministic skill gate.
+
+                Skipped silently unless exactly one dataset is targeted (skill
+                lookup is single-dataset by invariant). Any failure contributes
+                nothing instead of failing the recall.
+                """
+                from cognee.modules.search.methods.search import authorized_search
+
+                try:
+                    gate_user = user
+                    if gate_user is None:
+                        gate_user = await get_default_user()
+
+                    if dataset_ids and len(dataset_ids) == 1:
+                        gate_dataset_ids = list(dataset_ids)
+                    elif datasets and len(datasets) == 1:
+                        authorized = await get_authorized_existing_datasets(
+                            datasets, "read", gate_user
+                        )
+                        if len(authorized) != 1:
+                            return []
+                        gate_dataset_ids = [dataset.id for dataset in authorized]
+                    else:
+                        return []
+
+                    payloads = await authorized_search(
+                        query_text=query_text,
+                        query_type=SearchType.SKILLS,
+                        user=gate_user,
+                        dataset_ids=gate_dataset_ids,
+                        top_k=gate_top_k,
+                    )
+                except Exception as error:
+                    logger.warning("Skill gate lookup failed (non-fatal): %s", error)
+                    return []
+
+                entries: list[RecallResponse] = []
+                for payload in payloads or []:
+                    for item in getattr(payload, "completion", None) or []:
+                        if not isinstance(item, dict):
+                            continue
+                        name = item.get("name") or ""
+                        description = item.get("description") or ""
+                        entries.append(
+                            ResponseSkillEntry(
+                                source="skills",
+                                text=f"{name}: {description}" if description else name,
+                                skill={k: v for k, v in item.items() if k != "score"},
+                                score=item.get("score"),
+                            )
+                        )
+                return entries
+
             runners = {
                 "session": _run_session,
                 "trace": _run_trace,
@@ -909,22 +965,49 @@ async def _run_code() -> list[RecallResponse]:
                 "code": _run_code,
             }
 
+            # Deterministic skill gate: a procedural-looking query triggers a
+            # concurrent metadata-only SKILLS lookup (one vector search, no
+            # LLM call). Additive only — the main lanes never wait on it, and
+            # explicit SKILLS / AGENTIC_COMPLETION calls bypass it.
+            skills_task = None
+            if (
+                "graph" in sources
+                and not only_context
+                and query_type not in (SearchType.SKILLS, SearchType.AGENTIC_COMPLETION)
+            ):
+                from cognee.api.v1.recall.skill_gate import (
+                    DEFAULT_SKILL_GATE_TOP_K,
+                    should_search_skills,
+                    skill_gate_enabled,
+                )
+
+                if skill_gate_enabled() and should_search_skills(query_text).fired:
+                    skills_task = asyncio.create_task(_run_skill_gate(DEFAULT_SKILL_GATE_TOP_K))
+
             session_result_count = 0
-            for src in sources:
-                runner = runners.get(src)
-                if runner is None:
-                    continue
-                # Auto mode special case: session hit short-circuits graph.
-                if auto_fallthrough and src == "graph" and merged:
-                    break
-                # on_empty: the other sources gave cognee enough context — don't
-                # go back to the external database.
-                if src == "tools" and tools_trigger == "on_empty" and merged:
-                    continue
-                part = await runner()
-                if src == "session":
-                    session_result_count = len(part)
-                merged.extend(part)
+            try:
+                for src in sources:
+                    runner = runners.get(src)
+                    if runner is None:
+                        continue
+                    # Auto mode special case: session hit short-circuits graph.
+                    if auto_fallthrough and src == "graph" and merged:
+                        break
+                    # on_empty: the other sources gave cognee enough context — don't
+                    # go back to the external database.
+                    if src == "tools" and tools_trigger == "on_empty" and merged:
+                        continue
+                    part = await runner()
+                    if src == "session":
+                        session_result_count = len(part)
+                    merged.extend(part)
+            except BaseException:
+                if skills_task is not None:
+                    skills_task.cancel()
+                raise
+
+            if skills_task is not None:
+                merged.extend(await skills_task)
 
             if session_result_count:
                 span.set_attribute(COGNEE_SESSION_ENTRY_COUNT, session_result_count)

cognee/api/v1/recall/skill_gate.py

@@ -0,0 +1,86 @@
+"""
+Deterministic skill gate for recall().
+
+Decides — with regexes only, no LLM and no I/O — whether a query looks
+procedural enough to warrant a skill lookup. When the gate fires (and exactly
+one dataset is targeted), recall() runs a metadata-only SKILLS search
+concurrently with the main search and appends the hits tagged
+``source="skills"``. The gate is additive: the main answer is never replaced
+or blocked by it.
+
+Disable with ``SKILL_GATE_ENABLED=false``.
+"""
+
+import os
+import re
+from dataclasses import dataclass, field
+
+from cognee.api.v1.recall.query_router import _is_negated
+from cognee.shared.logging_utils import get_logger
+
+logger = get_logger("skill_gate")
+
+# How many skills the gate lane asks for. Deliberately small: gate hits are a
+# side-channel next to the main answer, not the answer itself.
+DEFAULT_SKILL_GATE_TOP_K = 3
+
+# Each rule: (pattern, weight). Weights accumulate; the gate fires at the
+# threshold. Weak signals (bare ops verbs) score 2.0 so one alone never fires;
+# procedural phrasings score 3.0+ and fire on their own.
+_GATE_RULES: list[tuple[re.Pattern, float]] = [
+    (re.compile(r"\b(how (do|can|should|would) (i|we|you)|how to)\b", re.IGNORECASE), 3.0),
+    (re.compile(r"\b(steps? (to|for)|step.by.step)\b", re.IGNORECASE), 3.0),
+    (re.compile(r"\b(procedure|playbook|runbook|checklist|workflow)\b", re.IGNORECASE), 4.0),
+    (re.compile(r"\b(walk (me|us) through|guide (to|for|on))\b", re.IGNORECASE), 3.0),
+    (re.compile(r"\bwhat('?s| is) the (process|procedure)\b", re.IGNORECASE), 4.0),
+    (re.compile(r"\bskills?\b", re.IGNORECASE), 4.0),
+    (
+        re.compile(
+            r"\b(set(ting)? up|setup|install(ing)?|configur(e|ing)|deploy(ing)?"
+            r"|migrat(e|ing)|provision(ing)?|onboard(ing)?|rotate|troubleshoot(ing)?)\b",
+            re.IGNORECASE,
+        ),
+        2.0,
+    ),
+]
+
+_GATE_THRESHOLD = 3.0
+
+
+@dataclass
+class GateResult:
+    """Gate decision with the score and matched fragments for observability."""
+
+    fired: bool
+    score: float = 0.0
+    matched: list[str] = field(default_factory=list)
+
+
+def skill_gate_enabled() -> bool:
+    """``SKILL_GATE_ENABLED`` env flag; on unless explicitly disabled."""
+    return os.getenv("SKILL_GATE_ENABLED", "true").strip().lower() not in ("false", "0", "no")
+
+
+def should_search_skills(query: str) -> GateResult:
+    """Decide whether ``query`` warrants a skill lookup. Pure function, no I/O.
+
+    Every rule whose pattern matches (and is not negated, same suppression as
+    the recall query router) adds its weight; the gate fires when the total
+    reaches the threshold.
+    """
+    q = (query or "").strip()
+    if not q:
+        return GateResult(fired=False)
+
+    score = 0.0
+    matched: list[str] = []
+    for pattern, weight in _GATE_RULES:
+        match = pattern.search(q)
+        if match and not _is_negated(q, match):
+            score += weight
+            matched.append(match.group(0))
+
+    fired = score >= _GATE_THRESHOLD
+    if fired:
+        logger.info("skill_gate fired: score=%.1f matched=%s query=%r", score, matched, q)
+    return GateResult(fired=fired, score=score, matched=matched)

cognee/api/v1/search/search.py

@@ -337,14 +337,19 @@ async def search(
             ):
                 operation_context.set_dataset(target_dataset_ids[0])
 
-            if query_type is SearchType.AGENTIC_COMPLETION:
+            if query_type in (SearchType.AGENTIC_COMPLETION, SearchType.SKILLS):
                 active_dataset_refs = dataset_ids if dataset_ids else datasets
                 if isinstance(active_dataset_refs, UUID):
                     active_dataset_refs = [active_dataset_refs]
                 if not active_dataset_refs or len(active_dataset_refs) != 1:
+                    if query_type is SearchType.AGENTIC_COMPLETION:
+                        raise CogneeValidationError(
+                            message="Agentic skill search requires exactly one explicit dataset.",
+                            name="InvalidAgenticDatasetScope",
+                        )
                     raise CogneeValidationError(
-                        message="Agentic skill search requires exactly one explicit dataset.",
-                        name="InvalidAgenticDatasetScope",
+                        message="SKILLS search requires exactly one explicit dataset.",
+                        name="InvalidSkillsDatasetScope",
                     )
 
             if any(v is not None for v in agentic_overrides.values()):

cognee/cli/config.py

@@ -44,6 +44,7 @@
     "CODE",
     "CYPHER",
     "GRAPH_REPORT",
+    "SKILLS",
 ]
 
 DEFAULT_SEARCH_TYPE = "HYBRID_COMPLETION"

cognee/modules/recall/methods/normalize_search_payload.py

@@ -36,6 +36,7 @@
     SearchType.CHUNKS_LEXICAL: SearchResultKind.CHUNK,
     SearchType.SUMMARIES: SearchResultKind.SUMMARY,
     SearchType.AGENTIC_COMPLETION: SearchResultKind.GRAPH_COMPLETION,
+    SearchType.SKILLS: SearchResultKind.SKILL,
 }
 
 

cognee/modules/recall/types/RecallResponse.py

@@ -53,6 +53,22 @@ class ResponseToolEntry(BaseModel):
     structured: Optional[dict] = None
 
 
+class ResponseSkillEntry(BaseModel):
+    """One skill surfaced by the deterministic skill gate.
+
+    Metadata-only: ``skill`` carries the projected Skill fields and never the
+    procedure body — progressive disclosure keeps bodies behind the
+    ``load_skill`` tool or ``GET /skills/{skill_id}``. ``text`` is a
+    renderable "name: description" line; ``score`` is the raw vector distance
+    (lower is better) when available.
+    """
+
+    source: Literal["skills"]
+    text: str
+    skill: dict
+    score: Optional[float] = None
+
+
 class ResponseMarkerEntry(BaseModel):
     """System-generated marker (not data), e.g. "memory still warming up".
 
@@ -74,6 +90,7 @@ class ResponseMarkerEntry(BaseModel):
     | ResponseGraphEntry
     | ResponseCodeEntry
     | ResponseToolEntry
+    | ResponseSkillEntry
     | ResponseMarkerEntry,
     Field(discriminator="source"),
 ]

cognee/modules/recall/types/SearchResultItem.py

@@ -37,6 +37,7 @@ class SearchResultKind(str, Enum):
     SUMMARY = "summary"
     CODING_RULE = "coding_rule"
     CODE = "code"
+    SKILL = "skill"
     NATURAL_LANGUAGE = "natural_language"
     TEMPORAL = "temporal"
     STRUCTURED = "structured"  # when a response_model was supplied