feat: integrate netx ops data access and unify prompt/workspace runtime

Migrate prompt templates into runtime workspaces, add network_ops netx toolchain (latest batch injection, raw fields, SQL query), and align docs/tests/admin/runtime wiring so ops analysis can query netx detail data through stable role-scoped paths.

Co-authored-by: Cursor <cursoragent@cursor.com>

runtime/prompt_templates/__init__.py

@@ -0,0 +1,17 @@
+from .loader import (
+    PromptDoc,
+    load_prompt_doc,
+    load_runtime_prompt_doc,
+    render_prompt,
+    render_prompt_for_lang,
+    render_runtime_prompt,
+)
+
+__all__ = [
+    "PromptDoc",
+    "load_prompt_doc",
+    "load_runtime_prompt_doc",
+    "render_prompt",
+    "render_prompt_for_lang",
+    "render_runtime_prompt",
+]

runtime/prompt_templates/frontmatter.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+try:
+    import yaml
+except ImportError:  # pragma: no cover - requirements.txt includes PyYAML
+    yaml = None  # type: ignore[assignment]
+
+
+def split_markdown_frontmatter(raw: str) -> tuple[str, str]:
+    """Split leading YAML frontmatter from markdown body (oclaw-style `---` fences)."""
+    txt = str(raw or "")
+    if not txt.startswith("---\n"):
+        return "", txt
+    end = txt.find("\n---\n", 4)
+    if end < 0:
+        return "", txt
+    return txt[4:end], txt[end + 5 :]
+
+
+def _legacy_line_parse(frontmatter_text: str) -> dict[str, Any]:
+    """Best-effort key: value lines (pre-YAML migration compatibility)."""
+    out: dict[str, Any] = {}
+    for ln in str(frontmatter_text or "").splitlines():
+        line = ln.strip()
+        if not line or line.startswith("#") or ":" not in line:
+            continue
+        k, v = line.split(":", 1)
+        key = str(k).strip()
+        val = str(v).strip()
+        if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
+            val = val[1:-1]
+        out[key] = val
+    meta_raw = str(out.get("metadata") or "").strip()
+    if meta_raw:
+        try:
+            out["metadata"] = json.loads(meta_raw)
+        except Exception:
+            out["metadata"] = {}
+    else:
+        out["metadata"] = {}
+    return out
+
+
+def frontmatter_strict_yaml() -> bool:
+    return str(os.getenv("AIA_PROMPT_FRONTMATTER_STRICT", "")).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def parse_frontmatter_dict(frontmatter_text: str, *, source: str = "prompt") -> dict[str, Any]:
+    """
+    Parse YAML frontmatter. On YAML failure, falls back to legacy line parser unless
+    AIA_PROMPT_FRONTMATTER_STRICT is truthy (then raises).
+    """
+    text = str(frontmatter_text or "").strip()
+    if not text:
+        return {}
+    strict = frontmatter_strict_yaml()
+    if yaml is None:
+        if strict:
+            raise RuntimeError("PyYAML is required when AIA_PROMPT_FRONTMATTER_STRICT is enabled")
+        return _legacy_line_parse(text)
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        if strict:
+            raise ValueError(f"invalid YAML frontmatter ({source}): {exc}") from exc
+        return _legacy_line_parse(text)
+    if data is None:
+        return {}
+    if not isinstance(data, dict):
+        if strict:
+            raise ValueError(f"frontmatter YAML root must be a mapping ({source})")
+        return _legacy_line_parse(text)
+    return dict(data)
+
+
+def parse_markdown_document(raw: str, *, source: str = "markdown") -> tuple[dict[str, Any], str]:
+    """Split file and parse frontmatter to a dict; body is stripped markdown."""
+    fm_text, body = split_markdown_frontmatter(raw)
+    meta = parse_frontmatter_dict(fm_text, source=source) if fm_text.strip() else {}
+    return meta, str(body or "").strip()
+
+
+__all__ = [
+    "frontmatter_strict_yaml",
+    "parse_frontmatter_dict",
+    "parse_markdown_document",
+    "split_markdown_frontmatter",
+]

runtime/prompt_templates/loader.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from oclaw.platform.config.paths import PROJECT_ROOT
+from oclaw.runtime.prompt_templates.frontmatter import parse_markdown_document
+
+_VAR_RE = re.compile(r"\{\{\s*([a-zA-Z0-9_]+)\s*\}\}")
+
+
+@dataclass(frozen=True)
+class PromptDoc:
+    frontmatter: dict[str, Any]
+    body: str
+
+
+def _system_prompt_root() -> Path:
+    """Builtin Markdown templates: ``runtime/workspaces/_system`` (replaces former ``prompts/``)."""
+    return (PROJECT_ROOT / "runtime" / "workspaces" / "_system").resolve()
+
+
+def _prompts_root() -> Path:
+    return _system_prompt_root()
+
+
+def _runtime_prompts_root() -> Path:
+    return _system_prompt_root()
+
+
+@lru_cache(maxsize=512)
+def load_prompt_doc(prompt_path: str) -> PromptDoc:
+    p = (_prompts_root() / str(prompt_path)).resolve()
+    if not p.exists():
+        raise FileNotFoundError(f"prompt not found: {prompt_path}")
+    raw = p.read_text(encoding="utf-8")
+    fm, body = parse_markdown_document(raw, source=f"prompt:{prompt_path}")
+    return PromptDoc(frontmatter=fm, body=body)
+
+
+@lru_cache(maxsize=512)
+def load_runtime_prompt_doc(prompt_path: str) -> PromptDoc:
+    p = (_runtime_prompts_root() / str(prompt_path)).resolve()
+    if not p.exists():
+        raise FileNotFoundError(f"runtime prompt not found: {prompt_path}")
+    raw = p.read_text(encoding="utf-8")
+    fm, body = parse_markdown_document(raw, source=f"runtime_prompt:{prompt_path}")
+    return PromptDoc(frontmatter=fm, body=body)
+
+
+def render_prompt_for_lang(stem: str, lang: str, *, variables: dict[str, Any] | None = None, strict: bool = True) -> str:
+    suf = "en" if (lang or "zh").strip().lower().startswith("en") else "zh"
+    return render_prompt(f"{stem}.{suf}.md", variables=variables, strict=strict)
+
+
+def render_prompt(prompt_path: str, *, variables: dict[str, Any] | None = None, strict: bool = True) -> str:
+    doc = load_prompt_doc(prompt_path)
+    vars_in = dict(variables or {})
+    missing: set[str] = set()
+
+    def _repl(m: re.Match[str]) -> str:
+        key = str(m.group(1) or "")
+        if key in vars_in:
+            return str(vars_in.get(key) or "")
+        missing.add(key)
+        return ""
+
+    out = _VAR_RE.sub(_repl, doc.body)
+    if strict and missing:
+        raise ValueError(f"missing prompt variables for {prompt_path}: {', '.join(sorted(missing))}")
+    return out.strip()
+
+
+def render_runtime_prompt(prompt_path: str, *, variables: dict[str, Any] | None = None, strict: bool = True) -> str:
+    doc = load_runtime_prompt_doc(prompt_path)
+    vars_in = dict(variables or {})
+    missing: set[str] = set()
+
+    def _repl(m: re.Match[str]) -> str:
+        key = str(m.group(1) or "")
+        if key in vars_in:
+            return str(vars_in.get(key) or "")
+        missing.add(key)
+        return ""
+
+    out = _VAR_RE.sub(_repl, doc.body)
+    if strict and missing:
+        raise ValueError(f"missing runtime prompt variables for {prompt_path}: {', '.join(sorted(missing))}")
+    return out.strip()
+
+
+__all__ = [
+    "PromptDoc",
+    "load_prompt_doc",
+    "load_runtime_prompt_doc",
+    "render_prompt",
+    "render_prompt_for_lang",
+    "render_runtime_prompt",
+]