feat(analyzer): add phase-1 structured skill summaries

src/skillspector/multi_skill.py

@@ -32,6 +32,7 @@
     validate_local_input_path,
 )
 from skillspector.logging_config import get_logger
+from skillspector.structured_skill import _SKIP_DIRS, extract_structured_skill_context
 
 logger = get_logger(__name__)
 
@@ -81,9 +82,11 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult:
     for child in sorted(directory.iterdir()):
         if _is_link_or_junction(child) or not child.is_dir():
             continue
+        if child.name in _SKIP_DIRS:
+            continue
         if child.name.startswith("."):
             continue
-        if _has_skill_md(child):
+        if _has_skill_md(child) or _is_structured_skill_bundle(child):
             name = _extract_skill_name(child)
             skills.append(
                 SkillDirectory(
@@ -101,6 +104,11 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult:
     )
 
 
+def _is_structured_skill_bundle(child_dir: Path) -> bool:
+    """Return true when a child directory contains a valid AISOP/AISP bundle."""
+    return extract_structured_skill_context(child_dir) is not None
+
+
 def _has_skill_md(directory: Path) -> bool:
     """Check if directory contains a SKILL.md or skill.md at root level."""
     return any(

src/skillspector/nodes/analyzers/__init__.py

@@ -79,6 +79,9 @@
     node as static_patterns_tool_misuse_node,
 )
 from skillspector.nodes.analyzers.static_yara import node as static_yara_node
+from skillspector.nodes.analyzers.structured_skill_roles import (
+    node as structured_skill_roles_node,
+)
 
 ANALYZER_NODE_IDS: list[str] = [
     "static_patterns_prompt_injection",
@@ -102,6 +105,7 @@
     "mcp_least_privilege",
     "mcp_tool_poisoning",
     "mcp_rug_pull",
+    "structured_skill_roles",
     "semantic_security_discovery",
     "semantic_developer_intent",
     "semantic_quality_policy",
@@ -129,6 +133,7 @@
     "mcp_least_privilege": mcp_least_privilege_node,
     "mcp_tool_poisoning": mcp_tool_poisoning_node,
     "mcp_rug_pull": mcp_rug_pull_node,
+    "structured_skill_roles": structured_skill_roles_node,
     "semantic_security_discovery": semantic_security_discovery_node,
     "semantic_developer_intent": semantic_developer_intent_node,
     "semantic_quality_policy": semantic_quality_policy_node,

src/skillspector/nodes/analyzers/structured_skill_roles.py

@@ -0,0 +1,62 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Structured skill role summary analyzer (SSR-*)."""
+
+from __future__ import annotations
+
+from skillspector.state import AnalyzerNodeResponse, SkillspectorState
+
+ANALYZER_ID = "structured_skill_roles"
+
+
+def _string_list(value: object) -> list[str]:
+    """Return a compact list of string values for summary payload fields."""
+    if not isinstance(value, list):
+        return []
+    return [str(item) for item in value if str(item)]
+
+
+def _build_summary(context: dict[str, object]) -> dict[str, object]:
+    """Build a single SSR-1 structured summary from validated context."""
+    protocol = str(context.get("protocol", "AISOP/AISP"))
+    layout_kind = str(context.get("layout_kind", "structured"))
+    bundle_path = str(context.get("bundle_path", ""))
+    declared_tools = sorted(_string_list(context.get("declared_tools")))
+    workflow_nodes = _string_list(context.get("workflow_nodes"))
+    constraints = _string_list(context.get("constraint_anchors"))
+    resources = _string_list(context.get("resource_anchors"))
+
+    return {
+        "id": "SSR-1",
+        "message": f"Structured {layout_kind} bundle detected ({protocol})",
+        "file": bundle_path,
+        "protocol": protocol,
+        "layout_kind": layout_kind,
+        "declared_tools": declared_tools,
+        "workflow_nodes": workflow_nodes,
+        "constraints": constraints,
+        "resources": resources,
+        "tags": ["AISOP", "AISP", "structured-skill"],
+    }
+
+
+def node(state: SkillspectorState) -> AnalyzerNodeResponse:
+    """Emit one SSR-1 structured summary when structured context is present."""
+    context = state.get("structured_skill_context")
+    if not isinstance(context, dict):
+        return {"findings": []}
+
+    return {"findings": [], "structured_summaries": [_build_summary(context)]}

src/skillspector/nodes/build_context.py

@@ -48,6 +48,7 @@
 from skillspector.logging_config import get_logger
 from skillspector.python_ast import prewarm_python_ast_cache
 from skillspector.state import SkillspectorState
+from skillspector.structured_skill import extract_structured_skill_context
 
 logger = get_logger(__name__)
 
@@ -579,8 +580,9 @@ def build_context(state: SkillspectorState) -> dict[str, object]:
     component_metadata, has_executable_scripts = _build_component_metadata(
         skill_dir, metadata_components, file_cache, recognized_oms_signatures
     )
+    structured_skill_context = extract_structured_skill_context(skill_dir)
 
-    return {
+    result = {
         "components": components,
         "file_cache": file_cache,
         "inspection_ledger": [
@@ -597,3 +599,8 @@ def build_context(state: SkillspectorState) -> dict[str, object]:
         "component_metadata": component_metadata,
         "has_executable_scripts": has_executable_scripts,
     }
+
+    if structured_skill_context is not None:
+        result["structured_skill_context"] = structured_skill_context
+
+    return result

src/skillspector/nodes/report.py

@@ -136,6 +136,24 @@ def _build_sarif_properties(finding: Finding) -> dict[str, object] | None:
     return cleaned or None
 
 
+def _sanitize_summary_value(value: object) -> object:
+    """Return a recursively sanitized copy of structured-summary content."""
+    if isinstance(value, str):
+        return _clean_text(value)
+    if isinstance(value, list):
+        return [_sanitize_summary_value(item) for item in value]
+    if isinstance(value, tuple):
+        return [_sanitize_summary_value(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _sanitize_summary_value(item) for key, item in value.items()}
+    return value
+
+
+def _sanitize_structured_summary(summary: dict[str, object]) -> dict[str, object]:
+    """Return a structured summary with control/ANSI bytes stripped from text fields."""
+    return {str(key): _sanitize_summary_value(value) for key, value in summary.items()}
+
+
 def _severity_to_sarif_level(severity: str) -> Literal["error", "warning", "note"]:
     """Map Finding.severity to SARIF result level."""
     return {
@@ -146,6 +164,36 @@ def _severity_to_sarif_level(severity: str) -> Literal["error", "warning", "note
     }.get(severity.upper(), "note")  # type: ignore[return-value]
 
 
+def _summary_display_value(value: object) -> str | None:
+    """Format a summary field for terminal / markdown output."""
+    if value is None:
+        return None
+    if isinstance(value, list):
+        values = [str(item) for item in value if str(item)]
+        return ", ".join(values) if values else None
+    text = str(value)
+    return text or None
+
+
+def _structured_summary_notification(summary: dict[str, object]) -> str:
+    """Build a note-level SARIF notification message for a structured summary."""
+    summary_id = str(summary.get("id") or "SSR")
+    message = str(summary.get("message") or "Structured skill summary")
+    bits = [f"{summary_id}: {message}"]
+
+    file = _summary_display_value(summary.get("file"))
+    if file:
+        bits.append(f"file={file}")
+    protocol = _summary_display_value(summary.get("protocol"))
+    if protocol:
+        bits.append(f"protocol={protocol}")
+    layout_kind = _summary_display_value(summary.get("layout_kind"))
+    if layout_kind:
+        bits.append(f"layout={layout_kind}")
+
+    return " | ".join(bits)
+
+
 _SEVERITY_POINTS: dict[str, int] = {
     "CRITICAL": 50,
     "HIGH": 25,
@@ -250,6 +298,7 @@ def _build_sarif(
     degraded_notice: str | None = None,
     analysis_completeness: Mapping[str, object] | None = None,
     execution_successful: bool = True,
+    structured_summaries: list[dict[str, object]] | None = None,
 ) -> dict[str, object]:
     """Build one SARIF invocation with canonical inspection notifications."""
     results: list[SarifResult] = []
@@ -316,6 +365,14 @@ def _build_sarif(
 
     notifications: list[SarifNotification] = []
     completeness = analysis_completeness or {}
+    for summary in structured_summaries or []:
+        notifications.append(
+            SarifNotification(
+                message=SarifMessage(text=_structured_summary_notification(summary)),
+                level="note",
+                properties={"kind": "structured_summary"},
+            )
+        )
 
     def notification_from_exception(
         exception: Mapping[str, object], level: Literal["error", "warning", "note"]
@@ -474,6 +531,7 @@ def _format_terminal(
     use_llm: bool = True,
     llm_call_log: Sequence[Mapping[str, object]] | None = None,
     suppressed: list[SuppressedFinding] | None = None,
+    structured_summaries: list[dict[str, object]] | None = None,
     show_suppressed: bool = False,
     analysis_completeness: Mapping[str, object] | None = None,
     execution_successful: bool = True,
@@ -562,6 +620,30 @@ def _format_terminal(
     else:
         console.print("\n[green]No security issues detected.[/green]\n")
 
+    if structured_summaries:
+        console.print("\n")
+        console.print(f"[bold]Structured Skill Summary ({len(structured_summaries)})[/bold]\n")
+        for summary in structured_summaries:
+            console.print(
+                f"  [cyan]{summary.get('id', 'SSR-1')}[/cyan]: {summary.get('message', '')}"
+            )
+            file = _summary_display_value(summary.get("file"))
+            if file:
+                console.print(f"    [dim]File:[/dim] {file}")
+            for key, label in (
+                ("protocol", "Protocol"),
+                ("layout_kind", "Layout"),
+                ("declared_tools", "Declared tools"),
+                ("workflow_nodes", "Workflow nodes"),
+                ("constraints", "Constraints"),
+                ("resources", "Resources"),
+                ("tags", "Tags"),
+            ):
+                value = _summary_display_value(summary.get(key))
+                if value:
+                    console.print(f"    [dim]{label}:[/dim] {value}")
+            console.print()
+
     if suppressed:
         console.print(
             f"[dim]Suppressed by baseline: {len(suppressed)} (not counted toward risk score)[/dim]"
@@ -675,6 +757,7 @@ def _format_json(
     analysis_completeness: Mapping[str, object] | None = None,
     suppressed: list[SuppressedFinding] | None = None,
     execution_successful: bool = True,
+    structured_summaries: list[dict[str, object]] | None = None,
 ) -> str:
     """Generate JSON report string."""
     suppressed = suppressed or []
@@ -700,6 +783,7 @@ def _format_json(
             }
             for c in component_metadata
         ],
+        "structured_summaries": structured_summaries or [],
         "issues": [f.to_dict() for f in findings],
         "suppressed_count": len(suppressed),
         "suppressed": [sf.to_dict() for sf in suppressed],
@@ -786,6 +870,7 @@ def _format_markdown(
     use_llm: bool = True,
     llm_call_log: Sequence[Mapping[str, object]] | None = None,
     suppressed: list[SuppressedFinding] | None = None,
+    structured_summaries: list[dict[str, object]] | None = None,
     show_suppressed: bool = False,
     analysis_completeness: Mapping[str, object] | None = None,
     execution_successful: bool = True,
@@ -827,6 +912,28 @@ def _format_markdown(
         lines.append(f"| `{path}` | {typ} | {line_count} | {exec_marker} |")
     lines.append("")
 
+    if structured_summaries:
+        lines.append(f"## Structured Skill Summary ({len(structured_summaries)})\n")
+        for summary in structured_summaries:
+            lines.append(f"### {summary.get('id', 'SSR-1')}\n")
+            lines.append(f"**Message:** {summary.get('message', '')}  ")
+            file = _summary_display_value(summary.get("file"))
+            if file:
+                lines.append(f"**File:** `{file}`  ")
+            for key, label in (
+                ("protocol", "Protocol"),
+                ("layout_kind", "Layout"),
+                ("declared_tools", "Declared tools"),
+                ("workflow_nodes", "Workflow nodes"),
+                ("constraints", "Constraints"),
+                ("resources", "Resources"),
+                ("tags", "Tags"),
+            ):
+                value = _summary_display_value(summary.get(key))
+                if value:
+                    lines.append(f"**{label}:** {value}  ")
+            lines.append("")
+
     lines.append(f"## Issues ({len(findings)})\n")
     if not findings:
         lines.append("No security issues detected.\n")
@@ -892,6 +999,13 @@ def report(state: SkillspectorState) -> dict[str, object]:
         selected_findings = state.get("filtered_findings", raw_findings)
     selected_findings = [_sanitize_finding(finding) for finding in selected_findings]
 
+    raw_structured_summaries = state.get("structured_summaries") or []
+    structured_summaries = [
+        _sanitize_structured_summary(summary)
+        for summary in raw_structured_summaries
+        if isinstance(summary, dict)
+    ]
+
     empty_completeness: AnalysisCompleteness = {
         "total_components": 0,
         "scanned_components": 0,
@@ -966,6 +1080,7 @@ def report(state: SkillspectorState) -> dict[str, object]:
         degraded_notice=degraded_notice,
         analysis_completeness=analysis_completeness,
         execution_successful=execution_successful,
+        structured_summaries=structured_summaries,
     )
     if output_format == "terminal":
         report_body = _format_terminal(
@@ -980,6 +1095,7 @@ def report(state: SkillspectorState) -> dict[str, object]:
             use_llm=use_llm,
             llm_call_log=llm_call_log,
             suppressed=suppressed,
+            structured_summaries=structured_summaries,
             show_suppressed=show_suppressed,
             analysis_completeness=analysis_completeness,
             execution_successful=execution_successful,
@@ -1000,6 +1116,7 @@ def report(state: SkillspectorState) -> dict[str, object]:
             analysis_completeness=analysis_completeness,
             suppressed=suppressed,
             execution_successful=execution_successful,
+            structured_summaries=structured_summaries,
         )
     elif output_format == "markdown":
         report_body = _format_markdown(
@@ -1014,6 +1131,7 @@ def report(state: SkillspectorState) -> dict[str, object]:
             use_llm=use_llm,
             llm_call_log=llm_call_log,
             suppressed=suppressed,
+            structured_summaries=structured_summaries,
             show_suppressed=show_suppressed,
             analysis_completeness=analysis_completeness,
             execution_successful=execution_successful,

src/skillspector/state.py

@@ -112,6 +112,10 @@ class SkillspectorState(TypedDict, total=False):
     # Component metadata for reporting and risk scoring (from build_context)
     component_metadata: list[dict[str, object]]
     has_executable_scripts: bool
+    # Structured workflow context for phase-1 AISOP/AISP summaries
+    structured_skill_context: dict[str, object]
+    # Report-only structured skill summaries emitted outside the finding pipeline
+    structured_summaries: Annotated[list[dict[str, object]], operator.add]
 
     # Output: report node writes formatted string here
     output_format: str
@@ -158,6 +162,7 @@ class AnalyzerNodeResponse(TypedDict):
     findings: list[Finding]
     inspection_ledger: NotRequired[list[InspectionLedgerEvent]]
     analyzer_status_events: NotRequired[list[AnalyzerStatusEvent]]
+    structured_summaries: NotRequired[list[dict[str, object]]]
     # LLM-backed analyzers also report one telemetry record; static analyzers
     # omit it (NotRequired keeps the key optional for them).
     llm_call_log: NotRequired[list[LLMCallRecord]]