executor.py

"""
Experience Runtime — 确定性执行器（Executor）
==============================================
不依赖 LLM 的执行内核：
  1. 从 provider 执行 workflow 步骤，构建上下文
  2. 填充模板（{var} 占位符，缺失时保留原样并记录）
  3. 求值 decision_rules（安全表达式，非 eval）
  4. 求值 exception_rules（命中则降级 delegate）
  5. 输出：模板结果 + workflow 汇总 + 规则应用 + 警告

安全表达式（SafeEvaluator）：
  支持比较 / 布尔 / in / 括号 / 数字 / 字符串 / 上下文变量。
  用递归下降解析实现，绝不使用 eval/exec。
"""

import json
import logging
import re
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional

logger = logging.getLogger("experience_runtime.executor")

_PLACEHOLDER_RE = re.compile(r"\{(\w+)\}")

# ─────────────────────────────────────────────
# 安全表达式求值器
# ─────────────────────────────────────────────

_TOKEN_SPEC = [
    ("NUMBER", r"\d+(?:\.\d+)?"),
    ("STRING", r"'[^']*'|\"[^\"]*\""),
    ("OP", r"==|!=|>=|<=|>|<|\b(in|and|or|not)\b|\(|\)"),
    ("WORD", r"[A-Za-z_][A-Za-z0-9_.]*"),
    ("SPACE", r"\s+"),
]

_TOKEN_RE = re.compile("|".join(f"(?P<{n}>{p})" for n, p in _TOKEN_SPEC))


class _ExprParser:
    def __init__(self, tokens, ctx):
        self.tokens = tokens
        self.pos = 0
        self.ctx = ctx

    def peek(self):
        return self.tokens[self.pos] if self.pos < len(self.tokens) else None

    def next(self):
        t = self.peek()
        self.pos += 1
        return t

    def parse(self):
        v = self.or_expr()
        if self.peek() is not None:
            raise ValueError(f"多余 token: {self.peek()}")
        return v

    def or_expr(self):
        v = self.and_expr()
        while self.peek() and self.peek()[0] == "OP" and self.peek()[1] == "or":
            self.next()
            r = self.and_expr()
            v = bool(v) or bool(r)
        return v

    def and_expr(self):
        v = self.not_expr()
        while self.peek() and self.peek()[0] == "OP" and self.peek()[1] == "and":
            self.next()
            r = self.not_expr()
            v = bool(v) and bool(r)
        return v

    def not_expr(self):
        if self.peek() and self.peek()[0] == "OP" and self.peek()[1] == "not":
            self.next()
            return not bool(self.not_expr())
        return self.comparison()

    def comparison(self):
        left = self.operand()
        t = self.peek()
        if t and t[0] == "OP" and t[1] in ("==", "!=", ">", "<", ">=", "<=", "in"):
            self.next()
            right = self.operand()
            op = t[1]
            if op == "==":
                return left == right
            if op == "!=":
                return left != right
            if op == ">":
                return left > right
            if op == "<":
                return left < right
            if op == ">=":
                return left >= right
            if op == "<=":
                return left <= right
            if op == "in":
                return left in right
        return left

    def operand(self):
        t = self.next()
        if t is None:
            raise ValueError("表达式意外结束")
        kind, val = t
        if kind == "NUMBER":
            return float(val) if "." in val else int(val)
        if kind == "STRING":
            return val[1:-1]
        if kind == "WORD":
            low = val.lower()
            if low == "true":
                return True
            if low == "false":
                return False
            if low in ("none", "null"):
                return None
            return _resolve(self.ctx, val)
        if kind == "OP" and val == "(":
            v = self.or_expr()
            t2 = self.next()
            if not (t2 and t2[0] == "OP" and t2[1] == ")"):
                raise ValueError("缺少右括号")
            return v
        raise ValueError(f"意外的 token: {t}")


def _resolve(ctx: Dict, path: str) -> Any:
    parts = path.split(".")
    cur = ctx
    for p in parts:
        if isinstance(cur, dict) and p in cur:
            cur = cur[p]
        elif isinstance(cur, list) and p.isdigit() and int(p) < len(cur):
            cur = cur[int(p)]
        else:
            return None
    return cur


class SafeEvaluator:
    """安全表达式求值入口：SafeEvaluator.evaluate("x > 3 and y == 'a'", ctx)。"""

    @staticmethod
    def tokenize(expr: str):
        tokens = []
        for m in _TOKEN_RE.finditer(expr):
            kind = m.lastgroup
            val = m.group()
            if kind == "SPACE":
                continue
            if kind == "OP":
                val = val.strip()
            tokens.append((kind, val))
        return tokens

    @staticmethod
    def evaluate(expr: str, ctx: Dict) -> Any:
        expr = (expr or "").strip()
        if not expr:
            return True
        tokens = SafeEvaluator.tokenize(expr)
        if not tokens:
            return True
        return _ExprParser(tokens, ctx).parse()


# ─────────────────────────────────────────────
# 执行器
# ─────────────────────────────────────────────

class ExperienceExecutor:
    """执行一条经验，返回确定性输出。"""

    def __init__(self, data_providers: Optional[Dict[str, Callable]] = None):
        self.data_providers = data_providers or {}

    def run(self, experience: Dict, state: Dict) -> Dict:
        ctx = self._build_context(state, experience)

        # 1. workflow：按序调用宿主 provider，结果合并进 ctx
        workflow_results = []
        for step in experience.get("exec", {}).get("workflow", []):
            res = self._run_step(step, ctx)
            workflow_results.append(res)
            if res.get("status") == "ok" and isinstance(res.get("data"), dict):
                ctx.update(res["data"])

        # 2. 模板填充（缺失占位符自动尝试同名 provider 补齐）
        template = experience.get("exec", {}).get("template", {})
        output, missing = _fill_template(template, ctx)
        output, missing = self._auto_resolve_missing(template, output, missing, ctx)

        # 3. decision_rules
        decisions = []
        for rule in experience.get("exec", {}).get("decision_rules", []):
            try:
                cond = SafeEvaluator.evaluate(rule.get("if", ""), ctx)
            except Exception:
                cond = False
            if cond:
                applied = _apply_action(rule.get("then", ""), ctx, output)
                decisions.append({"rule": rule.get("if", ""), "action": rule.get("then", ""),
                                  "applied": applied})

        # 4. exception_rules → 降级 delegate
        exception_hit = False
        for rule in experience.get("exec", {}).get("exception_rules", []):
            try:
                if SafeEvaluator.evaluate(rule.get("if", ""), ctx):
                    exception_hit = True
                    break
            except Exception:
                continue

        warnings = list(ctx.get("_warnings", []))
        if missing:
            warnings.append(f"模板存在未填充占位符: {', '.join(missing[:5])}")

        return {
            "decision": "delegate" if exception_hit else "execute",
            "output": output,
            "exec_context": {
                k: v for k, v in ctx.items()
                if not k.startswith("_") and _is_jsonable(v)
            },
            "workflow_results": workflow_results,
            "decisions_applied": decisions,
            "missing_placeholders": missing,
            "warnings": warnings,
            "exception_hit": exception_hit,
            "executed_at": datetime.now().isoformat(timespec="seconds"),
        }

    # ── 内部 ──

    def _build_context(self, state: Dict, experience: Dict) -> Dict:
        ctx = {
            "input": state.get("input", ""),
            "entities": dict(state.get("entities", {}) or {}),
            "context": dict(state.get("context", {}) or {}),
            "provider_state": dict(state.get("provider_state", {}) or {}),
            "tokens": list(state.get("tokens", [])),
            "_warnings": [],
        }
        # 实体扁平化到顶层（模板可直接用 {time_range} / {amount}）
        ctx.update(ctx["entities"])
        ctx.update(ctx["provider_state"])  # provider_state 变量也可用于模板占位符 {d} 等
        return ctx

    def _auto_resolve_missing(self, template, output, missing, ctx):
        """对缺失占位符，尝试调用同名 data_provider 补齐（最多 2 轮）。"""
        for _pass in range(2):
            if not missing:
                break
            changed = False
            for name in list(missing):
                fn = self.data_providers.get(name)
                if not callable(fn):
                    continue
                try:
                    data = fn({}, ctx) or {}
                except Exception:
                    data = {}
                if isinstance(data, dict) and data:
                    ctx.update(data)
                    changed = True
            if not changed:
                break
            output, missing = _fill_template(template, ctx)
        return output, missing

    def _run_step(self, step: Dict, ctx: Dict) -> Dict:
        name = str(step.get("name") or "")
        provider = str(step.get("provider") or "")
        params = step.get("params") or {}
        fn = self.data_providers.get(provider)
        if not callable(fn):
            return {"name": name, "provider": provider, "status": "skipped",
                    "error": f"provider 未注册: {provider}"}
        try:
            data = fn(params, ctx)
            if data is None:
                data = {}
            return {"name": name, "provider": provider, "status": "ok", "data": data}
        except Exception as e:
            logger.warning("[EXECUTOR] workflow 步骤失败 %s: %s", name, e)
            return {"name": name, "provider": provider, "status": "error", "error": str(e)}


def _fill_template(template, ctx: Dict):
    """递归填充 {var}；缺失的占位符保留原样并记录。"""
    missing = []

    def fill_value(v):
        if isinstance(v, dict):
            return {k: fill_value(x) for k, x in v.items()}
        if isinstance(v, list):
            return [fill_value(x) for x in v]
        if isinstance(v, str):
            def repl(m):
                key = m.group(1)
                if key in ctx:
                    return _fmt(ctx[key])
                missing.append(key)
                return m.group(0)
            return _PLACEHOLDER_RE.sub(repl, v)
        return v

    out = fill_value(template)
    return out, sorted(set(missing))


def _fmt(value: Any) -> str:
    if isinstance(value, float):
        return f"{value:,.2f}" if abs(value - round(value)) > 1e-9 else f"{value:,.0f}"
    if isinstance(value, (dict, list)):
        try:
            return json.dumps(value, ensure_ascii=False)
        except Exception:
            return str(value)
    return str(value)


def _apply_action(action: str, ctx: Dict, output: Any) -> bool:
    action = (action or "").strip()
    if not action:
        return False
    if action.startswith("set:"):
        parts = action.split(":", 2)
        if len(parts) == 3:
            ctx[parts[1]] = _parse_scalar(parts[2])
            return True
    if action.startswith("append:"):
        parts = action.split(":", 2)
        if len(parts) == 3:
            ctx.setdefault("_warnings", []).append(parts[2])
            return True
    if action.startswith("warning:"):
        ctx.setdefault("_warnings", []).append(action.split(":", 1)[1])
        return True
    return False


def _parse_scalar(s: str) -> Any:
    try:
        return int(s)
    except ValueError:
        pass
    try:
        return float(s)
    except ValueError:
        pass
    return s


def _is_jsonable(v: Any) -> bool:
    try:
        json.dumps(v)
        return True
    except Exception:
        return False