feat: implement conditional rule discovery engine

CHANGELOG.md

@@ -28,6 +28,7 @@ All notable changes to this project are documented in this file.
 - Added `/keyword-mode` aliases (`status|detect|apply`) to inspect and apply keyword-triggered execution modes.
 - Added keyword mode controls for global enable/disable and per-keyword toggles (`disable-keyword` / `enable-keyword`) with persisted config state.
 - Added `instructions/conditional_rules_schema.md` defining rule frontmatter schema, discovery precedence, conflict resolution, and validation requirements for Epic 9 Task 9.1.
+- Added `scripts/rules_engine.py` implementing frontmatter parsing, layered rule discovery, path-based matching, deterministic precedence sorting, and duplicate-id conflict reporting.
 
 ### Changes
 - Documented extension evaluation outcomes and when each tool is the better fit.
@@ -56,6 +57,7 @@ All notable changes to this project are documented in this file.
 - Expanded keyword mode docs with examples and anti-pattern guidance, plus stronger selftest/install smoke coverage for keyword toggle behavior.
 - Added `/keyword-mode doctor --json` diagnostics and integrated keyword subsystem health into unified `/doctor` checks.
 - Expanded keyword mode verification for false-positive resistance (partial words and code-literal contexts) and opt-out/toggle smoke scenarios.
+- Expanded selftest coverage for conditional rule discovery and effective-stack resolution behavior.
 
 ## v0.2.0 - 2026-02-12
 

IMPLEMENTATION_ROADMAP.md

@@ -45,7 +45,7 @@ Use this map to avoid overlapping implementations.
 | E6 | Session Intelligence and Resume Tooling | paused | Medium | E2 | TBD | Resume when core orchestration stabilizes |
 | E7 | Tmux Visual Multi-Agent Mode | postponed | Low | E2 | TBD | Optional power-user feature |
 | E8 | Keyword-Triggered Execution Modes | done | High | E1, E4 | bd-302, bd-2fb, bd-2zq, bd-3dp | Fast power-mode activation from prompt text |
-| E9 | Conditional Rules Injector | in_progress | High | E1 | bd-1q8 | Enforce project conventions with scoped rules |
+| E9 | Conditional Rules Injector | in_progress | High | E1 | bd-1q8, bd-3rj | Enforce project conventions with scoped rules |
 | E10 | Auto Slash Command Detector | paused | Medium | E1, E8 | TBD | Resume only if intent precision stays high in prototypes |
 | E11 | Context-Window Resilience Toolkit | planned | High | E4 | TBD | Improve long-session stability and recovery |
 | E12 | Provider/Model Fallback Visibility | planned | Medium | E5 | TBD | Explain why model routing decisions happen |
@@ -417,10 +417,11 @@ Every command-oriented epic must ship all of the following:
   - [x] Subtask 9.1.2: Define project/user rule search paths
   - [x] Subtask 9.1.3: Define rule conflict resolution strategy
   - [x] Notes: Added `instructions/conditional_rules_schema.md` with deterministic discovery, matching, precedence, conflict handling, and validation requirements.
-- [ ] Task 9.2: Implement rule discovery and matching engine
-  - [ ] Subtask 9.2.1: Discover markdown rule files recursively
-  - [ ] Subtask 9.2.2: Match rules by file path and operation context
-  - [ ] Subtask 9.2.3: Inject effective rule set into execution context
+- [x] Task 9.2: Implement rule discovery and matching engine
+  - [x] Subtask 9.2.1: Discover markdown rule files recursively
+  - [x] Subtask 9.2.2: Match rules by file path and operation context
+  - [x] Subtask 9.2.3: Inject effective rule set into execution context
+  - [x] Notes: Added `scripts/rules_engine.py` with frontmatter parsing, layered discovery, deterministic precedence sorting, duplicate-id conflict reporting, and effective rule stack resolution helpers.
 - [ ] Task 9.3: Operations and diagnostics
   - [ ] Subtask 9.3.1: Add `/rules status` and `/rules explain <path>` commands
   - [ ] Subtask 9.3.2: Add per-rule disable list in config

scripts/rules_engine.py

@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+from fnmatch import fnmatch
+from pathlib import Path
+from typing import Any
+
+
+def _parse_scalar(value: str) -> Any:
+    lowered = value.lower()
+    if lowered in ("true", "false"):
+        return lowered == "true"
+    try:
+        return int(value)
+    except ValueError:
+        return value
+
+
+def parse_frontmatter(markdown: str) -> tuple[dict[str, Any], str]:
+    lines = markdown.splitlines()
+    if not lines or lines[0].strip() != "---":
+        return {}, markdown
+
+    frontmatter_lines: list[str] = []
+    end_index = -1
+    for idx in range(1, len(lines)):
+        if lines[idx].strip() == "---":
+            end_index = idx
+            break
+        frontmatter_lines.append(lines[idx])
+
+    if end_index == -1:
+        return {}, markdown
+
+    payload: dict[str, Any] = {}
+    current_key: str | None = None
+    for raw in frontmatter_lines:
+        line = raw.rstrip()
+        if not line.strip():
+            continue
+        if line.lstrip().startswith("- ") and current_key:
+            item = line.lstrip()[2:].strip()
+            payload.setdefault(current_key, [])
+            if isinstance(payload[current_key], list):
+                payload[current_key].append(_parse_scalar(item))
+            continue
+        if ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        current_key = key.strip()
+        value = value.strip()
+        if not value:
+            payload[current_key] = []
+            continue
+        payload[current_key] = _parse_scalar(value)
+
+    body = "\n".join(lines[end_index + 1 :]).lstrip("\n")
+    return payload, body
+
+
+def normalize_rule_id(raw: str | None, fallback_stem: str) -> str:
+    value = (raw or "").strip().lower()
+    if value:
+        return value
+    return fallback_stem.strip().lower().replace(" ", "-")
+
+
+def validate_rule(rule: dict[str, Any]) -> list[str]:
+    problems: list[str] = []
+    if (
+        not isinstance(rule.get("description"), str)
+        or not str(rule["description"]).strip()
+    ):
+        problems.append("description is required")
+    priority = rule.get("priority")
+    if not isinstance(priority, int) or priority < 0 or priority > 100:
+        problems.append("priority must be an integer between 0 and 100")
+    if "alwaysApply" in rule and not isinstance(rule.get("alwaysApply"), bool):
+        problems.append("alwaysApply must be a boolean")
+    if "globs" in rule:
+        globs = rule.get("globs")
+        if not isinstance(globs, list) or not all(
+            isinstance(item, str) for item in globs
+        ):
+            problems.append("globs must be a list of strings")
+    return problems
+
+
+def discover_rules(
+    project_root: Path, home: Path | None = None
+) -> list[dict[str, Any]]:
+    base_home = home or Path.home()
+    user_root = base_home / ".config" / "opencode" / "rules"
+    project_root_rules = project_root / ".opencode" / "rules"
+
+    roots = [("user", user_root), ("project", project_root_rules)]
+    discovered: list[dict[str, Any]] = []
+    for scope, root in roots:
+        if not root.exists():
+            continue
+        for path in sorted(root.rglob("*.md")):
+            text = path.read_text(encoding="utf-8")
+            frontmatter, body = parse_frontmatter(text)
+            rule = dict(frontmatter)
+            rule["id"] = normalize_rule_id(
+                str(frontmatter.get("id"))
+                if frontmatter.get("id") is not None
+                else None,
+                path.stem,
+            )
+            rule["scope"] = scope
+            rule["path"] = str(path)
+            rule["body"] = body
+            rule["problems"] = validate_rule(rule)
+            discovered.append(rule)
+    return discovered
+
+
+def rule_applies(rule: dict[str, Any], target_path: str) -> bool:
+    if rule.get("problems"):
+        return False
+    if rule.get("alwaysApply") is True:
+        return True
+    globs = rule.get("globs")
+    if isinstance(globs, list):
+        return any(
+            isinstance(pattern, str) and fnmatch(target_path, pattern)
+            for pattern in globs
+        )
+    return False
+
+
+def resolve_effective_rules(
+    rules: list[dict[str, Any]], target_path: str
+) -> dict[str, Any]:
+    applicable = [rule for rule in rules if rule_applies(rule, target_path)]
+    sorted_rules = sorted(
+        applicable,
+        key=lambda rule: (
+            -int(rule.get("priority", 0)),
+            0 if rule.get("scope") == "project" else 1,
+            str(rule.get("id", "")),
+        ),
+    )
+
+    effective: list[dict[str, Any]] = []
+    conflicts: list[dict[str, str]] = []
+    seen_ids: set[str] = set()
+    for rule in sorted_rules:
+        rule_id = str(rule.get("id", ""))
+        if rule_id in seen_ids:
+            conflicts.append(
+                {
+                    "id": rule_id,
+                    "path": str(rule.get("path", "")),
+                    "reason": "duplicate_rule_id_overridden",
+                }
+            )
+            continue
+        seen_ids.add(rule_id)
+        effective.append(rule)
+
+    return {
+        "target_path": target_path,
+        "effective_rules": effective,
+        "conflicts": conflicts,
+        "discovered_count": len(rules),
+        "applicable_count": len(applicable),
+    }

scripts/selftest.py

@@ -24,6 +24,11 @@
     validate_schema,
 )
 from keyword_mode_schema import resolve_prompt_modes  # type: ignore
+from rules_engine import (  # type: ignore
+    discover_rules,
+    parse_frontmatter,
+    resolve_effective_rules,
+)
 
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -1615,6 +1620,78 @@ def run_bg(*args: str) -> subprocess.CompletedProcess[str]:
             "keyword-mode doctor should report PASS",
         )
 
+        frontmatter, body = parse_frontmatter(
+            """---\ndescription: Rule example\npriority: 60\nglobs:\n  - scripts/*.py\n---\nUse safe edits.\n"""
+        )
+        expect(
+            frontmatter.get("priority") == 60,
+            "rules frontmatter should parse integer fields",
+        )
+        expect(
+            frontmatter.get("globs") == ["scripts/*.py"],
+            "rules frontmatter should parse list fields",
+        )
+        expect(
+            body.strip() == "Use safe edits.",
+            "rules frontmatter parser should return markdown body",
+        )
+
+        project_tmp = tmp / "rules-project"
+        project_rules_dir = project_tmp / ".opencode" / "rules"
+        user_rules_dir = home / ".config" / "opencode" / "rules"
+        project_rules_dir.mkdir(parents=True, exist_ok=True)
+        user_rules_dir.mkdir(parents=True, exist_ok=True)
+
+        (project_rules_dir / "python-safe.md").write_text(
+            """---\nid: style-python\ndescription: Python strict style\npriority: 80\nglobs:\n  - scripts/*.py\n---\nPrefer explicit typing for new functions.\n""",
+            encoding="utf-8",
+        )
+        (user_rules_dir / "python-safe.md").write_text(
+            """---\nid: style-python\ndescription: User python defaults\npriority: 70\nglobs:\n  - scripts/*.py\n---\nPrefer concise comments.\n""",
+            encoding="utf-8",
+        )
+        (user_rules_dir / "docs-rule.md").write_text(
+            """---\ndescription: Docs guidance\npriority: 50\nglobs:\n  - README.md\n---\nKeep examples concise.\n""",
+            encoding="utf-8",
+        )
+
+        discovered_rules = discover_rules(project_tmp, home=home)
+        expect(
+            len(discovered_rules) == 3,
+            "rules discovery should include user and project markdown rules",
+        )
+        resolved_rules = resolve_effective_rules(
+            discovered_rules, "scripts/selftest.py"
+        )
+        effective_rule_ids = [
+            str(rule.get("id")) for rule in resolved_rules.get("effective_rules", [])
+        ]
+        expect(
+            "style-python" in effective_rule_ids,
+            "rules resolution should include matching python rule",
+        )
+        winning_python_rule = next(
+            rule
+            for rule in resolved_rules.get("effective_rules", [])
+            if rule.get("id") == "style-python"
+        )
+        expect(
+            winning_python_rule.get("scope") == "project",
+            "project-scoped rule should win when duplicate ids conflict",
+        )
+        expect(
+            len(resolved_rules.get("conflicts", [])) == 1,
+            "rules resolution should report duplicate-id conflicts",
+        )
+        readme_rules = resolve_effective_rules(discovered_rules, "README.md")
+        expect(
+            any(
+                str(rule.get("id")) == "docs-rule"
+                for rule in readme_rules.get("effective_rules", [])
+            ),
+            "rules resolution should match README-targeted docs rule",
+        )
+
         wizard_state_path = (
             home / ".config" / "opencode" / "my_opencode-install-state.json"
         )