diff --git a/src/opensquilla/engine/pipeline.py b/src/opensquilla/engine/pipeline.py
index 48c8c36d7..06d552b70 100644
--- a/src/opensquilla/engine/pipeline.py
+++ b/src/opensquilla/engine/pipeline.py
@@ -107,16 +107,19 @@ async def run_pipeline(ctx: TurnContext, steps: list[TurnStep]) -> TurnContext:
routing_source = cast(RoutingSource, ctx.metadata.get("routing_source", "none"))
confidence = ctx.metadata.get("routing_confidence")
filtered_skill_ids = None
+ filtered_skill_reasons = None
elif step_name == "filter_skills":
routed_tier = None
routing_source = "none"
confidence = None
filtered_skill_ids = ctx.metadata.get("filtered_skill_ids")
+ filtered_skill_reasons = ctx.metadata.get("filtered_skill_reasons")
else:
routed_tier = None
routing_source = "none"
confidence = None
filtered_skill_ids = None
+ filtered_skill_reasons = None
records = ctx.metadata.setdefault("pipeline_steps", records)
records.append(
@@ -125,6 +128,7 @@ async def run_pipeline(ctx: TurnContext, steps: list[TurnStep]) -> TurnContext:
applied=applied,
routed_tier=routed_tier,
filtered_skill_ids=filtered_skill_ids,
+ filtered_skill_reasons=filtered_skill_reasons,
routing_source=routing_source,
confidence=confidence,
fallback_reason=None,
diff --git a/src/opensquilla/engine/steps/skills_filter.py b/src/opensquilla/engine/steps/skills_filter.py
index c3ac5de34..86147a12d 100644
--- a/src/opensquilla/engine/steps/skills_filter.py
+++ b/src/opensquilla/engine/steps/skills_filter.py
@@ -111,26 +111,46 @@ def _deterministic_gate(
skills: list[SkillSpec],
available_tools: set[str],
elig_ctx: EligibilityContext | None = None,
+ drop_reasons: dict[str, str] | None = None,
) -> list[SkillSpec]:
- """Pure-Python gate: eligibility, requires_tools, fallback, visibility."""
+ """Pure-Python gate: eligibility, requires_tools, fallback, visibility.
+
+ When ``drop_reasons`` is provided, each dropped skill's stable reason
+ code is recorded there (skill ID -> reason code) so the step output can
+ explain *why* a skill was dropped, not only *that* it survived.
+ """
ctx_elig = elig_ctx or _elig_ctx
gated: list[SkillSpec] = []
+
+ def _drop(skill: SkillSpec, reason: str) -> None:
+ if drop_reasons is not None:
+ drop_reasons[_skill_id(skill)] = reason
+
with _elig_ctx_lock:
for s in skills:
if s.disable_model_invocation:
+ _drop(s, "disable_model_invocation")
continue
if not check_eligibility(s, ctx_elig):
+ _drop(s, "eligibility_failed")
continue
if s.requires_tools and not all(t in available_tools for t in s.requires_tools):
+ _drop(s, "missing_required_tools")
continue
if s.fallback_for_toolsets and any(
t in available_tools for t in s.fallback_for_toolsets
):
+ _drop(s, "superseded_by_toolset")
continue
gated.append(s)
return gated
+def _skill_id(skill: SkillSpec) -> str:
+ """Stable skill identifier used across filter metadata and logs."""
+ return getattr(skill, "id", None) or getattr(skill, "name", None) or ""
+
+
async def filter_skills(ctx: TurnContext) -> TurnContext:
"""Gate, optionally filter, and inject skills into the system prompt.
@@ -145,6 +165,7 @@ async def filter_skills(ctx: TurnContext) -> TurnContext:
tools_cfg = getattr(ctx.config, "tools", None) if ctx.config else None
if getattr(tools_cfg, "profile", None) == "memory_only":
ctx.metadata["filtered_skill_ids"] = []
+ ctx.metadata["filtered_skill_reasons"] = {}
ctx.metadata["skill_count"] = 0
ctx.metadata["skills_prompt_chars"] = 0
log.debug("skills_filter.skipped", reason="memory_only")
@@ -174,12 +195,26 @@ async def filter_skills(ctx: TurnContext) -> TurnContext:
# ── deterministic gate (no LLM, pure Python) ──
available_tools = {t.name for t in ctx.tool_defs} if ctx.tool_defs else set()
skills_cfg_for_gate = getattr(ctx.config, "skills", None) if ctx.config else None
- gated = _deterministic_gate(all_skills, available_tools, _eligibility_ctx(skills_cfg_for_gate))
+ gate_drop_reasons: dict[str, str] = {}
+ gated = _deterministic_gate(
+ all_skills,
+ available_tools,
+ _eligibility_ctx(skills_cfg_for_gate),
+ drop_reasons=gate_drop_reasons,
+ )
# Hide meta-skills from the model whenever auto-trigger is off (manual-only
# mode) or the subsystem is fully disabled. They remain in the loader so the
# /meta command can still enumerate and run them.
+ meta_drop_reasons: dict[str, str] = {}
if not (meta_skill_enabled and meta_auto_trigger):
+ meta_hidden_ids = {
+ _skill_id(s)
+ for s in gated
+ if getattr(s, "kind", "skill") == "meta" and _skill_id(s)
+ }
gated = [s for s in gated if getattr(s, "kind", "skill") != "meta"]
+ for hidden_id in meta_hidden_ids:
+ meta_drop_reasons[hidden_id] = "meta_hidden_auto_trigger_off"
for key in (
"meta_match",
"meta_match_trigger",
@@ -258,19 +293,47 @@ async def filter_skills(ctx: TurnContext) -> TurnContext:
else:
filtered = filterable
+ # Record why each filterable skill was dropped by retrieval. With the
+ # filter disabled nothing is dropped; a retriever that returns fewer
+ # than top_k non-empty results with candidates available indicates a
+ # full retrieval failure (see HybridRetriever.retrieve's fail paths),
+ # otherwise anything not selected is recorded as not_in_top_k.
+ retrieval_drop_reasons: dict[str, str] = {}
+ if filter_enabled and filtered != filterable:
+ if not filtered and filterable:
+ failed_reason = "retrieval_failed"
+ for s in filterable:
+ sid = _skill_id(s)
+ if sid:
+ retrieval_drop_reasons[sid] = failed_reason
+ else:
+ kept = {_skill_id(s) for s in filtered}
+ for s in filterable:
+ sid = _skill_id(s)
+ if sid and sid not in kept:
+ retrieval_drop_reasons[sid] = "not_in_top_k"
+
final = pinned + filtered
# Publish the post-filter skill-ID list so the pipeline wrapper can
# surface it in the decision log's PipelineStepRecord. Non-mutating
# additive read for callers that don't consume the metadata.
+ # ``filtered_skill_reasons`` complements the survivor list with a stable
+ # reason code per dropped skill (gate / meta-visibility / retrieval),
+ # addressing issue #54: "filtered_skill_ids records which skills
+ # survived but not why any were dropped".
try:
ctx.metadata["filtered_skill_ids"] = [
- getattr(s, "id", None) or getattr(s, "name", None)
- for s in filtered
- if getattr(s, "id", None) or getattr(s, "name", None)
+ _skill_id(s) for s in filtered if _skill_id(s)
]
+ ctx.metadata["filtered_skill_reasons"] = {
+ **gate_drop_reasons,
+ **meta_drop_reasons,
+ **retrieval_drop_reasons,
+ }
except Exception: # pragma: no cover — metadata is best-effort
ctx.metadata["filtered_skill_ids"] = []
+ ctx.metadata["filtered_skill_reasons"] = {}
from opensquilla.skills.injector import SkillInjector
@@ -310,11 +373,7 @@ async def filter_skills(ctx: TurnContext) -> TurnContext:
# but cannot tell which skills were chosen vs missed — the diagnostic
# signal needed to debug recall quality (e.g. "why did 'commit my
# changes to git' not surface `git`?").
- pinned_ids = [
- getattr(s, "id", None) or getattr(s, "name", None)
- for s in pinned
- if getattr(s, "id", None) or getattr(s, "name", None)
- ]
+ pinned_ids = [_skill_id(s) for s in pinned if _skill_id(s)]
filtered_ids = ctx.metadata.get("filtered_skill_ids") or []
log.debug(
"skills_filter.applied",
@@ -335,6 +394,7 @@ async def filter_skills(ctx: TurnContext) -> TurnContext:
),
pinned_skills=pinned_ids,
filtered_skills=filtered_ids,
+ dropped_skills=dict(ctx.metadata.get("filtered_skill_reasons") or {}),
)
if skills_prompt and injection_mode == "user_context":
diff --git a/src/opensquilla/observability/decision_log.py b/src/opensquilla/observability/decision_log.py
index 4a56c0733..2ad917362 100644
--- a/src/opensquilla/observability/decision_log.py
+++ b/src/opensquilla/observability/decision_log.py
@@ -105,6 +105,10 @@ class PipelineStepRecord:
applied: bool
routed_tier: str | None = None
filtered_skill_ids: list[str] | None = None
+ # skill ID -> stable drop reason code (gate / meta-visibility / retrieval).
+ # Additive companion to ``filtered_skill_ids``; readers of older rows see
+ # None via _filter_payload.
+ filtered_skill_reasons: dict[str, str] | None = None
routing_source: RoutingSource = "none"
confidence: float | None = None
fallback_reason: str | None = None
diff --git a/tests/test_observability/test_filtered_skill_reasons.py b/tests/test_observability/test_filtered_skill_reasons.py
new file mode 100644
index 000000000..960c32090
--- /dev/null
+++ b/tests/test_observability/test_filtered_skill_reasons.py
@@ -0,0 +1,268 @@
+"""filtered_skill_reasons: per-skill drop reason codes for the skills_filter step.
+
+Addresses issue #54: ``filtered_skill_ids`` records which skills survived the
+skills_filter step but not why any were dropped. The step now publishes a
+companion ``filtered_skill_reasons`` mapping (skill ID -> stable reason code)
+covering the three drop surfaces: the deterministic gate, meta-skill visibility,
+and retrieval top-k selection. The mapping flows into the decision log's
+``PipelineStepRecord`` and round-trips through the JSONL writer/reader.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from opensquilla.engine import pipeline as pipeline_mod
+from opensquilla.engine.steps import skills_filter
+from opensquilla.engine.steps.skills_filter import filter_skills
+from opensquilla.observability.decision_log import (
+ DecisionEntry,
+ PipelineStepRecord,
+ load_entries,
+ write_decision_entry,
+)
+from opensquilla.skills.types import SkillLayer, SkillSpec
+
+
+def _skill(name: str) -> SkillSpec:
+ return SkillSpec(
+ name=name,
+ description=f"{name} skill",
+ layer=SkillLayer.BUNDLED,
+ always=False,
+ triggers=[],
+ content="body",
+ )
+
+
+def _make_entry(**overrides) -> DecisionEntry:
+ defaults = dict(
+ turn_id="t1",
+ session_key="s1",
+ prompt_hash="a" * 16,
+ system_prompt_hash="b" * 16,
+ tool_list_hash="c" * 16,
+ tool_choice="auto",
+ tokens_input=10,
+ tokens_output=20,
+ model="claude",
+ provider="anthropic",
+ latency_ms=100,
+ ts="2026-01-01T00:00:00Z",
+ pipeline_steps=[
+ PipelineStepRecord(
+ step_name="filter_skills",
+ applied=True,
+ filtered_skill_ids=["weather-local"],
+ filtered_skill_reasons={"github-local": "not_in_top_k"},
+ )
+ ],
+ )
+ defaults.update(overrides)
+ return DecisionEntry(**defaults)
+
+
+def test_gate_reason_code_disable_model_invocation() -> None:
+ spec = _skill("hidden")
+ spec.disable_model_invocation = True
+ drop_reasons: dict[str, str] = {}
+ gated = skills_filter._deterministic_gate(
+ [spec], available_tools=set(), drop_reasons=drop_reasons
+ )
+ assert gated == []
+ assert drop_reasons == {"hidden": "disable_model_invocation"}
+
+
+def test_gate_reason_code_missing_required_tools() -> None:
+ spec = _skill("needs-git")
+ spec.requires_tools = ["git_status"]
+ drop_reasons: dict[str, str] = {}
+ gated = skills_filter._deterministic_gate(
+ [spec], available_tools=set(), drop_reasons=drop_reasons
+ )
+ assert gated == []
+ assert drop_reasons == {"needs-git": "missing_required_tools"}
+
+
+def test_gate_reason_code_superseded_by_toolset() -> None:
+ spec = _skill("legacy-git")
+ spec.fallback_for_toolsets = ["git_status"]
+ drop_reasons: dict[str, str] = {}
+ gated = skills_filter._deterministic_gate(
+ [spec], available_tools={"git_status"}, drop_reasons=drop_reasons
+ )
+ assert gated == []
+ assert drop_reasons == {"legacy-git": "superseded_by_toolset"}
+
+
+def test_pipeline_step_record_roundtrips_filtered_skill_reasons(tmp_path: Path) -> None:
+ entry = _make_entry()
+ write_decision_entry(entry, log_dir=tmp_path)
+ loaded = load_entries(next(tmp_path.glob("decisions-*.jsonl")))
+ assert len(loaded) == 1
+ (step,) = loaded[0].pipeline_steps
+ assert step.filtered_skill_ids == ["weather-local"]
+ assert step.filtered_skill_reasons == {"github-local": "not_in_top_k"}
+
+
+def test_old_row_without_filtered_skill_reasons_reads_as_none(tmp_path: Path) -> None:
+ """Backward-tolerant read: a pre-#54 row (no filtered_skill_reasons) must
+ hydrate cleanly with None via _filter_payload."""
+ legacy_step = {
+ "step_name": "filter_skills",
+ "applied": True,
+ "filtered_skill_ids": ["weather-local"],
+ "routing_source": "none",
+ }
+ payload = {
+ "turn_id": "old",
+ "session_key": "s",
+ "prompt_hash": "a" * 16,
+ "system_prompt_hash": "b" * 16,
+ "tool_list_hash": "c" * 16,
+ "tool_choice": "auto",
+ "tokens_input": 1,
+ "tokens_output": 2,
+ "model": "x",
+ "provider": "y",
+ "latency_ms": 3,
+ "ts": "2026-01-01T00:00:00Z",
+ "pipeline_steps": [legacy_step],
+ }
+ path = tmp_path / "decisions-20260101.jsonl"
+ path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
+ loaded = load_entries(path)
+ assert len(loaded) == 1
+ (step,) = loaded[0].pipeline_steps
+ assert step.filtered_skill_ids == ["weather-local"]
+ assert step.filtered_skill_reasons is None
+
+
+@pytest.mark.asyncio
+async def test_run_pipeline_wires_filtered_skill_reasons_into_record() -> None:
+ """run_pipeline must copy ctx.metadata['filtered_skill_reasons'] into the
+ filter_skills PipelineStepRecord (and None for other steps)."""
+
+ async def other_step(ctx):
+ return ctx
+
+ async def filter_skills(ctx):
+ ctx.metadata["filtered_skill_ids"] = ["weather-local"]
+ ctx.metadata["filtered_skill_reasons"] = {"github-local": "not_in_top_k"}
+ return ctx
+
+ ctx = pipeline_mod.TurnContext(
+ message="hi",
+ session_key="agent:main:webchat:default",
+ config=None,
+ provider=None,
+ model="test-model",
+ tool_defs=[],
+ system_prompt="base",
+ metadata={"pipeline_steps": []},
+ )
+ ctx = await pipeline_mod.run_pipeline(ctx, [other_step, filter_skills])
+ (first, second) = ctx.metadata["pipeline_steps"]
+ assert first.filtered_skill_reasons is None
+ assert second.step_name == "filter_skills"
+ assert second.filtered_skill_ids == ["weather-local"]
+ assert second.filtered_skill_reasons == {"github-local": "not_in_top_k"}
+
+
+@pytest.mark.asyncio
+async def test_retrieval_top_k_drop_recorded_as_not_in_top_k(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """filter_enabled + top_k smaller than candidate count -> survivors get IDs,
+ dropped candidates get the not_in_top_k reason code (the exact issue #54
+ scenario, with a real loader and lexical retriever)."""
+ from opensquilla.skills.loader import SkillLoader
+
+ workspace = tmp_path / "workspace"
+ for name, description, triggers in (
+ ("weather-local", "Fetch weather forecasts.", "[weather, forecast]"),
+ ("github-local", "Inspect GitHub pull requests.", "[github, pull request]"),
+ ):
+ skill_dir = workspace / name
+ skill_dir.mkdir(parents=True)
+ (skill_dir / "SKILL.md").write_text(
+ "---\n"
+ f"name: {name}\n"
+ f"description: {description}\n"
+ f"triggers: {triggers}\n"
+ "---\n\n"
+ f"# {name}\n",
+ encoding="utf-8",
+ )
+
+ monkeypatch.setattr(skills_filter, "_retriever", None)
+ loader = SkillLoader(workspace_dir=workspace, snapshot_path=tmp_path / "snapshot.json")
+ ctx = pipeline_mod.TurnContext(
+ message="please check the weather forecast",
+ session_key="agent:main:webchat:default",
+ config=SimpleNamespace(
+ tools=SimpleNamespace(profile="standard"),
+ skills=SimpleNamespace(
+ filter_enabled=True,
+ filter_top_k=1,
+ filter_strategy="lexical",
+ filter_lexical_top_n=20,
+ filter_semantic_top_n=20,
+ filter_rrf_k=60,
+ filter_embedding_model="BAAI/bge-small-zh-v1.5",
+ max_skills_prompt_chars=100_000,
+ injection_mode="system",
+ ),
+ ),
+ provider=None,
+ model="test-model",
+ tool_defs=[
+ SimpleNamespace(name=name)
+ for name in (
+ "background_process",
+ "cron",
+ "exec_command",
+ "memory_get",
+ "memory_save",
+ "memory_search",
+ "process",
+ )
+ ],
+ system_prompt="base",
+ metadata={"skill_loader": loader},
+ )
+
+ ctx = await filter_skills(ctx)
+
+ assert ctx.metadata["filtered_skill_ids"] == ["weather-local"]
+ assert ctx.metadata["filtered_skill_reasons"] == {"github-local": "not_in_top_k"}
+
+
+def test_retrieval_failure_shape_maps_to_retrieval_failed_code() -> None:
+ """Empty retriever output with candidates available -> retrieval_failed.
+ The step-level branch is exercised via its public contract here."""
+ from opensquilla.skills.retrieval import HybridRetriever
+
+ retriever = HybridRetriever(embedder=None, strategy="lexical")
+ # A query with zero lexical hits (no FTS/substring match) makes rank()
+ # return [] for every layer, so retrieve() hits its full-failure path
+ # and returns [] — the exact shape the step maps to retrieval_failed.
+ dropped = retriever.retrieve([_skill("a"), _skill("b")], "zzzzqqqqxxxx", top_k=5)
+ assert dropped == []
+
+ # Mirror the step's branch: empty result + non-empty filterable -> the
+ # stable retrieval_failed code (never not_in_top_k).
+ filterable = [_skill("a"), _skill("b")]
+ retrieval_drop_reasons: dict[str, str] = {}
+ if not dropped and filterable:
+ for s in filterable:
+ retrieval_drop_reasons[skills_filter._skill_id(s)] = "retrieval_failed"
+ assert retrieval_drop_reasons == {
+ "a": "retrieval_failed",
+ "b": "retrieval_failed",
+ }
diff --git a/tests/test_skills/test_skill_disable_toggle.py b/tests/test_skills/test_skill_disable_toggle.py
index 77056d9ea..adb52075b 100644
--- a/tests/test_skills/test_skill_disable_toggle.py
+++ b/tests/test_skills/test_skill_disable_toggle.py
@@ -63,6 +63,27 @@ def test_enabled_when_not_disabled(self):
)
assert {s.name for s in gated} == {"code-task"}
+ def test_drop_reasons_recorded_per_condition(self):
+ ctx = EligibilityContext.auto(disabled_set={"code-task"})
+ drop_reasons: dict[str, str] = {}
+ gated = skills_filter._deterministic_gate(
+ [_skill("code-task"), _skill("git-diff")],
+ available_tools=set(),
+ elig_ctx=ctx,
+ drop_reasons=drop_reasons,
+ )
+ assert {s.name for s in gated} == {"git-diff"}
+ assert drop_reasons == {"code-task": "eligibility_failed"}
+
+ def test_drop_reasons_optional_and_absent_by_default(self):
+ ctx = EligibilityContext.auto(disabled_set={"code-task"})
+ gated = skills_filter._deterministic_gate(
+ [_skill("code-task")], available_tools=set(), elig_ctx=ctx
+ )
+ assert gated == []
+ # No caller-provided dict: behaviour identical to pre-#54 signature.
+ assert skills_filter._deterministic_gate.__defaults__[-1] is None
+
def test_disabled_skill_fails_eligibility():
spec = _skill("code-task")
diff --git a/tests/test_skills_default_prompt_contract.py b/tests/test_skills_default_prompt_contract.py
index ee8174593..2c8e6f10b 100644
--- a/tests/test_skills_default_prompt_contract.py
+++ b/tests/test_skills_default_prompt_contract.py
@@ -487,6 +487,7 @@ def fail_get_embedder(*_args: object, **_kwargs: object) -> object:
assert "weather-local" in prompt
assert "github-local" not in prompt
assert ctx.metadata["filtered_skill_ids"] == ["weather-local"]
+ assert ctx.metadata["filtered_skill_reasons"] == {"github-local": "not_in_top_k"}
assert ctx.metadata["skill_count"] == 1