feat(validation): add semantic validation service

Added SemanticValidationService with built-in Python checks for syntax
errors, missing imports, broken references, dependency cycles, API misuse,
and missing symbols. Implemented SemanticRuleRegistry for extensible
validator registration. Integrated with ValidationPipeline as informational
by default. Added per-project/per-plan config keys, severity mapping
(info/warn/error), output schema normalization, and file-hash-based caching.

Includes Behave BDD scenarios, Robot integration tests, ASV benchmarks,
and reference documentation.

ISSUES CLOSED: #207

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Added semantic validation service with AST-based rules for syntax errors, missing imports,
+  broken references, duplicate imports, API misuse, and missing symbols. Includes rule registry,
+  file-hash LRU cache, severity mapping, and ValidationPipeline integration. (#448)
 - Added comprehensive M6 autonomy acceptance test suite covering ACP local facade dispatch
   (session/plan/registry/context/event operations), event queue pub/sub with local callbacks
   and close semantics, HTTP transport stub rejection, version negotiation, automation profile

benchmarks/semantic_validation_bench.py

@@ -0,0 +1,235 @@
+"""ASV benchmarks for the semantic validation service.
+
+Measures the performance of:
+- Individual rule checks (syntax, import, reference, cycle, misuse, symbol)
+- Rule registry operations (register, lookup, list)
+- Validation cache operations (put, get, compute_hash)
+- Service orchestration (check_file, as_pipeline_results)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.semantic_validation_rules import (
+    APIMisuseRule,
+    BrokenReferenceRule,
+    DuplicateImportRule,
+    MissingImportRule,
+    MissingSymbolRule,
+    SemanticCheckResult,
+    SemanticValidationSeverity,
+    SyntaxCheckRule,
+)
+from cleveragents.application.services.semantic_validation_service import (
+    SemanticRuleRegistry,
+    SemanticValidationCache,
+    SemanticValidationService,
+    create_default_registry,
+)
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+VALID_PYTHON = """\
+import os
+import sys
+import json
+
+x = 10
+
+def hello():
+    return x
+
+class Greeter:
+    def greet(self):
+        return hello()
+
+def compute():
+    return x + 1
+"""
+
+LARGE_PYTHON = VALID_PYTHON * 50  # ~800 lines
+
+
+# ---------------------------------------------------------------------------
+# Rule benchmark suites
+# ---------------------------------------------------------------------------
+
+
+class SyntaxCheckRuleSuite:
+    """Benchmarks for SyntaxCheckRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = SyntaxCheckRule()
+        self.source = VALID_PYTHON
+        self.large = LARGE_PYTHON
+
+    def time_syntax_check_small(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+    def time_syntax_check_large(self) -> None:
+        self.rule.check(self.large, "test.py")
+
+
+class MissingImportRuleSuite:
+    """Benchmarks for MissingImportRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = MissingImportRule()
+        self.source = VALID_PYTHON
+
+    def time_import_check(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+
+class BrokenReferenceRuleSuite:
+    """Benchmarks for BrokenReferenceRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = BrokenReferenceRule()
+        self.source = VALID_PYTHON
+
+    def time_reference_check(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+
+class APIMisuseRuleSuite:
+    """Benchmarks for APIMisuseRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = APIMisuseRule()
+        self.source = VALID_PYTHON
+
+    def time_api_misuse_check(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+
+class DuplicateImportRuleSuite:
+    """Benchmarks for DuplicateImportRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = DuplicateImportRule()
+        self.source = VALID_PYTHON
+
+    def time_duplicate_import_check(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+
+class MissingSymbolRuleSuite:
+    """Benchmarks for MissingSymbolRule."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.rule = MissingSymbolRule()
+        self.source = VALID_PYTHON
+
+    def time_symbol_check(self) -> None:
+        self.rule.check(self.source, "test.py")
+
+
+# ---------------------------------------------------------------------------
+# Registry benchmarks
+# ---------------------------------------------------------------------------
+
+
+class RegistrySuite:
+    """Benchmarks for SemanticRuleRegistry."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.registry = create_default_registry()
+
+    def time_list_rules(self) -> None:
+        self.registry.list_rules()
+
+    def time_lookup_rule(self) -> None:
+        self.registry.get("syntax_error")
+
+    def time_all_rules(self) -> None:
+        self.registry.all_rules()
+
+
+# ---------------------------------------------------------------------------
+# Cache benchmarks
+# ---------------------------------------------------------------------------
+
+
+class CacheSuite:
+    """Benchmarks for SemanticValidationCache."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.cache = SemanticValidationCache()
+        self.result = SemanticCheckResult(
+            passed=True,
+            message="ok",
+            severity=SemanticValidationSeverity.INFO,
+        )
+        self.cache.put("rule", "hash1", self.result)
+
+    def time_compute_hash(self) -> None:
+        SemanticValidationCache.compute_hash(VALID_PYTHON)
+
+    def time_cache_get_hit(self) -> None:
+        self.cache.get("rule", "hash1")
+
+    def time_cache_get_miss(self) -> None:
+        self.cache.get("rule", "missing")
+
+    def time_cache_put(self) -> None:
+        self.cache.put("rule", "hash2", self.result)
+
+
+# ---------------------------------------------------------------------------
+# Service benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ServiceSuite:
+    """Benchmarks for SemanticValidationService orchestration."""
+
+    timeout = 120
+
+    def setup(self) -> None:
+        self.service = SemanticValidationService()
+        self.source = VALID_PYTHON
+        self.large = LARGE_PYTHON
+
+    def time_check_file_small(self) -> None:
+        # Clear cache to measure full check cost
+        self.service.cache.clear()
+        self.service.check_file(self.source, "test.py")
+
+    def time_check_file_large(self) -> None:
+        self.service.cache.clear()
+        self.service.check_file(self.large, "test.py")
+
+    def time_check_file_cached(self) -> None:
+        # First call populates cache, second uses it
+        self.service.check_file(self.source, "test.py")
+        self.service.check_file(self.source, "test.py")
+
+    def time_pipeline_results(self) -> None:
+        self.service.cache.clear()
+        self.service.as_pipeline_results(self.source, "test.py")

docs/reference/semantic_validation.md

@@ -0,0 +1,265 @@
+# Semantic Validation
+
+The **Semantic Validation Service** provides lightweight, AST-based semantic
+analysis for Python projects during the strategize/execute phases. It detects
+syntax errors, missing imports, broken references, duplicate imports, API
+misuse, and missing symbols before code is applied. Non-Python files are
+automatically skipped.
+
+## Overview
+
+Semantic checks are exposed as Validation tools attachable per resource via the
+Tool Registry. Results integrate into the `ValidationPipeline` as informational
+by default, meaning failures are reported but do not block the plan lifecycle
+unless configured otherwise.
+
+```
+Source Code ──► SemanticValidationService.check_file() ──► [SemanticCheckResult, ...]
+                        │
+                        ├── SyntaxCheckRule
+                        ├── MissingImportRule
+                        ├── BrokenReferenceRule
+                        ├── DuplicateImportRule
+                        ├── APIMisuseRule
+                        └── MissingSymbolRule
+```
+
+> **Note:** `check_file()` returns `[]` immediately for non-Python files
+> (files without `.py` or `.pyi` extension).
+
+## Built-in Rules
+
+| Rule               | Description                                           | Default Severity |
+|--------------------|-------------------------------------------------------|-----------------|
+| `syntax_error`     | Parses source via `ast.parse`                          | `error`         |
+| `missing_import`   | Flags imports not in `sys.stdlib_module_names`          | `warn`          |
+| `broken_reference` | Detects names not defined in any scope (scope-aware)    | `warn`          |
+| `duplicate_import` | Finds duplicate relative imports                        | `error`         |
+| `api_misuse`       | AST-based detection of dangerous calls (see below)      | `warn`          |
+| `missing_symbol`   | Finds undefined names in functions and class methods    | `info`          |
+
+> **Rename:** The rule was previously registered as `dependency_cycle` and is
+> now registered as `duplicate_import` to better reflect its purpose.
+> The class alias `DependencyCycleRule` remains available for backward
+> compatibility.
+
+### APIMisuseRule Dangerous Calls
+
+The `api_misuse` rule uses AST analysis (not regex) to detect calls to:
+
+- `eval()`, `exec()`, `compile()`
+- `__import__()`
+- `os.system()`, `os.popen()`
+- `subprocess.call()`, `subprocess.run()`, `subprocess.Popen()`
+- `pickle.load()`, `pickle.loads()`
+- `marshal.loads()`
+
+Because the rule inspects `ast.Call` nodes, it does **not** false-positive on
+string literals that happen to contain these function names.
+
+## Severity Mapping
+
+Each rule produces findings with a severity level. The severity determines
+how the finding integrates with the `ValidationPipeline`:
+
+| Severity | Pipeline Mode    | Behaviour                          |
+|----------|------------------|------------------------------------|
+| `error`  | `required`       | Failure **blocks** the operation   |
+| `warn`   | `informational`  | Failure is reported, does not block |
+| `info`   | `informational`  | Informational only                  |
+
+The mapping is configurable via `validation.semantic.severity_mapping`.
+
+## Required vs Informational Attachment Modes
+
+Semantic validations attach to the `ValidationPipeline` in one of two modes:
+
+- **Required** (`error` severity): If a required semantic validation fails,
+  `ValidationSummary.all_required_passed` becomes `False`, blocking the plan
+  from proceeding to the Apply phase.
+
+- **Informational** (`warn`/`info` severity): If an informational validation
+  fails, the finding is logged at INFO level for review but does **not**
+  affect the `all_required_passed` property.
+
+This allows teams to start with informational semantic checks and
+progressively promote them to required as confidence grows.
+
+## Configuration
+
+Three configuration keys control semantic validation:
+
+| Key                                      | Type   | Default | Description                            |
+|------------------------------------------|--------|---------|----------------------------------------|
+| `validation.semantic.enabled`            | `bool` | `true`  | Global on/off for semantic validation  |
+| `validation.semantic.python.enabled`     | `bool` | `true`  | Python-specific checks on/off          |
+| `validation.semantic.severity_mapping`   | `dict` | (below) | Per-rule severity overrides            |
+
+Default severity mapping:
+
+```yaml
+validation.semantic.severity_mapping:
+  syntax_error: error
+  missing_import: warn
+  broken_reference: warn
+  duplicate_import: error
+  api_misuse: warn
+  missing_symbol: info
+```
+
+### Per-project Configuration
+
+Set `validation.semantic.enabled: false` in the project configuration to
+disable semantic validation for a specific project:
+
+```yaml
+# .cleveragents/config.yaml
+validation:
+  semantic:
+    enabled: false
+```
+
+### Per-plan Configuration
+
+Override in plan metadata to enable/disable per plan execution:
+
+```yaml
+validation:
+  semantic:
+    enabled: true
+    python:
+      enabled: true
+```
+
+## Rule Registry
+
+The `SemanticRuleRegistry` manages pluggable validation rules:
+
+```python
+from cleveragents.application.services import (
+    SemanticRuleRegistry,
+    create_default_registry,
+)
+
+# Create with built-in rules
+registry = create_default_registry()
+
+# List registered rules
+print(registry.list_rules())
+# ['api_misuse', 'broken_reference', 'duplicate_import',
+#  'missing_import', 'missing_symbol', 'syntax_error']
+#
+# Note: 'duplicate_import' is provided by DuplicateImportRule
+
+# Look up a rule
+rule = registry.get("syntax_error")
+
+# Register a custom rule
+registry.register(my_custom_rule)
+```
+
+### Custom Rule Protocol
+
+Any object implementing the `SemanticValidationRule` protocol can be
+registered:
+
+```python
+class MyCustomRule:
+    @property
+    def name(self) -> str:
+        return "my_custom_check"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        # ... run analysis ...
+        return SemanticCheckResult(
+            passed=True,
+            message="All good",
+        )
+```
+
+## Caching
+
+The `SemanticValidationCache` avoids re-running checks on unchanged files
+by caching results keyed by `(rule_name, file_sha256)`. The cache uses an
+**LRU (Least Recently Used)** eviction policy with a configurable
+`max_size` (default: 4096 entries). All operations are **thread-safe**.
+
+```python
+from cleveragents.application.services import SemanticValidationCache
+
+cache = SemanticValidationCache(max_size=256)
+file_hash = cache.compute_hash(source_code)
+
+# Check cache before running expensive rule
+cached = cache.get("syntax_error", file_hash)
+if cached is None:
+    result = rule.check(source_code, filename)
+    cache.put("syntax_error", file_hash, result)
+```
+
+The service automatically uses caching when `check_file()` is called.
+Calling `check_file()` with the same source content will return cached
+results without re-executing rules.
+
+When the cache reaches `max_size`, the least-recently-used entry is
+automatically evicted to make room for the new one.
+
+## Output Schema Normalisation
+
+All semantic check results follow a consistent normalised schema:
+
+```json
+{
+  "passed": true,
+  "message": "No syntax errors in test.py",
+  "data": null
+}
+```
+
+The `data` field contains structured finding details when present:
+
+```json
+{
+  "passed": false,
+  "message": "Syntax error at test.py:3: invalid syntax",
+  "data": {
+    "filename": "test.py",
+    "line": 3,
+    "detail": "invalid syntax"
+  }
+}
+```
+
+The `normalise_output()` method on `SemanticValidationService` produces
+this schema from any `SemanticCheckResult`.
+
+## Pipeline Integration
+
+Use `as_pipeline_results()` to get results compatible with the
+`ValidationPipeline`:
+
+```python
+service = SemanticValidationService()
+results = service.as_pipeline_results(source, "module.py")
+
+# Each result dict has:
+# - passed: bool
+# - message: str
+# - data: dict | None
+# - severity: str ("info" | "warn" | "error")
+# - mode: str ("required" | "informational")
+```
+
+## API Reference
+
+### `SemanticValidationService`
+
+| Method                | Description                                    |
+|-----------------------|------------------------------------------------|
+| `check_file()`       | Run all (or selected) rules on source code     |
+| `as_pipeline_results()` | Same as `check_file` but returns pipeline dicts |
+| `normalise_output()`  | Normalise a single result to passed/message/data |
+| `enabled`            | Whether semantic validation is globally enabled |
+| `python_enabled`     | Whether Python-specific checks are enabled     |
+| `registry`           | The rule registry instance                     |
+| `cache`              | The result cache instance                      |

features/semantic_validation.feature

@@ -0,0 +1,467 @@
+Feature: Semantic validation service
+  As a plan executor
+  I want semantic validation checks for Python source code
+  So that syntax errors, missing imports, and broken references are caught early
+
+  Background:
+    Given a semantic validation test environment
+
+  # ── Severity enum ──────────────────────────────────────────────
+
+  Scenario: SemanticValidationSeverity has three levels
+    Then the semantic severity enum has values "info", "warn", "error"
+
+  # ── SemanticCheckResult model ──────────────────────────────────
+
+  Scenario: SemanticCheckResult normalised output has passed message data
+    Given a semantic check result with passed true and message "All good"
+    Then the semantic check result passed is true
+    And the semantic check result message is "All good"
+    And the semantic check result data is none
+
+  Scenario: SemanticCheckResult with failure and data
+    Given a semantic check result with passed false and message "Error found" and data
+    Then the semantic check result passed is false
+    And the semantic check result has data key "detail"
+
+  # ── SyntaxCheckRule ────────────────────────────────────────────
+
+  Scenario: SyntaxCheckRule passes for valid Python
+    Given valid Python source code
+    When the syntax check rule runs
+    Then the semantic check result passed is true
+
+  Scenario: SyntaxCheckRule fails for invalid Python
+    Given Python source code with a syntax error
+    When the syntax check rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "Syntax error"
+    And the semantic check result severity is "error"
+
+  # ── MissingImportRule ──────────────────────────────────────────
+
+  Scenario: MissingImportRule passes for valid imports
+    Given Python source with standard imports
+    When the missing import rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingImportRule flags suspicious private imports
+    Given Python source with suspicious private imports
+    When the missing import rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "suspicious import"
+
+  Scenario: MissingImportRule skips on syntax error
+    Given Python source code with a syntax error
+    When the missing import rule runs
+    Then the semantic check result passed is true
+    And the semantic check result message contains "Skipped"
+
+  # ── BrokenReferenceRule ────────────────────────────────────────
+
+  Scenario: BrokenReferenceRule passes for well-defined code
+    Given Python source with all names defined
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule flags undefined names
+    Given Python source with undefined references
+    When the broken reference rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "broken reference"
+
+  # ── DependencyCycleRule ────────────────────────────────────────
+
+  Scenario: DependencyCycleRule passes for no duplicate relative imports
+    Given Python source with unique relative imports
+    When the dependency cycle rule runs
+    Then the semantic check result passed is true
+
+  Scenario: DependencyCycleRule flags duplicate relative imports
+    Given Python source with duplicate relative imports
+    When the dependency cycle rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "Duplicate relative imports"
+
+  # ── APIMisuseRule ──────────────────────────────────────────────
+
+  Scenario: APIMisuseRule passes for clean code
+    Given Python source without API misuse
+    When the API misuse rule runs
+    Then the semantic check result passed is true
+
+  Scenario: APIMisuseRule flags eval usage
+    Given Python source using eval
+    When the API misuse rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "API misuse"
+
+  # ── MissingSymbolRule ──────────────────────────────────────────
+
+  Scenario: MissingSymbolRule passes for well-scoped functions
+    Given Python source with well-scoped functions
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule flags undefined symbols in functions
+    Given Python source with undefined symbols in functions
+    When the missing symbol rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "missing symbol"
+
+  # ── SemanticRuleRegistry ───────────────────────────────────────
+
+  Scenario: Default registry has all six built-in rules
+    Given a default semantic rule registry
+    Then the registry has 6 rules
+    And the registry contains rule "syntax_error"
+    And the registry contains rule "missing_import"
+    And the registry contains rule "broken_reference"
+    And the registry contains rule "duplicate_import"
+    And the registry contains rule "api_misuse"
+    And the registry contains rule "missing_symbol"
+
+  Scenario: Registry lookup returns None for unknown rule
+    Given a default semantic rule registry
+    Then the registry returns None for rule "nonexistent"
+
+  Scenario: Registry remove returns true for existing rule
+    Given a default semantic rule registry
+    When the rule "syntax_error" is removed from the registry
+    Then the registry has 5 rules
+    And the registry returns None for rule "syntax_error"
+
+  Scenario: Registry remove returns false for unknown rule
+    Given a default semantic rule registry
+    Then removing rule "nonexistent" returns false
+
+  # ── SemanticValidationCache ────────────────────────────────────
+
+  Scenario: Cache returns None for uncached entries
+    Given a semantic validation cache
+    Then the cache returns None for rule "check" and hash "abc123"
+
+  Scenario: Cache stores and retrieves results by rule and hash
+    Given a semantic validation cache
+    And a cached semantic result for rule "check" and hash "abc123"
+    Then the cache returns the result for rule "check" and hash "abc123"
+
+  Scenario: Cache invalidation removes entries
+    Given a semantic validation cache
+    And a cached semantic result for rule "check" and hash "abc123"
+    When the cache entry for rule "check" and hash "abc123" is invalidated
+    Then the cache returns None for rule "check" and hash "abc123"
+
+  Scenario: Cache clear removes all entries
+    Given a semantic validation cache
+    And a cached semantic result for rule "a" and hash "h1"
+    And a cached semantic result for rule "b" and hash "h2"
+    When the semantic cache is cleared
+    Then the semantic cache has 0 entries
+
+  Scenario: Cache compute_hash returns consistent SHA-256
+    Then the semantic cache hash for "hello" is consistent
+
+  # ── Severity mapping ───────────────────────────────────────────
+
+  Scenario: ERROR severity maps to REQUIRED mode
+    Then semantic severity "error" maps to validation mode "required"
+
+  Scenario: WARN severity maps to INFORMATIONAL mode
+    Then semantic severity "warn" maps to validation mode "informational"
+
+  Scenario: INFO severity maps to INFORMATIONAL mode
+    Then semantic severity "info" maps to validation mode "informational"
+
+  Scenario: resolve_severity uses default mapping
+    Then resolve_severity for "syntax_error" returns "error"
+    And resolve_severity for "missing_import" returns "warn"
+    And resolve_severity for "duplicate_import" returns "error"
+    And resolve_severity for "missing_symbol" returns "info"
+
+  Scenario: resolve_severity uses custom mapping
+    Given a custom severity mapping with "syntax_error" as "warn"
+    Then resolve_severity with custom mapping for "syntax_error" returns "warn"
+
+  Scenario: resolve_severity falls back to info for unknown rule
+    Then resolve_severity for "unknown_rule" returns "info"
+
+  # ── SemanticValidationService ──────────────────────────────────
+
+  Scenario: Service check_file runs all rules on valid Python
+    Given a semantic validation service
+    And valid Python source code
+    When the service checks the file
+    Then all semantic check results passed
+
+  Scenario: Service check_file detects syntax error
+    Given a semantic validation service
+    And Python source code with a syntax error
+    When the service checks the file
+    Then at least one semantic check result failed
+    And a failing result has message containing "Syntax error"
+
+  Scenario: Service check_file with specific rules
+    Given a semantic validation service
+    And valid Python source code
+    When the service checks the file with rules "syntax_error"
+    Then only 1 semantic check result is returned
+
+  Scenario: Service returns empty when disabled
+    Given a semantic validation service with semantic disabled
+    And valid Python source code
+    When the service checks the file
+    Then 0 semantic check results are returned
+
+  Scenario: Service returns empty for Python when python disabled
+    Given a semantic validation service with python disabled
+    And valid Python source code
+    When the service checks the python file
+    Then 0 semantic check results are returned
+
+  Scenario: Service uses cache for repeated checks
+    Given a semantic validation service
+    And valid Python source code
+    When the service checks the file twice
+    Then the cache has entries
+
+  # ── Pipeline integration ───────────────────────────────────────
+
+  Scenario: as_pipeline_results returns normalised dicts
+    Given a semantic validation service
+    And valid Python source code
+    When the service returns pipeline results
+    Then each pipeline result has keys "passed" and "message" and "data"
+    And each pipeline result has key "severity"
+    And each pipeline result has key "mode"
+
+  Scenario: normalise_output returns standard schema
+    Given a semantic check result with passed true and message "OK"
+    When the result is normalised
+    Then the normalised output has keys "passed" and "message" and "data"
+
+  # ── Config keys ────────────────────────────────────────────────
+
+  Scenario: Config key constants are defined
+    Then config key "validation.semantic.enabled" exists
+    And config key "validation.semantic.python.enabled" exists
+    And config key "validation.semantic.severity_mapping" exists
+
+  # ── Required vs informational attachment modes ─────────────────
+
+  Scenario: Error severity findings integrate as required in pipeline
+    Given a semantic validation service
+    And Python source code with a syntax error
+    When the service returns pipeline results
+    Then a pipeline result with severity "error" has mode "required"
+
+  Scenario: Warn severity findings integrate as informational in pipeline
+    Given a semantic validation service
+    And Python source using eval
+    When the service returns pipeline results
+    Then a pipeline result with severity "warn" has mode "informational"
+
+  # ── BrokenReferenceRule scope-awareness ─────────────────────────
+
+  Scenario: BrokenReferenceRule passes for function-local variables
+    Given Python source with function-local variables
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for annotated assignments
+    Given Python source with annotated assignment
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for with-statement variables
+    Given Python source with with-statement variable
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  # ── MissingSymbolRule class/nested function support ────────────
+
+  Scenario: MissingSymbolRule detects undefined symbols in class methods
+    Given Python source with a class method using undefined symbol
+    When the missing symbol rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "missing symbol"
+
+  Scenario: MissingSymbolRule passes for nested function calls
+    Given Python source with nested functions
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  # ── MissingSymbolRule comprehension scope ───────────────────────
+
+  Scenario: MissingSymbolRule does not flag comprehension variables
+    Given Python source with comprehension variables
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  # ── DuplicateImportRule alias ──────────────────────────────────
+
+  Scenario: DuplicateImportRule passes for no duplicate relative imports
+    Given Python source with unique relative imports
+    When the duplicate import rule runs
+    Then the semantic check result passed is true
+
+  Scenario: DuplicateImportRule flags duplicate relative imports
+    Given Python source with duplicate relative imports
+    When the duplicate import rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "Duplicate relative imports"
+
+  # ── APIMisuseRule AST-based detection ──────────────────────────
+
+  Scenario: APIMisuseRule flags exec usage
+    Given Python source using exec
+    When the API misuse rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "API misuse"
+
+  Scenario: APIMisuseRule flags os.popen usage
+    Given Python source using os.popen
+    When the API misuse rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "API misuse"
+
+  Scenario: APIMisuseRule flags subprocess.run usage
+    Given Python source using subprocess.run
+    When the API misuse rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "API misuse"
+
+  Scenario: APIMisuseRule ignores eval in string literals
+    Given Python source with eval in a string literal
+    When the API misuse rule runs
+    Then the semantic check result passed is true
+
+  # ── Non-Python file handling ───────────────────────────────────
+
+  Scenario: Service returns empty for non-Python files
+    Given a semantic validation service
+    And valid Python source code
+    And a non-Python filename
+    When the service checks the non-python file
+    Then 0 semantic check results are returned
+
+  # ── Empty source ───────────────────────────────────────────────
+
+  Scenario: Service handles empty source code
+    Given a semantic validation service
+    And empty Python source code
+    When the service checks the file
+    Then all semantic check results passed
+
+  # ── Cache with max size ────────────────────────────────────────
+
+  Scenario: Cache respects max size with LRU eviction
+    Given a semantic validation cache with max size 2
+    And a cached semantic result for rule "a" and hash "h1"
+    And a cached semantic result for rule "b" and hash "h2"
+    And a cached semantic result for rule "c" and hash "h3"
+    Then the semantic cache has 2 entries
+
+  # ── Safe initialisation and cleanup ────────────────────────────
+
+  Scenario: Cache rejects invalid max size on construction
+    Then creating a semantic validation cache with max size 0 raises ValueError
+
+  Scenario: Cache put updates existing entry without growing
+    Given a semantic validation cache
+    And a cached semantic result for rule "r1" and hash "h1"
+    When the cache entry for rule "r1" and hash "h1" is overwritten with message "updated"
+    Then the cache returns the updated result for rule "r1" and hash "h1" with message "updated"
+    And the semantic cache has 1 entries
+
+  Scenario: Registry list_rules returns sorted names
+    Given a default semantic rule registry
+    Then the registry list_rules returns sorted rule names
+
+  Scenario: Service exposes registry property for inspection
+    Given a semantic validation service
+    Then the service registry property returns a registry with 6 rules
+
+  Scenario: resolve_severity falls back to info for invalid severity value
+    Given a custom severity mapping with "syntax_error" as "bogus_value"
+    Then resolve_severity with custom mapping for "syntax_error" returns "info"
+
+  Scenario: Service functions with non-dict severity mapping config
+    Given a semantic validation service with non-dict severity mapping
+    And valid Python source code
+    When the service checks the file
+    Then all semantic check results passed
+
+  # ── MissingImportRule from-import form ─────────────────────────
+
+  Scenario: MissingImportRule flags suspicious from-import with private module
+    Given Python source with from-import of private module
+    When the missing import rule runs
+    Then the semantic check result passed is false
+    And the semantic check result message contains "suspicious import"
+
+  # ── BrokenReferenceRule advanced scope patterns ────────────────
+
+  Scenario: BrokenReferenceRule passes for code with varargs and kwargs
+    Given Python source with varargs and kwargs function
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for from-import bindings
+    Given Python source with from-import bindings
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for except handler variable
+    Given Python source with except handler variable
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for tuple and starred unpacking
+    Given Python source with tuple and starred unpacking
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  Scenario: BrokenReferenceRule passes for with-statement without as clause
+    Given Python source with with-statement without as clause
+    When the broken reference rule runs
+    Then the semantic check result passed is true
+
+  # ── APIMisuseRule safe attribute on flagged module ──────────────
+
+  Scenario: APIMisuseRule passes for safe method call on flagged module
+    Given Python source calling a safe method on a flagged module
+    When the API misuse rule runs
+    Then the semantic check result passed is true
+
+  # ── MissingSymbolRule function-local binding patterns ──────────
+
+  Scenario: MissingSymbolRule passes for function with all parameter kinds
+    Given Python source with function using varargs kwonly and kwargs
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule passes for function with for-loop variable
+    Given Python source with function containing for-loop variable
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule passes for function with with-statement variable
+    Given Python source with function containing with-statement variable
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule passes for function with except handler variable
+    Given Python source with function containing except handler variable
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule passes for function with local import
+    Given Python source with function containing local import
+    When the missing symbol rule runs
+    Then the semantic check result passed is true
+
+  Scenario: MissingSymbolRule passes for function with local from-import
+    Given Python source with function containing local from-import
+    When the missing symbol rule runs
+    Then the semantic check result passed is true

features/steps/semantic_validation_steps.py

@@ -0,0 +1,696 @@
+"""Step definitions for semantic validation scenarios.
+
+Covers SemanticValidationService, built-in rules, rule registry,
+cache, severity mapping, config keys, and pipeline integration.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from behave import given, then, when
+
+if TYPE_CHECKING:
+    from behave.runner import Context
+
+from cleveragents.application.services.semantic_validation_rules import (
+    APIMisuseRule,
+    BrokenReferenceRule,
+    DependencyCycleRule,
+    DuplicateImportRule,
+    MissingImportRule,
+    MissingSymbolRule,
+    SemanticCheckResult,
+    SemanticValidationSeverity,
+    SyntaxCheckRule,
+)
+from cleveragents.application.services.semantic_validation_service import (
+    CONFIG_KEY_ENABLED,
+    CONFIG_KEY_PYTHON_ENABLED,
+    CONFIG_KEY_SEVERITY_MAPPING,
+    DEFAULT_CONFIG,
+    SemanticValidationCache,
+    SemanticValidationService,
+    create_default_registry,
+    map_severity_to_mode,
+    resolve_severity,
+)
+
+# -- Fixtures ----------------------------------------------------------------
+VALID_PY = "import os\n\nx = 10\n\ndef hello():\n    return x\n"
+SYNTAX_ERR_PY = "def broken(\n    return 42\n"
+SUSPICIOUS_IMPORT_PY = "import _nonexistent_private_module\n"
+STD_IMPORT_PY = "import os\nimport sys\nimport json\n"
+ALL_DEFINED_PY = "import os\n\nx = 10\n\ndef foo():\n    return x\n"
+UNDEF_REF_PY = "def foo():\n    return bar_undefined_xyz\n"
+UNIQUE_REL_PY = "from . import alpha\nfrom . import beta\n"
+DUP_REL_PY = "from . import alpha\nfrom . import alpha\n"
+CLEAN_PY = "import json\ndata = json.loads('{}')\n"
+EVAL_PY = "result = eval('1 + 2')\n"
+SCOPED_PY = "import os\n\ndef greet(name):\n    msg = f'Hi {name}'\n    return msg\n"
+UNDEF_SYM_PY = "def compute(x):\n    return x + undefined_thing_xyz\n"
+FUNC_LOCAL_PY = "def process(items):\n    result = []\n    for item in items:\n        result.append(item)\n    return result\n"
+CLASS_METHOD_PY = (
+    "class MyClass:\n    def method(self):\n        return undefined_xyz\n"
+)
+NESTED_FUNC_PY = (
+    "def outer():\n    def inner():\n        return 1\n    return inner()\n"
+)
+ANNASSIGN_PY = "x: int = 10\n\ndef foo():\n    return x\n"
+WITH_STMT_PY = "import io\ndef read_file():\n    with io.open('test.txt') as f:\n        return f.read()\n"
+EXEC_PY = "exec('print(1)')\n"
+OS_POPEN_PY = "import os\nos.popen('ls')\n"
+SUBPROCESS_RUN_PY = "import subprocess\nsubprocess.run(['echo', 'hello'])\n"
+COMPREHENSION_PY = "def summarise(items):\n    return [x * 2 for x in items]\n"
+NON_PY_FILENAME = "config.yaml"
+EMPTY_PY = ""
+
+# -- Fixtures for coverage improvement ----------------------------------------
+FROM_PRIVATE_IMPORT_PY = "from _nonexistent_private_mod import something\n"
+VARARGS_KWARGS_PY = (
+    "def func(*args, **kwargs):\n    return args, kwargs\n\nfunc(1, key=2)\n"
+)
+FROM_IMPORT_BINDINGS_PY = "from os.path import join\n\nresult = join('a', 'b')\n"
+EXCEPT_HANDLER_PY = "try:\n    x = 1\nexcept Exception as exc:\n    y = exc\n"
+TUPLE_STARRED_PY = "a, b = 1, 2\nc, *d = [3, 4, 5]\nresult = a + b + c\n"
+WITH_NO_AS_PY = "import warnings\nwith warnings.catch_warnings():\n    x = 1\n"
+SAFE_FLAGGED_MODULE_PY = "import os\nresult = os.getcwd()\n"
+FUNC_ALL_PARAMS_PY = (
+    "def func(a, /, b, *args, c=1, **kwargs):\n"
+    "    return a + b + c + len(args) + len(kwargs)\n"
+    "\nfunc(1, 2, 3, c=4, d=5)\n"
+)
+FUNC_FOR_LOOP_PY = (
+    "def total(items):\n"
+    "    s = 0\n"
+    "    for item in items:\n"
+    "        s += item\n"
+    "    return s\n"
+)
+FUNC_WITH_STMT_PY = (
+    "def reader(path):\n    with open(path) as fh:\n        return fh.read()\n"
+)
+FUNC_EXCEPT_PY = (
+    "def safe_div(a, b):\n"
+    "    try:\n"
+    "        return a / b\n"
+    "    except ZeroDivisionError as err:\n"
+    "        return err\n"
+)
+FUNC_LOCAL_IMPORT_PY = "def get_cwd():\n    import os\n    return os.getcwd()\n"
+FUNC_LOCAL_FROM_IMPORT_PY = (
+    "def get_path():\n    from os.path import join\n    return join('a', 'b')\n"
+)
+
+
+# -- Background --------------------------------------------------------------
+@given("a semantic validation test environment")
+def step_env(context: Context) -> None:
+    context.sv_result = None
+    context.sv_results = None
+    context.sv_source = ""
+    context.sv_filename = "test.py"
+    context.sv_service = None
+    context.sv_cache = None
+    context.sv_registry = None
+    context.sv_pipeline_results = None
+    context.sv_normalised = None
+    context.sv_custom_mapping = None
+
+
+# -- Severity enum -----------------------------------------------------------
+@then('the semantic severity enum has values "info", "warn", "error"')
+def step_sev_vals(context: Context) -> None:
+    assert SemanticValidationSeverity.INFO == "info"
+    assert SemanticValidationSeverity.WARN == "warn"
+    assert SemanticValidationSeverity.ERROR == "error"
+
+
+# -- SemanticCheckResult model -----------------------------------------------
+@given('a semantic check result with passed true and message "{msg}"')
+def step_cr_pass(context: Context, msg: str) -> None:
+    context.sv_result = SemanticCheckResult(passed=True, message=msg)
+
+
+@given('a semantic check result with passed false and message "{msg}" and data')
+def step_cr_fail(context: Context, msg: str) -> None:
+    context.sv_result = SemanticCheckResult(
+        passed=False,
+        message=msg,
+        data={"detail": "x"},
+        severity=SemanticValidationSeverity.ERROR,
+    )
+
+
+@then("the semantic check result passed is true")
+def step_r_true(context: Context) -> None:
+    assert context.sv_result.passed is True
+
+
+@then("the semantic check result passed is false")
+def step_r_false(context: Context) -> None:
+    assert context.sv_result.passed is False
+
+
+@then('the semantic check result message is "{msg}"')
+def step_r_msg(context: Context, msg: str) -> None:
+    assert context.sv_result.message == msg
+
+
+@then("the semantic check result data is none")
+def step_r_none(context: Context) -> None:
+    assert context.sv_result.data is None
+
+
+@then('the semantic check result has data key "{key}"')
+def step_r_dk(context: Context, key: str) -> None:
+    assert context.sv_result.data is not None and key in context.sv_result.data
+
+
+@then('the semantic check result message contains "{text}"')
+def step_r_mc(context: Context, text: str) -> None:
+    assert text in context.sv_result.message
+
+
+@then('the semantic check result severity is "{sev}"')
+def step_r_sev(context: Context, sev: str) -> None:
+    assert context.sv_result.severity.value == sev
+
+
+# -- Source code given steps -------------------------------------------------
+@given("valid Python source code")
+def step_src_valid(context: Context) -> None:
+    context.sv_source = VALID_PY
+
+
+@given("Python source code with a syntax error")
+def step_src_syn(context: Context) -> None:
+    context.sv_source = SYNTAX_ERR_PY
+
+
+@given("Python source with standard imports")
+def step_src_std(context: Context) -> None:
+    context.sv_source = STD_IMPORT_PY
+
+
+@given("Python source with suspicious private imports")
+def step_src_sus(context: Context) -> None:
+    context.sv_source = SUSPICIOUS_IMPORT_PY
+
+
+@given("Python source with all names defined")
+def step_src_def(context: Context) -> None:
+    context.sv_source = ALL_DEFINED_PY
+
+
+@given("Python source with undefined references")
+def step_src_uref(context: Context) -> None:
+    context.sv_source = UNDEF_REF_PY
+
+
+@given("Python source with unique relative imports")
+def step_src_urel(context: Context) -> None:
+    context.sv_source = UNIQUE_REL_PY
+
+
+@given("Python source with duplicate relative imports")
+def step_src_drel(context: Context) -> None:
+    context.sv_source = DUP_REL_PY
+
+
+@given("Python source without API misuse")
+def step_src_clean(context: Context) -> None:
+    context.sv_source = CLEAN_PY
+
+
+@given("Python source using eval")
+def step_src_eval(context: Context) -> None:
+    context.sv_source = EVAL_PY
+
+
+@given("Python source with well-scoped functions")
+def step_src_scoped(context: Context) -> None:
+    context.sv_source = SCOPED_PY
+
+
+@given("Python source with undefined symbols in functions")
+def step_src_usym(context: Context) -> None:
+    context.sv_source = UNDEF_SYM_PY
+
+
+@given("Python source with function-local variables")
+def step_src_func_local(context: Context) -> None:
+    context.sv_source = FUNC_LOCAL_PY
+
+
+@given("Python source with a class method using undefined symbol")
+def step_src_cls_method(context: Context) -> None:
+    context.sv_source = CLASS_METHOD_PY
+
+
+@given("Python source with nested functions")
+def step_src_nested(context: Context) -> None:
+    context.sv_source = NESTED_FUNC_PY
+
+
+@given("Python source with annotated assignment")
+def step_src_annassign(context: Context) -> None:
+    context.sv_source = ANNASSIGN_PY
+
+
+@given("Python source with with-statement variable")
+def step_src_with_stmt(context: Context) -> None:
+    context.sv_source = WITH_STMT_PY
+
+
+@given("Python source using exec")
+def step_src_exec(context: Context) -> None:
+    context.sv_source = EXEC_PY
+
+
+@given("Python source using os.popen")
+def step_src_os_popen(context: Context) -> None:
+    context.sv_source = OS_POPEN_PY
+
+
+@given("Python source using subprocess.run")
+def step_src_subprocess_run(context: Context) -> None:
+    context.sv_source = SUBPROCESS_RUN_PY
+
+
+@given("Python source with comprehension variables")
+def step_src_comprehension(context: Context) -> None:
+    context.sv_source = COMPREHENSION_PY
+
+
+@given("empty Python source code")
+def step_src_empty(context: Context) -> None:
+    context.sv_source = EMPTY_PY
+
+
+@given("Python source with eval in a string literal")
+def step_src_eval_str(context: Context) -> None:
+    context.sv_source = 'description = "Do not use eval() in production"\n'
+
+
+@given("a non-Python filename")
+def step_non_py_filename(context: Context) -> None:
+    context.sv_filename = NON_PY_FILENAME
+
+
+# -- Rule execution ----------------------------------------------------------
+@when("the syntax check rule runs")
+def step_run_syn(context: Context) -> None:
+    context.sv_result = SyntaxCheckRule().check(context.sv_source, context.sv_filename)
+
+
+@when("the missing import rule runs")
+def step_run_imp(context: Context) -> None:
+    context.sv_result = MissingImportRule().check(
+        context.sv_source, context.sv_filename
+    )
+
+
+@when("the broken reference rule runs")
+def step_run_ref(context: Context) -> None:
+    context.sv_result = BrokenReferenceRule().check(
+        context.sv_source, context.sv_filename
+    )
+
+
+@when("the dependency cycle rule runs")
+def step_run_cyc(context: Context) -> None:
+    context.sv_result = DependencyCycleRule().check(
+        context.sv_source, context.sv_filename
+    )
+
+
+@when("the API misuse rule runs")
+def step_run_api(context: Context) -> None:
+    context.sv_result = APIMisuseRule().check(context.sv_source, context.sv_filename)
+
+
+@when("the missing symbol rule runs")
+def step_run_sym(context: Context) -> None:
+    context.sv_result = MissingSymbolRule().check(
+        context.sv_source, context.sv_filename
+    )
+
+
+@when("the duplicate import rule runs")
+def step_run_dup(context: Context) -> None:
+    context.sv_result = DuplicateImportRule().check(
+        context.sv_source, context.sv_filename
+    )
+
+
+# -- Registry ----------------------------------------------------------------
+@given("a default semantic rule registry")
+def step_reg(context: Context) -> None:
+    context.sv_registry = create_default_registry()
+
+
+@then("the registry has {count:d} rules")
+def step_reg_cnt(context: Context, count: int) -> None:
+    assert len(context.sv_registry) == count
+
+
+@then('the registry contains rule "{name}"')
+def step_reg_has(context: Context, name: str) -> None:
+    assert context.sv_registry.get(name) is not None
+
+
+@then('the registry returns None for rule "{name}"')
+def step_reg_none(context: Context, name: str) -> None:
+    assert context.sv_registry.get(name) is None
+
+
+@when('the rule "{name}" is removed from the registry')
+def step_reg_rm(context: Context, name: str) -> None:
+    context.sv_registry.remove(name)
+
+
+@then('removing rule "{name}" returns false')
+def step_reg_rm_f(context: Context, name: str) -> None:
+    assert context.sv_registry.remove(name) is False
+
+
+# -- Cache -------------------------------------------------------------------
+@given("a semantic validation cache")
+def step_cache(context: Context) -> None:
+    context.sv_cache = SemanticValidationCache()
+
+
+@given("a semantic validation cache with max size {n:d}")
+def step_cache_max(context: Context, n: int) -> None:
+    context.sv_cache = SemanticValidationCache(max_size=n)
+
+
+@given('a cached semantic result for rule "{rule}" and hash "{h}"')
+def step_cache_put(context: Context, rule: str, h: str) -> None:
+    context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message="cached"))
+
+
+@then('the cache returns None for rule "{rule}" and hash "{h}"')
+def step_cache_miss(context: Context, rule: str, h: str) -> None:
+    assert context.sv_cache.get(rule, h) is None
+
+
+@then('the cache returns the result for rule "{rule}" and hash "{h}"')
+def step_cache_hit(context: Context, rule: str, h: str) -> None:
+    r = context.sv_cache.get(rule, h)
+    assert r is not None and r.message == "cached"
+
+
+@when('the cache entry for rule "{rule}" and hash "{h}" is invalidated')
+def step_cache_inv(context: Context, rule: str, h: str) -> None:
+    context.sv_cache.invalidate(rule, h)
+
+
+@when("the semantic cache is cleared")
+def step_cache_clr(context: Context) -> None:
+    context.sv_cache.clear()
+
+
+@then("the semantic cache has {count:d} entries")
+def step_cache_cnt(context: Context, count: int) -> None:
+    assert len(context.sv_cache) == count
+
+
+@then('the semantic cache hash for "{text}" is consistent')
+def step_cache_hash(context: Context, text: str) -> None:
+    h1 = SemanticValidationCache.compute_hash(text)
+    assert h1 == SemanticValidationCache.compute_hash(text) and len(h1) == 64
+
+
+# -- Severity mapping --------------------------------------------------------
+@then('semantic severity "{sev}" maps to validation mode "{mode}"')
+def step_sev_map(context: Context, sev: str, mode: str) -> None:
+    assert map_severity_to_mode(SemanticValidationSeverity(sev)).value == mode
+
+
+@then('resolve_severity for "{rule}" returns "{expected}"')
+def step_resolve(context: Context, rule: str, expected: str) -> None:
+    assert resolve_severity(rule).value == expected
+
+
+@given('a custom severity mapping with "{rule}" as "{sev}"')
+def step_cust_map(context: Context, rule: str, sev: str) -> None:
+    context.sv_custom_mapping = {rule: sev}
+
+
+@then('resolve_severity with custom mapping for "{rule}" returns "{expected}"')
+def step_resolve_c(context: Context, rule: str, expected: str) -> None:
+    assert resolve_severity(rule, context.sv_custom_mapping).value == expected
+
+
+# -- Service -----------------------------------------------------------------
+@given("a semantic validation service")
+def step_svc(context: Context) -> None:
+    context.sv_service = SemanticValidationService()
+
+
+@given("a semantic validation service with semantic disabled")
+def step_svc_off(context: Context) -> None:
+    context.sv_service = SemanticValidationService(config={CONFIG_KEY_ENABLED: False})
+
+
+@given("a semantic validation service with python disabled")
+def step_svc_pyoff(context: Context) -> None:
+    context.sv_service = SemanticValidationService(
+        config={CONFIG_KEY_PYTHON_ENABLED: False},
+    )
+
+
+@when("the service checks the file")
+def step_svc_chk(context: Context) -> None:
+    context.sv_results = context.sv_service.check_file(
+        context.sv_source,
+        context.sv_filename,
+    )
+
+
+@when("the service checks the python file")
+def step_svc_chk_py(context: Context) -> None:
+    context.sv_results = context.sv_service.check_file(context.sv_source, "module.py")
+
+
+@when('the service checks the file with rules "{rules}"')
+def step_svc_chk_r(context: Context, rules: str) -> None:
+    context.sv_results = context.sv_service.check_file(
+        context.sv_source,
+        context.sv_filename,
+        rule_names=[r.strip() for r in rules.split(",")],
+    )
+
+
+@when("the service checks the file twice")
+def step_svc_chk2(context: Context) -> None:
+    context.sv_service.check_file(context.sv_source, context.sv_filename)
+    context.sv_results = context.sv_service.check_file(
+        context.sv_source,
+        context.sv_filename,
+    )
+
+
+@when("the service checks the non-python file")
+def step_svc_chk_nonpy(context: Context) -> None:
+    context.sv_filename = NON_PY_FILENAME
+    context.sv_results = context.sv_service.check_file(
+        context.sv_source,
+        context.sv_filename,
+    )
+
+
+@then("all semantic check results passed")
+def step_all_pass(context: Context) -> None:
+    assert all(r.passed for r in context.sv_results)
+
+
+@then("at least one semantic check result failed")
+def step_some_fail(context: Context) -> None:
+    assert any(not r.passed for r in context.sv_results)
+
+
+@then('a failing result has message containing "{text}"')
+def step_fail_msg(context: Context, text: str) -> None:
+    assert any(text in r.message for r in context.sv_results if not r.passed)
+
+
+@then("only {count:d} semantic check result is returned")
+def step_cnt1(context: Context, count: int) -> None:
+    assert len(context.sv_results) == count
+
+
+@then("{count:d} semantic check results are returned")
+def step_cnt(context: Context, count: int) -> None:
+    assert len(context.sv_results) == count
+
+
+@then("the cache has entries")
+def step_cache_has(context: Context) -> None:
+    assert len(context.sv_service.cache) > 0
+
+
+# -- Pipeline integration ---------------------------------------------------
+@when("the service returns pipeline results")
+def step_pipe(context: Context) -> None:
+    context.sv_pipeline_results = context.sv_service.as_pipeline_results(
+        context.sv_source,
+        context.sv_filename,
+    )
+
+
+@then('each pipeline result has keys "passed" and "message" and "data"')
+def step_pipe_k(context: Context) -> None:
+    for pr in context.sv_pipeline_results:
+        assert "passed" in pr and "message" in pr and "data" in pr
+
+
+@then('each pipeline result has key "{key}"')
+def step_pipe_k1(context: Context, key: str) -> None:
+    for pr in context.sv_pipeline_results:
+        assert key in pr
+
+
+@when("the result is normalised")
+def step_norm(context: Context) -> None:
+    context.sv_normalised = SemanticValidationService().normalise_output(
+        context.sv_result
+    )
+
+
+@then('the normalised output has keys "passed" and "message" and "data"')
+def step_norm_k(context: Context) -> None:
+    n = context.sv_normalised
+    assert "passed" in n and "message" in n and "data" in n
+
+
+# -- Config keys -------------------------------------------------------------
+@then('config key "{key}" exists')
+def step_cfg(context: Context, key: str) -> None:
+    assert key in DEFAULT_CONFIG
+
+
+# -- Required vs informational -----------------------------------------------
+@then('a pipeline result with severity "{sev}" has mode "{mode}"')
+def step_pipe_mode(context: Context, sev: str, mode: str) -> None:
+    matching = [r for r in context.sv_pipeline_results if r["severity"] == sev]
+    assert len(matching) > 0
+    assert all(r["mode"] == mode for r in matching)
+
+
+# -- Safe initialisation and cleanup -----------------------------------------
+@then("creating a semantic validation cache with max size 0 raises ValueError")
+def step_cache_invalid_size(context: Context) -> None:
+    raised = False
+    try:
+        SemanticValidationCache(max_size=0)
+    except ValueError:
+        raised = True
+    assert raised, "Expected ValueError for max_size=0"
+
+
+@when(
+    'the cache entry for rule "{rule}" and hash "{h}" is overwritten '
+    'with message "{msg}"'
+)
+def step_cache_overwrite(context: Context, rule: str, h: str, msg: str) -> None:
+    context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message=msg))
+
+
+@then(
+    'the cache returns the updated result for rule "{rule}" and hash "{h}" '
+    'with message "{msg}"'
+)
+def step_cache_updated(context: Context, rule: str, h: str, msg: str) -> None:
+    r = context.sv_cache.get(rule, h)
+    assert r is not None and r.message == msg
+
+
+@then("the registry list_rules returns sorted rule names")
+def step_reg_list(context: Context) -> None:
+    names = context.sv_registry.list_rules()
+    assert names == sorted(names)
+    assert len(names) == len(context.sv_registry)
+
+
+@then("the service registry property returns a registry with {count:d} rules")
+def step_svc_reg(context: Context, count: int) -> None:
+    assert len(context.sv_service.registry) == count
+
+
+@given("a semantic validation service with non-dict severity mapping")
+def step_svc_nondict_sev(context: Context) -> None:
+    context.sv_service = SemanticValidationService(
+        config={CONFIG_KEY_SEVERITY_MAPPING: "not-a-dict"},
+    )
+
+
+# -- MissingImportRule from-import form --------------------------------------
+@given("Python source with from-import of private module")
+def step_src_from_priv(context: Context) -> None:
+    context.sv_source = FROM_PRIVATE_IMPORT_PY
+
+
+# -- BrokenReferenceRule advanced scope patterns -----------------------------
+@given("Python source with varargs and kwargs function")
+def step_src_varargs(context: Context) -> None:
+    context.sv_source = VARARGS_KWARGS_PY
+
+
+@given("Python source with from-import bindings")
+def step_src_from_import(context: Context) -> None:
+    context.sv_source = FROM_IMPORT_BINDINGS_PY
+
+
+@given("Python source with except handler variable")
+def step_src_except(context: Context) -> None:
+    context.sv_source = EXCEPT_HANDLER_PY
+
+
+@given("Python source with tuple and starred unpacking")
+def step_src_tuple_star(context: Context) -> None:
+    context.sv_source = TUPLE_STARRED_PY
+
+
+@given("Python source with with-statement without as clause")
+def step_src_with_no_as(context: Context) -> None:
+    context.sv_source = WITH_NO_AS_PY
+
+
+# -- APIMisuseRule safe attribute on flagged module --------------------------
+@given("Python source calling a safe method on a flagged module")
+def step_src_safe_flagged(context: Context) -> None:
+    context.sv_source = SAFE_FLAGGED_MODULE_PY
+
+
+# -- MissingSymbolRule function-local binding patterns -----------------------
+@given("Python source with function using varargs kwonly and kwargs")
+def step_src_func_all_params(context: Context) -> None:
+    context.sv_source = FUNC_ALL_PARAMS_PY
+
+
+@given("Python source with function containing for-loop variable")
+def step_src_func_for(context: Context) -> None:
+    context.sv_source = FUNC_FOR_LOOP_PY
+
+
+@given("Python source with function containing with-statement variable")
+def step_src_func_with(context: Context) -> None:
+    context.sv_source = FUNC_WITH_STMT_PY
+
+
+@given("Python source with function containing except handler variable")
+def step_src_func_except(context: Context) -> None:
+    context.sv_source = FUNC_EXCEPT_PY
+
+
+@given("Python source with function containing local import")
+def step_src_func_import(context: Context) -> None:
+    context.sv_source = FUNC_LOCAL_IMPORT_PY
+
+
+@given("Python source with function containing local from-import")
+def step_src_func_from_import(context: Context) -> None:
+    context.sv_source = FUNC_LOCAL_FROM_IMPORT_PY

robot/helper_semantic_validation.py

@@ -0,0 +1,286 @@
+"""Helper script for Robot Framework semantic validation tests.
+
+Self-contained Python helper that exercises the SemanticValidationService
+and related components, printing sentinel strings on success and exiting
+with code 1 on failure.
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+# Ensure src is importable when run from workspace root
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cleveragents.application.services.semantic_validation_rules import (
+    APIMisuseRule,
+    BrokenReferenceRule,
+    DuplicateImportRule,
+    MissingSymbolRule,
+    SemanticCheckResult,
+    SemanticValidationSeverity,
+    SyntaxCheckRule,
+)
+from cleveragents.application.services.semantic_validation_service import (
+    CONFIG_KEY_ENABLED,
+    SemanticValidationCache,
+    SemanticValidationService,
+    create_default_registry,
+    map_severity_to_mode,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+# ---------------------------------------------------------------------------
+# Source code fixtures
+# ---------------------------------------------------------------------------
+
+VALID_PYTHON = """\
+import os
+
+x = 10
+
+def hello():
+    return x
+"""
+
+SYNTAX_ERROR_PYTHON = """\
+def broken(
+    return 42
+"""
+
+EVAL_PYTHON = """\
+result = eval("1 + 2")
+"""
+
+FUNC_LOCAL_PYTHON = """\
+def process(items):
+    result = []
+    for item in items:
+        result.append(item)
+    return result
+"""
+
+CLASS_METHOD_PYTHON = """\
+class MyClass:
+    def method(self):
+        return undefined_xyz
+"""
+
+
+# ---------------------------------------------------------------------------
+# Sub-commands
+# ---------------------------------------------------------------------------
+
+
+def test_models() -> None:
+    """Verify SemanticCheckResult and SemanticValidationSeverity."""
+    result = SemanticCheckResult(
+        passed=True,
+        message="All good",
+        severity=SemanticValidationSeverity.INFO,
+    )
+    assert result.passed is True
+    assert result.message == "All good"
+    assert result.severity == SemanticValidationSeverity.INFO
+    assert result.data is None
+
+    assert SemanticValidationSeverity.INFO == "info"
+    assert SemanticValidationSeverity.WARN == "warn"
+    assert SemanticValidationSeverity.ERROR == "error"
+    print("models-ok")
+
+
+def test_syntax_pass() -> None:
+    """SyntaxCheckRule passes for valid Python."""
+    rule = SyntaxCheckRule()
+    result = rule.check(VALID_PYTHON, "test.py")
+    assert result.passed is True
+    print("syntax-pass-ok")
+
+
+def test_syntax_fail() -> None:
+    """SyntaxCheckRule fails for invalid Python."""
+    rule = SyntaxCheckRule()
+    result = rule.check(SYNTAX_ERROR_PYTHON, "test.py")
+    assert result.passed is False
+    assert "Syntax error" in result.message
+    print("syntax-fail-ok")
+
+
+def test_registry() -> None:
+    """Default registry has all six built-in rules."""
+    registry = create_default_registry()
+    assert len(registry) == 6
+    names = registry.list_rules()
+    assert "syntax_error" in names
+    assert "missing_import" in names
+    assert "broken_reference" in names
+    assert "duplicate_import" in names
+    assert "api_misuse" in names
+    assert "missing_symbol" in names
+    print("registry-ok")
+
+
+def test_cache() -> None:
+    """Cache stores and retrieves correctly."""
+    cache = SemanticValidationCache()
+    assert cache.get("rule", "hash1") is None
+
+    result = SemanticCheckResult(passed=True, message="cached")
+    cache.put("rule", "hash1", result)
+    cached = cache.get("rule", "hash1")
+    assert cached is not None
+    assert cached.message == "cached"
+
+    cache.invalidate("rule", "hash1")
+    assert cache.get("rule", "hash1") is None
+
+    h1 = SemanticValidationCache.compute_hash("hello")
+    h2 = SemanticValidationCache.compute_hash("hello")
+    assert h1 == h2
+    assert len(h1) == 64
+    print("cache-ok")
+
+
+def test_service_check() -> None:
+    """Service runs all rules."""
+    service = SemanticValidationService()
+    results = service.check_file(VALID_PYTHON, "test.py")
+    assert len(results) > 0
+    assert all(r.passed for r in results)
+    print("service-check-ok")
+
+
+def test_severity_mapping() -> None:
+    """Severity maps to correct validation mode."""
+    assert (
+        map_severity_to_mode(SemanticValidationSeverity.ERROR)
+        == ValidationMode.REQUIRED
+    )
+    assert (
+        map_severity_to_mode(SemanticValidationSeverity.WARN)
+        == ValidationMode.INFORMATIONAL
+    )
+    assert (
+        map_severity_to_mode(SemanticValidationSeverity.INFO)
+        == ValidationMode.INFORMATIONAL
+    )
+    print("severity-mapping-ok")
+
+
+def test_pipeline_integration() -> None:
+    """as_pipeline_results returns dicts with expected keys."""
+    service = SemanticValidationService()
+    results = service.as_pipeline_results(VALID_PYTHON, "test.py")
+    assert len(results) > 0
+    for r in results:
+        assert "passed" in r
+        assert "message" in r
+        assert "data" in r
+        assert "severity" in r
+        assert "mode" in r
+    print("pipeline-integration-ok")
+
+
+def test_config_disabled() -> None:
+    """Service returns empty when disabled."""
+    service = SemanticValidationService(config={CONFIG_KEY_ENABLED: False})
+    results = service.check_file(VALID_PYTHON, "test.py")
+    assert len(results) == 0
+    print("config-disabled-ok")
+
+
+def test_api_misuse() -> None:
+    """APIMisuseRule detects eval usage via AST analysis."""
+    rule = APIMisuseRule()
+    result = rule.check(EVAL_PYTHON, "test.py")
+    assert result.passed is False
+    assert "API misuse" in result.message
+    print("api-misuse-ok")
+
+
+def test_broken_ref_scope() -> None:
+    """BrokenReferenceRule does not false-positive on function locals."""
+    rule = BrokenReferenceRule()
+    result = rule.check(FUNC_LOCAL_PYTHON, "test.py")
+    assert result.passed is True, f"Expected pass, got: {result.message}"
+    print("broken-ref-scope-ok")
+
+
+def test_missing_symbol_class() -> None:
+    """MissingSymbolRule detects undefined symbols in class methods."""
+    rule = MissingSymbolRule()
+    result = rule.check(CLASS_METHOD_PYTHON, "test.py")
+    assert result.passed is False, f"Expected fail, got: {result.message}"
+    assert "missing symbol" in result.message
+    print("missing-symbol-class-ok")
+
+
+def test_cache_lru() -> None:
+    """Cache evicts LRU entries when max_size is exceeded."""
+    cache = SemanticValidationCache(max_size=2)
+    r = SemanticCheckResult(passed=True, message="ok")
+    cache.put("a", "h1", r)
+    cache.put("b", "h2", r)
+    cache.put("c", "h3", r)
+    assert len(cache) == 2
+    assert cache.get("a", "h1") is None  # evicted
+    assert cache.get("b", "h2") is not None
+    assert cache.get("c", "h3") is not None
+    print("cache-lru-ok")
+
+
+def test_non_python_skip() -> None:
+    """Service returns empty for non-Python files."""
+    service = SemanticValidationService()
+    results = service.check_file(VALID_PYTHON, "config.yaml")
+    assert len(results) == 0
+    print("non-python-skip-ok")
+
+
+def test_duplicate_import_alias() -> None:
+    """DuplicateImportRule is accessible via DependencyCycleRule alias."""
+    rule = DuplicateImportRule()
+    assert rule.name == "duplicate_import"
+    print("duplicate-import-alias-ok")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Callable[[], None]] = {
+    "models": test_models,
+    "syntax_pass": test_syntax_pass,
+    "syntax_fail": test_syntax_fail,
+    "registry": test_registry,
+    "cache": test_cache,
+    "service_check": test_service_check,
+    "severity_mapping": test_severity_mapping,
+    "pipeline_integration": test_pipeline_integration,
+    "config_disabled": test_config_disabled,
+    "api_misuse": test_api_misuse,
+    "broken_ref_scope": test_broken_ref_scope,
+    "missing_symbol_class": test_missing_symbol_class,
+    "cache_lru": test_cache_lru,
+    "non_python_skip": test_non_python_skip,
+    "duplicate_import_alias": test_duplicate_import_alias,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: helper_semantic_validation.py <command>", file=sys.stderr)
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+    if cmd not in _COMMANDS:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        _COMMANDS[cmd]()
+    except Exception as exc:
+        print(f"FAILED: {exc}", file=sys.stderr)
+        sys.exit(1)

robot/semantic_validation.robot

@@ -0,0 +1,107 @@
+*** Settings ***
+Documentation    Integration tests for the semantic validation service
+Library          Process
+Library          OperatingSystem
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_semantic_validation.py
+
+*** Test Cases ***
+Semantic Validation Models Are Valid
+    [Documentation]    Verify SemanticCheckResult and SemanticValidationSeverity models
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    models    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    models-ok
+
+Semantic Validation Syntax Check Passes For Valid Code
+    [Documentation]    SyntaxCheckRule passes for valid Python source
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    syntax_pass    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    syntax-pass-ok
+
+Semantic Validation Syntax Check Fails For Invalid Code
+    [Documentation]    SyntaxCheckRule fails for code with syntax errors
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    syntax_fail    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    syntax-fail-ok
+
+Semantic Validation Registry Has Built-in Rules
+    [Documentation]    Default registry contains all six built-in rules
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    registry    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    registry-ok
+
+Semantic Validation Cache Works Correctly
+    [Documentation]    Cache stores and retrieves results by rule name and hash
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    cache    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cache-ok
+
+Semantic Validation Service Runs All Rules
+    [Documentation]    SemanticValidationService.check_file runs all registered rules
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    service_check    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    service-check-ok
+
+Semantic Validation Severity Mapping
+    [Documentation]    Severity levels map correctly to validation modes
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    severity_mapping    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    severity-mapping-ok
+
+Semantic Validation Pipeline Integration
+    [Documentation]    as_pipeline_results returns normalised dicts with expected keys
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    pipeline_integration    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    pipeline-integration-ok
+
+Semantic Validation Config Disabled
+    [Documentation]    Service returns empty results when disabled via config
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    config_disabled    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    config-disabled-ok
+
+Semantic Validation API Misuse Detection
+    [Documentation]    APIMisuseRule detects eval() usage
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    api_misuse    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    api-misuse-ok
+
+Semantic Validation Broken Reference Scope Aware
+    [Documentation]    BrokenReferenceRule does not false-positive on function-local variables
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    broken_ref_scope    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    broken-ref-scope-ok
+
+Semantic Validation Missing Symbol Detects Class Methods
+    [Documentation]    MissingSymbolRule detects undefined symbols inside class methods
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    missing_symbol_class    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    missing-symbol-class-ok
+
+Semantic Validation Cache LRU Eviction
+    [Documentation]    Cache evicts least-recently-used entries when max_size exceeded
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    cache_lru    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cache-lru-ok
+
+Semantic Validation Non Python Files Skipped
+    [Documentation]    Service returns empty results for non-Python files
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    non_python_skip    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    non-python-skip-ok
+
+Semantic Validation Duplicate Import Rule Alias
+    [Documentation]    DuplicateImportRule is accessible via DependencyCycleRule backward-compat alias
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    duplicate_import_alias    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    duplicate-import-alias-ok
+
+Semantic Validation Unknown Command Returns Error
+    [Documentation]    Running with unknown command exits with code 1
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    nonexistent_cmd    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    1
+    Should Contain    ${result.stderr}    Unknown command

src/cleveragents/application/services/__init__.py

@@ -23,6 +23,28 @@ from cleveragents.application.services.plan_execution_context import (
     RuntimeExecuteActor,
     RuntimeExecuteResult,
 )
+from cleveragents.application.services.semantic_validation_rules import (
+    APIMisuseRule,
+    BrokenReferenceRule,
+    DependencyCycleRule,
+    DuplicateImportRule,
+    MissingImportRule,
+    MissingSymbolRule,
+    SemanticCheckResult,
+    SemanticValidationSeverity,
+    SyntaxCheckRule,
+)
+from cleveragents.application.services.semantic_validation_service import (
+    NormalisedOutputDict,
+    PipelineResultDict,
+    SemanticRuleRegistry,
+    SemanticValidationCache,
+    SemanticValidationRule,
+    SemanticValidationService,
+    create_default_registry,
+    map_severity_to_mode,
+    resolve_severity,
+)
 from cleveragents.application.services.session_service import (
     PersistentSessionService,
 )
@@ -49,23 +71,38 @@ from cleveragents.application.services.validation_pipeline import (
 )
 
 __all__ = [
+    "APIMisuseRule",
     "ApplyValidationGate",
     "ApplyValidationResult",
     "ApplyValidationSummary",
     "AttachmentScope",
+    "BrokenReferenceRule",
     "ConfigEntry",
     "ConfigLevel",
     "ConfigService",
     "CorrectionService",
     "DecisionService",
     "DefaultValidationRunner",
+    "DependencyCycleRule",
+    "DuplicateImportRule",
     "InvariantService",
+    "MissingImportRule",
+    "MissingSymbolRule",
+    "NormalisedOutputDict",
     "PersistentSessionService",
+    "PipelineResultDict",
     "PlanExecutionContext",
     "ResolvedValue",
     "RuntimeExecuteActor",
     "RuntimeExecuteResult",
+    "SemanticCheckResult",
+    "SemanticRuleRegistry",
+    "SemanticValidationCache",
+    "SemanticValidationRule",
+    "SemanticValidationService",
+    "SemanticValidationSeverity",
     "SkillRegistryService",
+    "SyntaxCheckRule",
     "ToolRegistryService",
     "ValidationAttachment",
     "ValidationCommand",
@@ -73,4 +110,7 @@ __all__ = [
     "ValidationResult",
     "ValidationRunner",
     "ValidationSummary",
+    "create_default_registry",
+    "map_severity_to_mode",
+    "resolve_severity",
 ]

src/cleveragents/application/services/semantic_validation_rules.py

@@ -0,0 +1,610 @@
+"""Built-in semantic validation rules for Python projects.
+
+Provides pluggable rule implementations for syntax errors, missing
+imports, broken references, duplicate relative imports, API misuse,
+and missing symbols.  Each rule conforms to the ``SemanticValidationRule``
+protocol.
+
+Rules are designed to be lightweight, AST-based heuristics that
+avoid executing user code.
+
+Based on ``docs/specification.md`` and ADR-013 (Validation Abstraction).
+"""
+
+from __future__ import annotations
+
+import ast
+import builtins as _builtins_mod
+import sys
+from enum import StrEnum
+from typing import ClassVar
+
+from pydantic import BaseModel, ConfigDict, Field
+
+# ---------------------------------------------------------------------------
+# Shared severity enum + result model (canonical definitions)
+# ---------------------------------------------------------------------------
+
+
+class SemanticValidationSeverity(StrEnum):
+    """Severity level for semantic validation findings."""
+
+    INFO = "info"
+    WARN = "warn"
+    ERROR = "error"
+
+
+class SemanticCheckResult(BaseModel):
+    """Normalised result from a single semantic check.
+
+    Attributes:
+        passed: Whether the check passed (no issues found).
+        message: Human-readable description of the result.
+        data: Optional structured data with finding details.
+        severity: Severity level of the finding.
+    """
+
+    passed: bool = Field(..., description="Whether the check passed")
+    message: str = Field(..., description="Human-readable result description")
+    data: dict[str, object] | None = Field(
+        default=None, description="Optional structured finding data"
+    )
+    severity: SemanticValidationSeverity = Field(
+        default=SemanticValidationSeverity.INFO,
+        description="Severity of the finding",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Built-in rules
+# ---------------------------------------------------------------------------
+
+_BUILTINS_SET: frozenset[str] = frozenset(dir(_builtins_mod))
+
+# Use the authoritative stdlib module list (Python 3.10+) instead of a
+# hand-maintained subset so that internal modules like ``_thread``,
+# ``_abc``, etc. are recognised and do not trigger false positives.
+_STDLIB_MODULE_NAMES: frozenset[str] = frozenset(sys.stdlib_module_names)
+
+
+class SyntaxCheckRule:
+    """Check Python source for syntax errors via ``ast.parse``."""
+
+    @property
+    def name(self) -> str:
+        return "syntax_error"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            ast.parse(source, filename=filename)
+        except SyntaxError as exc:
+            lineno = exc.lineno or 0
+            return SemanticCheckResult(
+                passed=False,
+                message=f"Syntax error at {filename}:{lineno}: {exc.msg}",
+                data={"filename": filename, "line": lineno, "detail": exc.msg},
+                severity=SemanticValidationSeverity.ERROR,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No syntax errors in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )
+
+
+class MissingImportRule:
+    """Detect import statements that reference suspicious modules.
+
+    Uses ``ast`` to find import names that look like private
+    internal modules (leading underscore) that are not in the
+    standard library.  Uses ``sys.stdlib_module_names`` for an
+    authoritative list of stdlib top-level modules.
+    """
+
+    @property
+    def name(self) -> str:
+        return "missing_import"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError:
+            return SemanticCheckResult(
+                passed=True,
+                message="Skipped import check (syntax error)",
+                severity=SemanticValidationSeverity.INFO,
+            )
+
+        suspicious: list[dict[str, object]] = []
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    top = alias.name.split(".")[0]
+                    if top.startswith("_") and top not in _STDLIB_MODULE_NAMES:
+                        suspicious.append({"module": alias.name, "line": node.lineno})
+            elif (
+                isinstance(node, ast.ImportFrom)
+                and node.module
+                and node.module.startswith("_")
+            ):
+                top = node.module.split(".")[0]
+                if top not in _STDLIB_MODULE_NAMES:
+                    suspicious.append({"module": node.module, "line": node.lineno})
+
+        if suspicious:
+            return SemanticCheckResult(
+                passed=False,
+                message=(f"Found {len(suspicious)} suspicious import(s) in {filename}"),
+                data={"filename": filename, "imports": suspicious},
+                severity=SemanticValidationSeverity.WARN,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No missing imports detected in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Scope-aware name collection helpers
+# ---------------------------------------------------------------------------
+
+
+def _walk_and_collect(tree: ast.Module, *, deep: bool) -> set[str]:
+    """Shared name-collection walker.
+
+    When *deep* is ``False`` only direct children of *tree* are
+    inspected (module-scope bindings).  When *deep* is ``True``
+    the entire AST is walked to capture bindings at every scope
+    level, including ``ast.Name`` nodes in ``Store`` context
+    (e.g. comprehension targets).
+    """
+    names: set[str] = set()
+    iterator = ast.walk(tree) if deep else ast.iter_child_nodes(tree)
+    for node in iterator:
+        _collect_binding_names(node, names)
+        if deep and isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store):
+            names.add(node.id)
+    return names
+
+
+def _collect_defined_names(tree: ast.Module) -> set[str]:
+    """Collect names defined at module scope.
+
+    Handles function/class definitions, imports, assignments
+    (including annotated and augmented), for-loop targets,
+    with-statement variables, and exception handler names.
+
+    Uses :func:`_walk_and_collect` with ``deep=False`` to restrict
+    collection to top-level bindings only.
+    """
+    return _walk_and_collect(tree, deep=False)
+
+
+def _collect_binding_names(node: ast.AST, out: set[str]) -> None:
+    """Add names bound by *node* to *out*.
+
+    Recognised binding forms: ``FunctionDef``/``AsyncFunctionDef``
+    (name **and** parameters), ``ClassDef``, ``Import``,
+    ``ImportFrom``, ``Assign``, ``AnnAssign``, ``AugAssign``,
+    ``For``/``AsyncFor`` targets, ``With``/``AsyncWith`` optional
+    vars, and ``ExceptHandler`` names.
+    """
+    if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
+        out.add(node.name)
+        # Collect parameter names (positional, keyword-only, vararg, kwarg)
+        for arg in node.args.args + node.args.posonlyargs + node.args.kwonlyargs:
+            out.add(arg.arg)
+        if node.args.vararg:
+            out.add(node.args.vararg.arg)
+        if node.args.kwarg:
+            out.add(node.args.kwarg.arg)
+    elif isinstance(node, ast.ClassDef):
+        out.add(node.name)
+    elif isinstance(node, ast.Assign):
+        for target in node.targets:
+            _collect_target_names(target, out)
+    elif isinstance(node, (ast.AnnAssign, ast.AugAssign)) and node.target:
+        _collect_target_names(node.target, out)
+    elif isinstance(node, ast.Import):
+        for alias in node.names:
+            n = alias.asname if alias.asname else alias.name.split(".")[0]
+            out.add(n)
+    elif isinstance(node, ast.ImportFrom):
+        for alias in node.names:
+            n = alias.asname if alias.asname else alias.name
+            out.add(n)
+    elif isinstance(node, ast.For | ast.AsyncFor):
+        _collect_target_names(node.target, out)
+    elif isinstance(node, ast.With | ast.AsyncWith):
+        for item in node.items:
+            if item.optional_vars:
+                _collect_target_names(item.optional_vars, out)
+    elif isinstance(node, ast.ExceptHandler) and node.name:
+        out.add(node.name)
+
+
+def _collect_target_names(target: ast.AST, out: set[str]) -> None:
+    """Recursively extract names from an assignment target."""
+    if isinstance(target, ast.Name):
+        out.add(target.id)
+    elif isinstance(target, ast.Tuple | ast.List):
+        for elt in target.elts:
+            _collect_target_names(elt, out)
+    elif isinstance(target, ast.Starred):
+        _collect_target_names(target.value, out)
+
+
+def _collect_all_scope_names(tree: ast.Module) -> set[str]:
+    """Collect every name defined anywhere in the module.
+
+    Walks the entire AST to find all binding forms at every scope
+    level.  Used by ``BrokenReferenceRule`` so that names defined
+    inside functions, classes, loops, etc. are not falsely flagged
+    as broken references.
+
+    Uses :func:`_walk_and_collect` with ``deep=True`` to recurse
+    through all nested scopes.
+    """
+    return _walk_and_collect(tree, deep=True)
+
+
+class BrokenReferenceRule:
+    """Detect references to names not defined anywhere in the module.
+
+    Uses scope-aware collection that walks the entire AST so that
+    names defined inside functions, classes, loops, comprehensions,
+    and exception handlers are properly recognised.
+    """
+
+    @property
+    def name(self) -> str:
+        return "broken_reference"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError:
+            return SemanticCheckResult(
+                passed=True,
+                message="Skipped reference check (syntax error)",
+                severity=SemanticValidationSeverity.INFO,
+            )
+
+        defined = _collect_all_scope_names(tree)
+        defined.update(_BUILTINS_SET)
+
+        used: set[str] = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load):
+                used.add(node.id)
+
+        broken = used - defined
+        broken -= {
+            "__name__",
+            "__file__",
+            "__doc__",
+            "__all__",
+            "__annotations__",
+            "__spec__",
+            "__loader__",
+            "__package__",
+            "__builtins__",
+            "__cached__",
+            "__path__",
+        }
+
+        if broken:
+            limited = sorted(broken)[:10]
+            return SemanticCheckResult(
+                passed=False,
+                message=(
+                    f"Found {len(broken)} potentially broken reference(s) "
+                    f"in {filename}: {', '.join(limited)}"
+                ),
+                data={"filename": filename, "broken": limited},
+                severity=SemanticValidationSeverity.WARN,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No broken references in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )
+
+
+class DuplicateImportRule:
+    """Detect duplicate relative imports that hint at dependency issues.
+
+    Scans a single file for repeated relative import statements.
+    Duplicate relative imports within one module are a code smell
+    that *may* indicate copy-paste errors or structural problems.
+
+    .. note::
+
+       This rule does **not** perform cross-file cycle detection.
+       It only flags duplicate relative imports within a single
+       file.
+    """
+
+    @property
+    def name(self) -> str:
+        return "duplicate_import"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError:
+            return SemanticCheckResult(
+                passed=True,
+                message="Skipped duplicate import check (syntax error)",
+                severity=SemanticValidationSeverity.INFO,
+            )
+
+        relative_imports: list[str] = []
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.level and node.level > 0:
+                base = node.module or ""
+                names = ",".join(a.name for a in (node.names or []))
+                relative_imports.append(f"{base}:{names}")
+
+        if len(relative_imports) != len(set(relative_imports)):
+            duplicates = [
+                m for m in set(relative_imports) if relative_imports.count(m) > 1
+            ]
+            return SemanticCheckResult(
+                passed=False,
+                message=(
+                    f"Duplicate relative imports in {filename} "
+                    f"(potential cycle): {', '.join(duplicates)}"
+                ),
+                data={"filename": filename, "duplicates": duplicates},
+                severity=SemanticValidationSeverity.ERROR,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No duplicate relative imports in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )
+
+
+# Keep the old name as a public alias for backwards compatibility.
+DependencyCycleRule = DuplicateImportRule
+
+
+class APIMisuseRule:
+    """Detect common API misuse patterns in Python code.
+
+    Uses AST analysis to detect calls to dangerous functions such as
+    ``eval()``, ``exec()``, ``os.system()``, ``os.popen()``,
+    ``pickle.load()``, ``pickle.loads()``, ``marshal.loads()``,
+    ``compile()``, and ``__import__()``.  AST-based detection avoids
+    false positives from string literals and comments.
+    """
+
+    # Dangerous bare function names (resolved via ast.Name nodes).
+    _DANGEROUS_NAMES: ClassVar[frozenset[str]] = frozenset(
+        {"eval", "exec", "compile", "__import__"}
+    )
+
+    _DANGEROUS_NAME_DESCRIPTIONS: ClassVar[dict[str, str]] = {
+        "eval": "Use of eval() is discouraged",
+        "exec": "Use of exec() is discouraged",
+        "compile": "Use of compile() can execute arbitrary code",
+        "__import__": "Use importlib instead of __import__()",
+    }
+
+    # Dangerous qualified calls (module.func patterns).
+    _DANGEROUS_ATTRS: ClassVar[dict[str, dict[str, str]]] = {
+        "os": {
+            "system": "Use subprocess instead of os.system()",
+            "popen": "Use subprocess instead of os.popen()",
+        },
+        "subprocess": {
+            "call": "subprocess.call can execute arbitrary commands",
+            "run": "subprocess.run can execute arbitrary commands",
+            "Popen": "subprocess.Popen can execute arbitrary commands",
+        },
+        "pickle": {
+            "load": "pickle.load is unsafe with untrusted data",
+            "loads": "pickle.loads is unsafe with untrusted data",
+        },
+        "marshal": {
+            "loads": "marshal.loads is unsafe with untrusted data",
+        },
+    }
+
+    @property
+    def name(self) -> str:
+        return "api_misuse"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError:
+            return SemanticCheckResult(
+                passed=True,
+                message="Skipped API misuse check (syntax error)",
+                severity=SemanticValidationSeverity.INFO,
+            )
+
+        findings: list[dict[str, object]] = []
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.Call):
+                continue
+            func = node.func
+            # Bare name calls: eval(...), exec(...), compile(...), __import__(...)
+            if isinstance(func, ast.Name) and func.id in self._DANGEROUS_NAMES:
+                findings.append(
+                    {
+                        "line": node.lineno,
+                        "call": func.id,
+                        "description": self._DANGEROUS_NAME_DESCRIPTIONS[func.id],
+                    }
+                )
+            # Attribute calls: os.system(...), pickle.loads(...)
+            elif (
+                isinstance(func, ast.Attribute)
+                and isinstance(func.value, ast.Name)
+                and func.value.id in self._DANGEROUS_ATTRS
+            ):
+                attr_map = self._DANGEROUS_ATTRS[func.value.id]
+                if func.attr in attr_map:
+                    findings.append(
+                        {
+                            "line": node.lineno,
+                            "call": f"{func.value.id}.{func.attr}",
+                            "description": attr_map[func.attr],
+                        }
+                    )
+
+        if findings:
+            return SemanticCheckResult(
+                passed=False,
+                message=(f"Found {len(findings)} API misuse pattern(s) in {filename}"),
+                data={"filename": filename, "findings": findings},
+                severity=SemanticValidationSeverity.WARN,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No API misuse patterns in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )
+
+
+# ---------------------------------------------------------------------------
+# MissingSymbolRule helpers
+# ---------------------------------------------------------------------------
+
+
+def _collect_function_local_names(
+    node: ast.FunctionDef | ast.AsyncFunctionDef,
+) -> set[str]:
+    """Collect all names locally bound inside a function body.
+
+    Includes parameters, assignments, for-loop targets,
+    with-statement variables, exception handler names, nested
+    function/class definitions, and comprehension variables.
+    """
+    local_names: set[str] = set()
+
+    # Parameters
+    for arg in node.args.args:
+        local_names.add(arg.arg)
+    for arg in node.args.posonlyargs:
+        local_names.add(arg.arg)
+    for arg in node.args.kwonlyargs:
+        local_names.add(arg.arg)
+    if node.args.vararg:
+        local_names.add(node.args.vararg.arg)
+    if node.args.kwarg:
+        local_names.add(node.args.kwarg.arg)
+
+    # Walk the body for all binding forms
+    for child in ast.walk(node):
+        if child is node:
+            continue
+        # Nested function/class definitions bind their name
+        if isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
+            local_names.add(child.name)
+        # Store context names (assignments, comprehension targets)
+        elif isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store):
+            local_names.add(child.id)
+        # For-loop targets
+        elif isinstance(child, ast.For | ast.AsyncFor):
+            _collect_target_names(child.target, local_names)
+        # With-statement variables
+        elif isinstance(child, ast.With | ast.AsyncWith):
+            for item in child.items:
+                if item.optional_vars:
+                    _collect_target_names(item.optional_vars, local_names)
+        # Exception handler names
+        elif isinstance(child, ast.ExceptHandler) and child.name:
+            local_names.add(child.name)
+        # Import inside function
+        elif isinstance(child, ast.Import):
+            for alias in child.names:
+                n = alias.asname if alias.asname else alias.name.split(".")[0]
+                local_names.add(n)
+        elif isinstance(child, ast.ImportFrom):
+            for alias in child.names:
+                n = alias.asname if alias.asname else alias.name
+                local_names.add(n)
+
+    return local_names
+
+
+def _iter_functions(
+    tree: ast.AST,
+) -> list[ast.FunctionDef | ast.AsyncFunctionDef]:
+    """Yield all function/method definitions at any nesting depth."""
+    results: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
+            results.append(node)
+    return results
+
+
+class MissingSymbolRule:
+    """Detect undefined name usage in function and method bodies.
+
+    Inspects all functions and methods at any nesting depth
+    (including class methods and nested functions) for names that
+    are used but not defined locally, at module scope, or in
+    builtins.
+    """
+
+    @property
+    def name(self) -> str:
+        return "missing_symbol"
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        try:
+            tree = ast.parse(source, filename=filename)
+        except SyntaxError:
+            return SemanticCheckResult(
+                passed=True,
+                message="Skipped symbol check (syntax error)",
+                severity=SemanticValidationSeverity.INFO,
+            )
+
+        module_names = _collect_defined_names(tree)
+        functions = _iter_functions(tree)
+
+        missing: list[dict[str, object]] = []
+        for func_node in functions:
+            local_names = _collect_function_local_names(func_node)
+
+            for child in ast.walk(func_node):
+                if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load):
+                    nm = child.id
+                    if (
+                        nm not in local_names
+                        and nm not in module_names
+                        and nm not in _BUILTINS_SET
+                    ):
+                        missing.append(
+                            {
+                                "function": func_node.name,
+                                "symbol": nm,
+                                "line": child.lineno,
+                            }
+                        )
+
+        if missing:
+            limited = missing[:10]
+            return SemanticCheckResult(
+                passed=False,
+                message=f"Found {len(missing)} missing symbol(s) in {filename}",
+                data={"filename": filename, "missing": limited},
+                severity=SemanticValidationSeverity.INFO,
+            )
+        return SemanticCheckResult(
+            passed=True,
+            message=f"No missing symbols in {filename}",
+            severity=SemanticValidationSeverity.INFO,
+        )

src/cleveragents/application/services/semantic_validation_service.py

@@ -0,0 +1,445 @@
+"""Semantic validation service for CleverAgents.
+
+Provides semantic analysis checks for Python projects during the
+strategize/execute phases: syntax errors, missing imports, broken
+references, duplicate relative imports, API misuse, and missing
+symbols.
+
+Semantic checks are exposed as Validation tools attachable per
+resource via the Tool Registry.  Results integrate into the
+``ValidationPipeline`` as informational by default.
+
+Architecture:
+- ``SemanticValidationSeverity``: INFO / WARN / ERROR severity levels
+- ``SemanticCheckResult``: normalised result with passed/message/data/severity
+- ``SemanticValidationRule``: protocol for pluggable rule implementations
+- Built-in rules in ``semantic_validation_rules`` companion module
+- ``SemanticRuleRegistry``: register/lookup/list rules by name
+- ``SemanticValidationCache``: file-hash keyed LRU cache to skip unchanged files
+- ``SemanticValidationService``: orchestrator integrating with the pipeline
+- Config keys: ``validation.semantic.enabled``,
+  ``validation.semantic.python.enabled``,
+  ``validation.semantic.severity_mapping``
+
+Based on ``docs/specification.md`` and ADR-013 (Validation Abstraction).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import threading
+from collections import OrderedDict
+from collections.abc import Mapping
+from typing import Protocol, TypedDict, runtime_checkable
+
+import structlog
+
+from cleveragents.application.services.semantic_validation_rules import (
+    APIMisuseRule,
+    BrokenReferenceRule,
+    DuplicateImportRule,
+    MissingImportRule,
+    MissingSymbolRule,
+    SemanticCheckResult,
+    SemanticValidationSeverity,
+    SyntaxCheckRule,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+logger = structlog.get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Typed dicts for public return types
+# ---------------------------------------------------------------------------
+
+
+class PipelineResultDict(TypedDict):
+    """Typed dict for pipeline-compatible result dictionaries."""
+
+    passed: bool
+    message: str
+    data: dict[str, object] | None
+    severity: str
+    mode: str
+
+
+class NormalisedOutputDict(TypedDict):
+    """Typed dict for normalised output dictionaries."""
+
+    passed: bool
+    message: str
+    data: dict[str, object] | None
+
+
+# ---------------------------------------------------------------------------
+# Config key constants
+# ---------------------------------------------------------------------------
+
+CONFIG_KEY_ENABLED = "validation.semantic.enabled"
+CONFIG_KEY_PYTHON_ENABLED = "validation.semantic.python.enabled"
+CONFIG_KEY_SEVERITY_MAPPING = "validation.semantic.severity_mapping"
+
+_DEFAULT_SEVERITY_MAPPING: dict[str, str] = {
+    "syntax_error": "error",
+    "missing_import": "warn",
+    "broken_reference": "warn",
+    "duplicate_import": "error",
+    "api_misuse": "warn",
+    "missing_symbol": "info",
+}
+
+DEFAULT_CONFIG: dict[str, object] = {
+    CONFIG_KEY_ENABLED: True,
+    CONFIG_KEY_PYTHON_ENABLED: True,
+    CONFIG_KEY_SEVERITY_MAPPING: _DEFAULT_SEVERITY_MAPPING,
+}
+
+
+# ---------------------------------------------------------------------------
+# Rule protocol
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class SemanticValidationRule(Protocol):
+    """Protocol for pluggable semantic validation rules.
+
+    Each rule has a unique ``name`` and a ``check`` method that
+    inspects source code and returns a normalised result.
+    """
+
+    @property
+    def name(self) -> str:
+        """Unique name of this validation rule."""
+        ...  # pragma: no cover
+
+    def check(self, source: str, filename: str) -> SemanticCheckResult:
+        """Run the semantic check on source code.
+
+        Args:
+            source: Python source code to check.
+            filename: Filename (for error messages).
+
+        Returns:
+            SemanticCheckResult with pass/fail and details.
+        """
+        ...  # pragma: no cover
+
+
+# ---------------------------------------------------------------------------
+# Rule registry
+# ---------------------------------------------------------------------------
+
+
+class SemanticRuleRegistry:
+    """Registry for semantic validation rules.
+
+    Supports registration, lookup by name, and listing of all
+    registered rules.
+    """
+
+    def __init__(self) -> None:
+        self._rules: dict[str, SemanticValidationRule] = {}
+
+    def register(self, rule: SemanticValidationRule) -> None:
+        """Register a rule.  Overwrites if name already exists."""
+        self._rules[rule.name] = rule
+        logger.debug("Registered semantic rule: %s", rule.name)
+
+    def get(self, name: str) -> SemanticValidationRule | None:
+        """Look up a rule by name."""
+        return self._rules.get(name)
+
+    def list_rules(self) -> list[str]:
+        """Return sorted list of registered rule names."""
+        return sorted(self._rules.keys())
+
+    def all_rules(self) -> list[SemanticValidationRule]:
+        """Return all registered rules sorted by name."""
+        return [self._rules[n] for n in sorted(self._rules.keys())]
+
+    def remove(self, name: str) -> bool:
+        """Remove a rule by name.  Returns True if found."""
+        if name in self._rules:
+            del self._rules[name]
+            return True
+        return False
+
+    def __len__(self) -> int:
+        return len(self._rules)
+
+
+def create_default_registry() -> SemanticRuleRegistry:
+    """Create a registry pre-loaded with all built-in Python rules."""
+    registry = SemanticRuleRegistry()
+    registry.register(SyntaxCheckRule())
+    registry.register(MissingImportRule())
+    registry.register(BrokenReferenceRule())
+    registry.register(DuplicateImportRule())
+    registry.register(APIMisuseRule())
+    registry.register(MissingSymbolRule())
+    return registry
+
+
+# ---------------------------------------------------------------------------
+# Cache
+# ---------------------------------------------------------------------------
+
+_DEFAULT_CACHE_MAX_SIZE = 4096
+
+
+class SemanticValidationCache:
+    """File-hash keyed LRU cache for semantic check results.
+
+    Avoids re-running checks on unchanged files by caching results
+    keyed by ``(rule_name, file_hash)``.  Uses a bounded
+    ``OrderedDict`` with LRU eviction to prevent unbounded memory
+    growth.  All public methods are thread-safe.
+
+    Args:
+        max_size: Maximum number of entries before LRU eviction
+            kicks in.  Defaults to ``4096``.
+    """
+
+    def __init__(self, max_size: int = _DEFAULT_CACHE_MAX_SIZE) -> None:
+        if max_size < 1:
+            raise ValueError("max_size must be >= 1")
+        self._max_size = max_size
+        self._cache: OrderedDict[str, SemanticCheckResult] = OrderedDict()
+        self._lock = threading.Lock()
+
+    @staticmethod
+    def compute_hash(content: str) -> str:
+        """Compute SHA-256 hex digest of source content."""
+        return hashlib.sha256(content.encode("utf-8")).hexdigest()
+
+    def _key(self, rule_name: str, file_hash: str) -> str:
+        return f"{rule_name}:{file_hash}"
+
+    def get(self, rule_name: str, file_hash: str) -> SemanticCheckResult | None:
+        """Retrieve a cached result, or None if not cached."""
+        key = self._key(rule_name, file_hash)
+        with self._lock:
+            if key in self._cache:
+                self._cache.move_to_end(key)
+                return self._cache[key]
+            return None
+
+    def put(self, rule_name: str, file_hash: str, result: SemanticCheckResult) -> None:
+        """Store a result in the cache, evicting LRU entries if needed."""
+        key = self._key(rule_name, file_hash)
+        with self._lock:
+            if key in self._cache:
+                self._cache.move_to_end(key)
+            self._cache[key] = result
+            while len(self._cache) > self._max_size:
+                self._cache.popitem(last=False)
+
+    def invalidate(self, rule_name: str, file_hash: str) -> None:
+        """Remove a cached entry."""
+        key = self._key(rule_name, file_hash)
+        with self._lock:
+            self._cache.pop(key, None)
+
+    def clear(self) -> None:
+        """Clear all cached results."""
+        with self._lock:
+            self._cache.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._cache)
+
+
+# ---------------------------------------------------------------------------
+# Severity mapping
+# ---------------------------------------------------------------------------
+
+
+def map_severity_to_mode(
+    severity: SemanticValidationSeverity,
+) -> ValidationMode:
+    """Map a semantic severity to a ValidationPipeline mode.
+
+    - ERROR -> REQUIRED (failure blocks)
+    - WARN / INFO -> INFORMATIONAL (failure reported only)
+    """
+    if severity == SemanticValidationSeverity.ERROR:
+        return ValidationMode.REQUIRED
+    return ValidationMode.INFORMATIONAL
+
+
+def resolve_severity(
+    rule_name: str,
+    severity_mapping: dict[str, str] | None = None,
+) -> SemanticValidationSeverity:
+    """Resolve the configured severity for a rule.
+
+    Falls back to the default mapping if no override is provided.
+    """
+    mapping = severity_mapping or _DEFAULT_SEVERITY_MAPPING
+    raw = mapping.get(rule_name, "info")
+    try:
+        return SemanticValidationSeverity(raw)
+    except ValueError:
+        return SemanticValidationSeverity.INFO
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+
+_PYTHON_EXTENSIONS: frozenset[str] = frozenset({".py", ".pyi"})
+
+
+class SemanticValidationService:
+    """Orchestrator for semantic validation checks.
+
+    Integrates with the ``ValidationPipeline`` by producing
+    normalised ``SemanticCheckResult`` objects and converting them
+    to pipeline-compatible dicts.  Uses a rule registry and cache
+    for extensibility and efficiency.
+
+    Args:
+        registry: Rule registry to use (default: built-in rules).
+        cache: Cache instance (default: new empty cache).
+        config: Configuration overrides (default: DEFAULT_CONFIG).
+    """
+
+    def __init__(
+        self,
+        registry: SemanticRuleRegistry | None = None,
+        cache: SemanticValidationCache | None = None,
+        config: Mapping[str, object] | None = None,
+    ) -> None:
+        self._registry = registry or create_default_registry()
+        self._cache = cache or SemanticValidationCache()
+        self._config = {**DEFAULT_CONFIG, **(config or {})}
+
+    @property
+    def registry(self) -> SemanticRuleRegistry:
+        """The rule registry used by this service."""
+        return self._registry
+
+    @property
+    def cache(self) -> SemanticValidationCache:
+        """The result cache used by this service."""
+        return self._cache
+
+    @property
+    def enabled(self) -> bool:
+        """Whether semantic validation is globally enabled."""
+        return bool(self._config.get(CONFIG_KEY_ENABLED, True))
+
+    @property
+    def python_enabled(self) -> bool:
+        """Whether Python-specific semantic checks are enabled."""
+        return bool(self._config.get(CONFIG_KEY_PYTHON_ENABLED, True))
+
+    @staticmethod
+    def _is_python_file(filename: str) -> bool:
+        """Return True if *filename* has a Python extension."""
+        return filename.endswith((".py", ".pyi"))
+
+    def check_file(
+        self,
+        source: str,
+        filename: str,
+        rule_names: list[str] | None = None,
+    ) -> list[SemanticCheckResult]:
+        """Run semantic checks on a single file.
+
+        Args:
+            source: Python source code.
+            filename: Filename for error messages.
+            rule_names: Optional list of specific rules to run.
+                If None, all registered rules are run.
+
+        Returns:
+            List of SemanticCheckResult objects.
+        """
+        if not self.enabled:
+            return []
+
+        # All built-in rules are Python-specific.  Skip non-Python
+        # files entirely when Python checks are disabled, and also
+        # skip non-Python files unconditionally since the rules use
+        # ``ast.parse`` which only handles Python.
+        if not self._is_python_file(filename):
+            return []
+
+        if not self.python_enabled:
+            return []
+
+        file_hash = self._cache.compute_hash(source)
+        rules = (
+            [r for n in rule_names if (r := self._registry.get(n)) is not None]
+            if rule_names
+            else self._registry.all_rules()
+        )
+
+        results: list[SemanticCheckResult] = []
+        raw_mapping = self._config.get(CONFIG_KEY_SEVERITY_MAPPING)
+        severity_mapping: dict[str, str] | None = None
+        if isinstance(raw_mapping, dict):
+            severity_mapping = {str(k): str(v) for k, v in raw_mapping.items()}
+
+        for rule in rules:
+            cached = self._cache.get(rule.name, file_hash)
+            if cached is not None:
+                results.append(cached)
+                continue
+
+            result = rule.check(source, filename)
+            # Apply configured severity override
+            configured = resolve_severity(rule.name, severity_mapping)
+            if result.severity != configured:
+                result = SemanticCheckResult(
+                    passed=result.passed,
+                    message=result.message,
+                    data=result.data,
+                    severity=configured,
+                )
+            self._cache.put(rule.name, file_hash, result)
+            results.append(result)
+
+        return results
+
+    def as_pipeline_results(
+        self,
+        source: str,
+        filename: str,
+        rule_names: list[str] | None = None,
+    ) -> list[PipelineResultDict]:
+        """Run checks and return pipeline-compatible result dicts.
+
+        Each dict has ``passed``, ``message``, ``data``, ``severity``,
+        and ``mode`` keys for integration with ``ValidationPipeline``.
+        """
+        checks = self.check_file(source, filename, rule_names)
+        pipeline_results: list[PipelineResultDict] = []
+        for result in checks:
+            mode = map_severity_to_mode(result.severity)
+            pipeline_results.append(
+                PipelineResultDict(
+                    passed=result.passed,
+                    message=result.message,
+                    data=result.data,
+                    severity=result.severity.value,
+                    mode=mode.value,
+                )
+            )
+        return pipeline_results
+
+    def normalise_output(self, result: SemanticCheckResult) -> NormalisedOutputDict:
+        """Normalise a check result to the standard output schema.
+
+        Returns a dict with ``passed``, ``message``, and ``data`` keys.
+        """
+        return NormalisedOutputDict(
+            passed=result.passed,
+            message=result.message,
+            data=result.data,
+        )

vulture_whitelist.py

@@ -351,3 +351,46 @@ _decision_seq  # noqa: B018, F821
 mock_providers  # noqa: B018, F821
 validate_provider_availability  # noqa: B018, F821
 resolve_provider_by_name  # noqa: B018, F821
+
+# Semantic validation service — public API (M6.semantic)
+SemanticValidationSeverity  # noqa: B018, F821
+SemanticCheckResult  # noqa: B018, F821
+SemanticValidationRule  # noqa: B018, F821
+SemanticRuleRegistry  # noqa: B018, F821
+SemanticValidationCache  # noqa: B018, F821
+SemanticValidationService  # noqa: B018, F821
+SyntaxCheckRule  # noqa: B018, F821
+MissingImportRule  # noqa: B018, F821
+BrokenReferenceRule  # noqa: B018, F821
+DependencyCycleRule  # noqa: B018, F821
+DuplicateImportRule  # noqa: B018, F821
+APIMisuseRule  # noqa: B018, F821
+MissingSymbolRule  # noqa: B018, F821
+create_default_registry  # noqa: B018, F821
+map_severity_to_mode  # noqa: B018, F821
+resolve_severity  # noqa: B018, F821
+CONFIG_KEY_ENABLED  # noqa: B018, F821
+CONFIG_KEY_PYTHON_ENABLED  # noqa: B018, F821
+CONFIG_KEY_SEVERITY_MAPPING  # noqa: B018, F821
+DEFAULT_CONFIG  # noqa: B018, F821
+normalise_output  # noqa: B018, F821
+as_pipeline_results  # noqa: B018, F821
+check_file  # noqa: B018, F821
+python_enabled  # noqa: B018, F821
+list_rules  # noqa: B018, F821
+all_rules  # noqa: B018, F821
+compute_hash  # noqa: B018, F821
+invalidate  # noqa: B018, F821
+_BUILTINS_SET  # noqa: B018, F821
+_STDLIB_MODULE_NAMES  # noqa: B018, F821
+_DANGEROUS_NAMES  # noqa: B018, F821
+_DANGEROUS_NAME_DESCRIPTIONS  # noqa: B018, F821
+_DANGEROUS_ATTRS  # noqa: B018, F821
+_PYTHON_EXTENSIONS  # noqa: B018, F821
+_DEFAULT_CACHE_MAX_SIZE  # noqa: B018, F821
+_collect_binding_names  # noqa: B018, F821
+_collect_target_names  # noqa: B018, F821
+_collect_all_scope_names  # noqa: B018, F821
+_collect_function_local_names  # noqa: B018, F821
+_iter_functions  # noqa: B018, F821
+_is_python_file  # noqa: B018, F821