From 4f53b9ce139c64f8d45da09d3fad6f1928233123 Mon Sep 17 00:00:00 2001 From: alexps9 <23210240262@m.fudan.edu.cn> Date: Thu, 16 Jul 2026 15:22:00 +0800 Subject: [PATCH] fix: resolve 13 pre-existing pytest failures Root causes and fixes: - PolicyRule.matches() ANDed the flattened `conditions` list on top of `condition_expr`, breaking any rule using OR/NOT (trace boolean conditions, console/legacy DSL rules with OR). `condition_expr` is now the sole source of truth when set; `conditions` remains for introspection only. - Added `extract_condition_atoms()` to flatten AND/OR/NOT expressions into their referenced field atoms, used by the console DSL parser and the legacy `.rules` compat parser to populate `PolicyRule.conditions` for introspection (previously only naive top-level AND-split atoms were captured). - `dsl_compat.parse_legacy_rules` treated "unsupported DSL feature" blocks (history_arg, exists_path, allowlist.*, etc.) as hard errors, failing the whole file even when most rules were valid. Downgraded to warnings so tutorial/example `.rules` files load their supported subset. - `shared.rules.loader.load_rules_file` (and the client-side mirror) assumed `.rules`-suffixed files are always legacy DSL text; now sniffs for JSON content first so JSON rule lists written to a `.rules` path still load. - RuntimeManager could downgrade an already-won `deny` decision back to `human_check` when enqueuing review tickets for non-winning plugin outcomes. Escalation now only happens when the current decision is actually weaker than the ticket being escalated (rank-based comparison). - Ported the shared v3 PolicyRule schema (agent_id, condition_expr, trace_clause, principal/tool/target views) into the client-side agentguard.schemas.policy so client-only rule loading (Guard(policy=...)) supports TRACE clauses, matching the server-side engine. - Removed a stray/misplaced assertion in the llamaindex streaming adapter test that asserted a tool_invoke event which the test explicitly says should never happen (wrap_tools=False), which crashed with "coroutine raised StopIteration" via bare next(). - Updated two llm_dsl_generator tests whose Chinese prompt-section assertions were stale after the prompt templates were translated to English in a prior commit. - Fixed a Windows-only path-separator artifact in `agentguard check` output (Path str uses backslashes on Windows) by printing as_posix(). All 290 tests pass (was 277 passed / 13 failed). No new lint issues in changed files (ruff clean); node --test unaffected (175 passed). Co-authored-by: Cursor --- src/client/python/agentguard/cli.py | 6 +- src/client/python/agentguard/rules/loader.py | 19 +- .../python/agentguard/schemas/policy.py | 439 +++++++++++++++++- src/server/backend/console/dsl.py | 96 +--- src/server/backend/runtime/manager.py | 19 +- src/server/backend/runtime/plugins/manager.py | 9 +- src/shared/rules/dsl_compat.py | 20 +- src/shared/rules/loader.py | 19 +- src/shared/schemas/policy.py | 26 +- tests/test_attach_adapters.py | 1 - tests/test_llm_dsl_generator.py | 6 +- 11 files changed, 550 insertions(+), 110 deletions(-) diff --git a/src/client/python/agentguard/cli.py b/src/client/python/agentguard/cli.py index 5bb4f0b..3f6e2e6 100644 --- a/src/client/python/agentguard/cli.py +++ b/src/client/python/agentguard/cli.py @@ -97,7 +97,7 @@ def _cmd_check(args: argparse.Namespace) -> int: except PolicyError as exc: print(str(exc), file=sys.stderr) return 1 - print(f"ok: {target} ({len(rules)} rules)") + print(f"ok: {target.as_posix()} ({len(rules)} rules)") return 0 @@ -121,11 +121,11 @@ def _check_rules_file(path: Path) -> tuple[bool, int]: _, report = parse_legacy_rules(source) if report.ok: - print(f"ok: {path} ({report.rule_count} rule block(s))") + print(f"ok: {path.as_posix()} ({report.rule_count} rule block(s))") return True, report.rule_count for error in report.errors: - print(f"error: {path}: {error['message']}", file=sys.stderr) + print(f"error: {path.as_posix()}: {error['message']}", file=sys.stderr) return False, report.rule_count diff --git a/src/client/python/agentguard/rules/loader.py b/src/client/python/agentguard/rules/loader.py index b270160..be07c62 100644 --- a/src/client/python/agentguard/rules/loader.py +++ b/src/client/python/agentguard/rules/loader.py @@ -29,17 +29,22 @@ def load_rules_file(path: str | Path) -> list[PolicyRule]: p = Path(path) if not p.exists(): raise PolicyError(f"rule file not found: {p}") - if p.suffix.lower() == ".rules": - try: - parsed, report = parse_legacy_rules(p.read_text(encoding="utf-8")) - except OSError as exc: - raise PolicyError(f"cannot read rule file {p}: {exc}") from exc + try: + text = p.read_text(encoding="utf-8") + except OSError as exc: + raise PolicyError(f"cannot read rule file {p}: {exc}") from exc + # `.rules` is the legacy text-DSL extension, but some callers write plain + # JSON rule lists to a `.rules`-suffixed path; sniff the content rather + # than trusting the suffix so both work. + looks_like_json = text.lstrip()[:1] in ("[", "{") + if p.suffix.lower() == ".rules" and not looks_like_json: + parsed, report = parse_legacy_rules(text) if not report.ok: raise PolicyError(f"cannot parse rule file {p}: {report.errors[0]['message']}") return [PolicyRule.from_dict(rule.to_dict()) for rule in parsed] try: - data = json.loads(p.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: + data = json.loads(text) + except json.JSONDecodeError as exc: raise PolicyError(f"cannot read rule file {p}: {exc}") from exc return _coerce_rules(data) diff --git a/src/client/python/agentguard/schemas/policy.py b/src/client/python/agentguard/schemas/policy.py index 903436f..14a9ab0 100644 --- a/src/client/python/agentguard/schemas/policy.py +++ b/src/client/python/agentguard/schemas/policy.py @@ -8,6 +8,7 @@ from agentguard.schemas.decisions import DecisionType from agentguard.schemas.events import RuntimeEvent +from shared.rules.trace_pattern import TraceStep, match_with_bindings class PolicyEffect(str, Enum): @@ -35,6 +36,20 @@ def effect_to_decision(effect: PolicyEffect) -> DecisionType: return _EFFECT_TO_DECISION[effect] +@dataclass +class TraceClause: + steps: list[TraceStep] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return {"steps": [step.to_dict() for step in self.steps]} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> TraceClause: + return cls( + steps=[TraceStep.from_dict(item) for item in data.get("steps") or []], + ) + + @dataclass class RuleCondition: """A single field predicate. `field` is a dotted path into the event dict. @@ -56,8 +71,12 @@ def from_dict(cls, data: dict[str, Any]) -> RuleCondition: def _resolve(path: str, root: dict[str, Any]) -> Any: + parts = path.split(".") + bindings = root.get("_trace_bindings") + if isinstance(bindings, dict) and parts and parts[0] in bindings: + return _resolve_trace_binding(bindings[parts[0]], parts[1:]) cur: Any = root - for part in path.split("."): + for part in parts: if isinstance(cur, dict) and part in cur: cur = cur[part] else: @@ -65,11 +84,34 @@ def _resolve(path: str, root: dict[str, Any]) -> Any: return cur +def _resolve_trace_binding(binding: dict[str, Any], parts: list[str]) -> Any: + if not parts: + return binding + head = parts[0] + if head == "name": + return binding.get("tool_name") + if head in {"boundary", "sensitivity", "integrity"}: + return (binding.get("labels") or {}).get(head) + if head == "result": + return binding.get("result") + return (binding.get("arguments") or {}).get(head) + + def _apply_op(op: str, actual: Any, expected: Any) -> bool: if op == "eq": return actual == expected if op == "ne": return actual != expected + if op == "gte": + try: + return float(actual) >= float(expected) + except (TypeError, ValueError): + return False + if op == "lte": + try: + return float(actual) <= float(expected) + except (TypeError, ValueError): + return False if op == "in": return actual in (expected or []) if op == "not_in": @@ -102,6 +144,7 @@ def _apply_op(op: str, actual: Any, expected: Any) -> bool: class PolicyRule: rule_id: str effect: PolicyEffect + agent_id: str | None = None reason: str = "" priority: int = 0 event_types: list[str] = field(default_factory=list) @@ -109,13 +152,16 @@ class PolicyRule: capabilities: list[str] = field(default_factory=list) risk_signals: list[str] = field(default_factory=list) conditions: list[RuleCondition] = field(default_factory=list) + condition_expr: str = "" metadata: dict[str, Any] = field(default_factory=dict) + trace_clause: TraceClause | None = None # ---- serialization ------------------------------------------------- def to_dict(self) -> dict[str, Any]: return { "rule_id": self.rule_id, "effect": self.effect.value, + "agent_id": self.agent_id, "reason": self.reason, "priority": self.priority, "event_types": list(self.event_types), @@ -123,7 +169,9 @@ def to_dict(self) -> dict[str, Any]: "capabilities": list(self.capabilities), "risk_signals": list(self.risk_signals), "conditions": [c.to_dict() for c in self.conditions], + "condition_expr": self.condition_expr, "metadata": self.metadata, + "trace_clause": self.trace_clause.to_dict() if self.trace_clause else None, } @classmethod @@ -131,6 +179,11 @@ def from_dict(cls, data: dict[str, Any]) -> PolicyRule: return cls( rule_id=data["rule_id"], effect=PolicyEffect(data["effect"]), + agent_id=( + str(data.get("agent_id")).strip() + if data.get("agent_id") not in (None, "") + else None + ), reason=data.get("reason", ""), priority=int(data.get("priority", 0)), event_types=list(data.get("event_types") or []), @@ -138,7 +191,13 @@ def from_dict(cls, data: dict[str, Any]) -> PolicyRule: capabilities=list(data.get("capabilities") or []), risk_signals=list(data.get("risk_signals") or []), conditions=[RuleCondition.from_dict(c) for c in data.get("conditions") or []], + condition_expr=str(data.get("condition_expr") or ""), metadata=dict(data.get("metadata") or {}), + trace_clause=( + TraceClause.from_dict(data["trace_clause"]) + if data.get("trace_clause") + else None + ), ) # ---- matching ------------------------------------------------------ @@ -147,6 +206,11 @@ def matches( event: RuntimeEvent, trace_window: list[RuntimeEvent] | None = None, ) -> bool: + if self.agent_id not in (None, ""): + event_agent_id = str(event.context.agent_id or "").strip() + if event_agent_id != str(self.agent_id).strip(): + return False + if self.event_types and event.event_type.value not in self.event_types: return False @@ -165,17 +229,140 @@ def matches( return False event_dict = event.to_dict() + principal = _principal_view(event) + tool = _tool_view(event) + target = _target_view(tool) + trace_bindings = _trace_bindings(self.trace_clause, event, trace_window or []) + if self.trace_clause is not None and trace_bindings is None: + return False + match_root = { + **event_dict, + "principal": principal, + "tool": tool, + "target": target, + "_trace_bindings": trace_bindings or {}, + } + if self.condition_expr.strip(): + # `conditions` is a flattened list of the atoms referenced inside + # `condition_expr` (kept for introspection/UI display); the boolean + # structure (AND/OR/NOT) only lives in the expression itself, so it + # is the sole source of truth here and must not be ANDed again. + return _evaluate_condition_expr(self.condition_expr, match_root, trace_window or []) for cond in self.conditions: if cond.field.startswith("trace."): if not _match_trace(cond, trace_window or []): return False continue - actual = _resolve(cond.field, event_dict) + actual = _resolve(cond.field, match_root) if not _apply_op(cond.op, actual, cond.value): return False return True +def _principal_view(event: RuntimeEvent) -> dict[str, Any]: + context = event.context + metadata_principal = {} + if isinstance(event.metadata, dict): + metadata_principal = dict(event.metadata.get("principal") or {}) + context_principal = {} + if isinstance(context.metadata, dict): + context_principal = dict(context.metadata.get("principal") or {}) + principal = { + **context_principal, + **metadata_principal, + "agent_id": context.agent_id, + "user_id": context.user_id, + "session_id": context.session_id, + } + if "role" not in principal and isinstance(context.metadata, dict): + principal["role"] = context.metadata.get("role") + if "trust_level" not in principal and isinstance(context.metadata, dict): + principal["trust_level"] = context.metadata.get("trust_level") + return principal + + +def _tool_view(event: RuntimeEvent) -> dict[str, Any]: + payload = event.payload.to_dict() + tool: dict[str, Any] = {} + tool_name = payload.get("tool_name") + if tool_name is not None: + tool["name"] = tool_name + arguments = payload.get("arguments") + if isinstance(arguments, dict): + tool.update(arguments) + result = payload.get("result") + if result is not None: + tool["result"] = result + capabilities = payload.get("capabilities") + if isinstance(capabilities, list): + tool["capabilities"] = list(capabilities) + labels = {} + if isinstance(event.metadata, dict): + labels = dict(event.metadata.get("labels") or event.metadata.get("tool_labels") or {}) + for key in ("boundary", "sensitivity", "integrity"): + if key in labels and labels.get(key) not in (None, ""): + tool[key] = labels.get(key) + return tool + + +def _target_view(tool: dict[str, Any]) -> dict[str, Any]: + url = tool.get("url") or tool.get("uri") or tool.get("endpoint") + recipient = tool.get("to") or tool.get("addr") or tool.get("email") + raw = url or recipient + domain = _extract_domain(str(raw)) if raw not in (None, "") else None + return { + "url": url, + "domain": domain, + "raw": raw, + } + + +def _trace_bindings( + clause: TraceClause | None, + event: RuntimeEvent, + window: list[RuntimeEvent], +) -> dict[str, dict[str, Any]] | None: + if clause is None: + return {} + if not clause.steps: + return None + entries = [_trace_entry(item) for item in window if _is_tool_event(item)] + if _is_tool_event(event): + entries.append(_trace_entry(event)) + return match_with_bindings(clause.steps, entries) + + +def _is_tool_event(event: RuntimeEvent) -> bool: + return event.event_type.value == "tool_invoke" + + +def _trace_entry(event: RuntimeEvent) -> dict[str, Any]: + payload = event.payload.to_dict() + labels = {} + if isinstance(event.metadata, dict): + labels = dict(event.metadata.get("labels") or event.metadata.get("tool_labels") or {}) + return { + "tool_name": payload.get("tool_name"), + "arguments": dict(payload.get("arguments") or {}), + "result": payload.get("result"), + "labels": labels, + } + + +def _extract_domain(raw: str) -> str | None: + text = str(raw or "").strip() + if not text: + return None + if "@" in text and "://" not in text: + return text.rsplit("@", 1)[-1].lower() + match = re.match(r"^[A-Za-z][A-Za-z0-9+.-]*://([^/:?#]+)", text) + if match: + return match.group(1).lower() + if re.match(r"^[A-Za-z0-9.-]+\.[A-Za-z]{2,}$", text): + return text.lower() + return None + + def _wildcard_match(value: Any, patterns: list[str]) -> bool: if value is None: return False @@ -187,6 +374,254 @@ def _wildcard_match(value: Any, patterns: list[str]) -> bool: return False +def _evaluate_condition_expr( + expr: str, + match_root: dict[str, Any], + trace_window: list[RuntimeEvent], +) -> bool: + tokens = _tokenize_condition_expr(expr) + if not tokens: + return True + parsed, index = _parse_or(tokens, 0) + if parsed is None or index != len(tokens): + return False + return _eval_condition_node(parsed, match_root, trace_window) + + +def _tokenize_condition_expr(expr: str) -> list[str]: + tokens: list[str] = [] + current: list[str] = [] + in_quote = False + quote_char = "" + brace_depth = 0 + i = 0 + while i < len(expr): + ch = expr[i] + if in_quote: + current.append(ch) + if ch == "\\" and i + 1 < len(expr): + i += 1 + current.append(expr[i]) + elif ch == quote_char: + in_quote = False + i += 1 + continue + if ch in {'"', "'"}: + in_quote = True + quote_char = ch + current.append(ch) + i += 1 + continue + if ch == "{": + brace_depth += 1 + current.append(ch) + i += 1 + continue + if ch == "}": + brace_depth = max(0, brace_depth - 1) + current.append(ch) + i += 1 + continue + if brace_depth == 0 and ch in "()": + token = "".join(current).strip() + if token: + tokens.append(token) + tokens.append(ch) + current = [] + i += 1 + continue + if brace_depth == 0 and ch.isspace(): + token = "".join(current).strip() + if token: + upper = token.upper() + if upper in {"AND", "OR", "NOT"}: + tokens.append(upper) + else: + tokens.append(token) + current = [] + i += 1 + continue + current.append(ch) + i += 1 + token = "".join(current).strip() + if token: + upper = token.upper() + tokens.append(upper if upper in {"AND", "OR", "NOT"} else token) + return _merge_condition_atoms(tokens) + + +def _merge_condition_atoms(tokens: list[str]) -> list[str]: + merged: list[str] = [] + operators = {"AND", "OR", "NOT", "(", ")"} + i = 0 + while i < len(tokens): + token = tokens[i] + if token in {"(", ")"}: + merged.append(token) + i += 1 + continue + if token in {"AND", "OR", "NOT"}: + merged.append(token) + i += 1 + continue + parts = [token] + j = i + 1 + while j < len(tokens) and tokens[j] not in operators: + parts.append(tokens[j]) + j += 1 + merged.append(" ".join(parts).strip()) + i = j + return merged + + +def _parse_or(tokens: list[str], index: int) -> tuple[Any, int]: + left, index = _parse_and(tokens, index) + if left is None: + return None, index + while index < len(tokens) and tokens[index] == "OR": + right, next_index = _parse_and(tokens, index + 1) + if right is None: + return None, index + left = ("or", left, right) + index = next_index + return left, index + + +def _parse_and(tokens: list[str], index: int) -> tuple[Any, int]: + left, index = _parse_not(tokens, index) + if left is None: + return None, index + while index < len(tokens) and tokens[index] == "AND": + right, next_index = _parse_not(tokens, index + 1) + if right is None: + return None, index + left = ("and", left, right) + index = next_index + return left, index + + +def _parse_not(tokens: list[str], index: int) -> tuple[Any, int]: + if index < len(tokens) and tokens[index] == "NOT": + node, next_index = _parse_not(tokens, index + 1) + if node is None: + return None, index + return ("not", node), next_index + return _parse_primary(tokens, index) + + +def _parse_primary(tokens: list[str], index: int) -> tuple[Any, int]: + if index >= len(tokens): + return None, index + token = tokens[index] + if token == "(": + node, next_index = _parse_or(tokens, index + 1) + if node is None or next_index >= len(tokens) or tokens[next_index] != ")": + return None, index + return node, next_index + 1 + if token == ")": + return None, index + return ("atom", token), index + 1 + + +def _eval_condition_node( + node: Any, + match_root: dict[str, Any], + trace_window: list[RuntimeEvent], +) -> bool: + kind = node[0] + if kind == "atom": + cond = _parse_expr_atom(node[1]) + if cond is None: + return False + if cond.field.startswith("trace."): + return _match_trace(cond, trace_window) + return _apply_op(cond.op, _resolve(cond.field, match_root), cond.value) + if kind == "not": + return not _eval_condition_node(node[1], match_root, trace_window) + if kind == "and": + return _eval_condition_node(node[1], match_root, trace_window) and _eval_condition_node( + node[2], match_root, trace_window + ) + if kind == "or": + return _eval_condition_node(node[1], match_root, trace_window) or _eval_condition_node( + node[2], match_root, trace_window + ) + return False + + +def extract_condition_atoms(expr: str) -> list[RuleCondition]: + """Flatten an AND/OR/NOT `condition_expr` into its atomic field conditions. + + Used to populate `PolicyRule.conditions` for introspection/UI purposes; the + boolean structure itself is preserved in `condition_expr` and is the only + thing `PolicyRule.matches()` evaluates when `condition_expr` is set. + """ + tokens = _tokenize_condition_expr(expr) + conditions: list[RuleCondition] = [] + for token in tokens: + if token in {"AND", "OR", "NOT", "(", ")"}: + continue + cond = _parse_expr_atom(token) + if cond is not None: + conditions.append(cond) + return conditions + + +def _parse_expr_atom(expr: str) -> RuleCondition | None: + parsed = re.match( + r'^(?P[A-Za-z_][A-Za-z0-9_.]*)\s+' + r'(?PNOT IN|MATCHES|CONTAINS|==|!=|>=|<=|>|<|IN)\s+' + r'(?P.+)$', + str(expr or "").strip(), + flags=re.IGNORECASE, + ) + if not parsed: + return None + return RuleCondition( + field=parsed.group("path").strip(), + op=_normalize_expr_op(parsed.group("op")), + value=_parse_expr_value(parsed.group("value")), + ) + + +def _normalize_expr_op(token: str) -> str: + normalized = str(token or "").strip().upper() + return { + "==": "eq", + "!=": "ne", + ">": "gt", + "<": "lt", + ">=": "gte", + "<=": "lte", + "IN": "in", + "NOT IN": "not_in", + "CONTAINS": "contains", + "MATCHES": "regex", + }.get(normalized, "eq") + + +def _parse_expr_value(raw_value: str) -> Any: + value = str(raw_value or "").strip() + if value.startswith("{") and value.endswith("}"): + inner = value[1:-1].strip() + if not inner: + return [] + return [_unquote_expr_value(item.strip()) for item in inner.split(",") if item.strip()] + if re.fullmatch(r"-?\d+", value): + return int(value) + if re.fullmatch(r"-?\d+\.\d+", value): + return float(value) + if value.lower() in {"true", "false"}: + return value.lower() == "true" + return _unquote_expr_value(value) + + +def _unquote_expr_value(value: str) -> str: + if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: + return value[1:-1] + return value + + def _match_trace(cond: RuleCondition, window: list[RuntimeEvent]) -> bool: key = cond.field.split(".", 1)[1] if key == "contains_event_type": diff --git a/src/server/backend/console/dsl.py b/src/server/backend/console/dsl.py index ea1f237..25804de 100644 --- a/src/server/backend/console/dsl.py +++ b/src/server/backend/console/dsl.py @@ -21,7 +21,13 @@ from typing import Any from shared.rules.trace_pattern import parse_trace_pattern, trace_steps_to_pattern -from shared.schemas.policy import PolicyEffect, PolicyRule, RuleCondition, TraceClause +from shared.schemas.policy import ( + PolicyEffect, + PolicyRule, + RuleCondition, + TraceClause, + extract_condition_atoms, +) ACTION_TO_EFFECT = { "DENY": PolicyEffect.DENY, @@ -175,90 +181,24 @@ def _on_event_types(block: str) -> list[str]: def _parse_conditions(cond_text: str) -> tuple[list[RuleCondition], list[dict[str, Any]]]: - """Translate DSL conditions to runtime conditions and preserve full source.""" - enforce: list[RuleCondition] = [] + """Translate DSL conditions to runtime conditions and preserve full source. + + `raw` mirrors the top-level AND-separated source segments (used to + reconstruct DSL source text); `enforce` is the full set of atomic field + conditions referenced anywhere in the expression (including inside + OR/NOT/parenthesized groups), kept on the rule for introspection only -- + `PolicyRule.matches()` evaluates `condition_expr` itself, not this list. + """ raw: list[dict[str, Any]] = [] parts = re.split(r"\s+AND\s+", cond_text, flags=re.IGNORECASE) for part in parts: expr = part.strip() - if not expr: - continue - raw.append({"expr": expr}) - compiled = _compile_condition(expr.strip("()")) - if compiled is not None: - enforce.append(compiled) + if expr: + raw.append({"expr": expr}) + enforce = extract_condition_atoms(cond_text) return enforce, raw -def _compile_condition(expr: str) -> RuleCondition | None: - parsed = re.match( - r'^(?P[A-Za-z_][A-Za-z0-9_.]*)\s+' - r'(?PNOT IN|MATCHES|CONTAINS|==|!=|>=|<=|>|<|IN)\s+' - r'(?P.+)$', - expr.strip(), - flags=re.IGNORECASE, - ) - if not parsed: - return None - field = _condition_field(parsed.group("path")) - if field is None: - return None - op = _condition_op(parsed.group("op")) - value = _condition_value(parsed.group("value")) - return RuleCondition(field=field, op=op, value=value) - - -def _condition_field(path: str) -> str | None: - normalized = str(path or "").strip() - if not normalized: - return None - if re.match(r"^[A-Za-z_][A-Za-z0-9_-]*\.", normalized): - return normalized - if normalized.startswith("principal."): - return normalized - if normalized == "tool.name": - return "tool.name" - if normalized.startswith("tool."): - return normalized - if normalized.startswith("target."): - return normalized - if normalized.startswith("payload."): - return normalized - return None - - -def _condition_op(token: str) -> str: - normalized = str(token or "").strip().upper() - return { - "==": "eq", - "!=": "ne", - ">": "gt", - "<": "lt", - ">=": "gte", - "<=": "lte", - "IN": "in", - "NOT IN": "not_in", - "CONTAINS": "contains", - "MATCHES": "regex", - }.get(normalized, "eq") - - -def _condition_value(raw_value: str) -> Any: - value = str(raw_value or "").strip() - if value.startswith("{") and value.endswith("}"): - inner = value[1:-1].strip() - if not inner: - return [] - return [_unquote(item.strip()) for item in inner.split(",") if item.strip()] - if re.fullmatch(r"-?\d+", value): - return int(value) - if re.fullmatch(r"-?\d+\.\d+", value): - return float(value) - if value.lower() in {"true", "false"}: - return value.lower() == "true" - return _unquote(value) - - # ---- public API -------------------------------------------------------- def parse_source(source: str) -> tuple[list[ParsedRule], CheckReport]: report = CheckReport() diff --git a/src/server/backend/runtime/manager.py b/src/server/backend/runtime/manager.py index 08bc13d..4ff0437 100644 --- a/src/server/backend/runtime/manager.py +++ b/src/server/backend/runtime/manager.py @@ -16,6 +16,7 @@ from backend.runtime.plugins import server_plugin_manager from backend.runtime.plugins.base import CheckResult from backend.runtime.plugins.config_utils import merge_plugin_configs, normalize_plugin_config +from backend.runtime.plugins.manager import decision_type_rank from backend.runtime.policy.engine import PolicyEngine from backend.runtime.review import ReviewQueue from backend.runtime.storage import SessionPool, TraceStore, trace_entry_event_dict @@ -503,7 +504,23 @@ def decide(self, request: dict[str, Any]) -> dict[str, Any]: check=check, ) if review_tickets: - if not (decision.requires_user or decision.requires_remote): + # A review ticket only needs to *escalate* the final decision when + # nothing stronger already won the plugin chain (e.g. a later + # plugin defaulted to allow without being final). If the chain + # already settled on something at least as severe as review + # (e.g. deny), that decision must not be downgraded back to + # human_check/remote_review. + strongest_ticket_rank = max( + ( + decision_type_rank(DecisionType(ticket.get("decision_type"))) + for ticket in review_tickets + if ticket.get("decision_type") + ), + default=-1, + ) + if not (decision.requires_user or decision.requires_remote) and ( + decision_type_rank(decision.decision_type) < strongest_ticket_rank + ): final_ticket = review_tickets[-1] final_reason = str(final_ticket.get("reason") or "Review required by server plugin.") final_policy_id = ( diff --git a/src/server/backend/runtime/plugins/manager.py b/src/server/backend/runtime/plugins/manager.py index 81d817f..edd5552 100644 --- a/src/server/backend/runtime/plugins/manager.py +++ b/src/server/backend/runtime/plugins/manager.py @@ -279,10 +279,17 @@ def _plugin_outcome_dict(plugin: BasePlugin, res: CheckResult) -> dict[str, Any] } +def decision_type_rank(decision_type: DecisionType | None) -> int: + """Severity rank for a decision type; higher wins when candidates tie on `is_final`.""" + if decision_type is None: + return -1 + return _DECISION_RANK.get(decision_type, -1) + + def _decision_rank(decision: GuardDecision | None) -> int: if decision is None: return -1 - return _DECISION_RANK.get(decision.decision_type, -1) + return decision_type_rank(decision.decision_type) def _should_replace_decision( diff --git a/src/shared/rules/dsl_compat.py b/src/shared/rules/dsl_compat.py index 6bcb726..af54715 100644 --- a/src/shared/rules/dsl_compat.py +++ b/src/shared/rules/dsl_compat.py @@ -6,7 +6,13 @@ from typing import Any from shared.rules.trace_pattern import parse_trace_pattern -from shared.schemas.policy import PolicyEffect, PolicyRule, RuleCondition, TraceClause +from shared.schemas.policy import ( + PolicyEffect, + PolicyRule, + RuleCondition, + TraceClause, + extract_condition_atoms, +) _ACTION_TO_EFFECT = { "DENY": PolicyEffect.DENY, @@ -227,7 +233,7 @@ def _supports_runtime(fields: dict[str, str]) -> bool: def _runtime_rule(fields: dict[str, str], action: str) -> PolicyRule: condition_text = str(fields.get("CONDITION", "")).strip() raw_conditions = [part.strip() for part in re.split(r"\s+AND\s+", condition_text, flags=re.IGNORECASE) if part.strip()] - conditions = [_compile_condition(expr.strip("()")) for expr in raw_conditions] + conditions = extract_condition_atoms(condition_text) tool_pattern = _tool_pattern(fields.get("ON", "")) prompt = _unquote(fields.get("Prompt", "")) metadata = { @@ -252,7 +258,7 @@ def _runtime_rule(fields: dict[str, str], action: str) -> PolicyRule: priority=_PRIORITY_BY_ACTION.get(action, 50), event_types=_on_event_types(fields.get("ON", "")), tool_names=[] if tool_pattern in ("", "*") else [tool_pattern], - conditions=[condition for condition in conditions if condition is not None], + conditions=conditions, condition_expr=condition_text, metadata=metadata, trace_clause=( @@ -306,11 +312,15 @@ def parse_legacy_rules(source: str) -> tuple[list[PolicyRule], DSLCompatReport]: parsed.append(_runtime_rule(fields, action)) continue - report.errors.append( + # Not a syntax error -- the block is well-formed DSL, but references + # features the current runtime compiler doesn't execute yet (e.g. + # history_arg(), exists_path(), allowlist.*). Skip it without failing + # the whole file so the rest of a tutorial/example file still loads. + report.warnings.append( { "message": ( f"Rule block {index} ('{rule_id}') uses unsupported DSL features in the current " - "runtime compiler." + "runtime compiler; skipped." ) } ) diff --git a/src/shared/rules/loader.py b/src/shared/rules/loader.py index 47e70bc..f43b6cc 100644 --- a/src/shared/rules/loader.py +++ b/src/shared/rules/loader.py @@ -29,17 +29,22 @@ def load_rules_file(path: str | Path) -> list[PolicyRule]: p = Path(path) if not p.exists(): raise PolicyError(f"rule file not found: {p}") - if p.suffix.lower() == ".rules": - try: - parsed, report = parse_legacy_rules(p.read_text(encoding="utf-8")) - except OSError as exc: - raise PolicyError(f"cannot read rule file {p}: {exc}") from exc + try: + text = p.read_text(encoding="utf-8") + except OSError as exc: + raise PolicyError(f"cannot read rule file {p}: {exc}") from exc + # `.rules` is the legacy text-DSL extension, but some callers (and tests) + # write plain JSON rule lists to a `.rules`-suffixed path; sniff the + # content rather than trusting the suffix so both work. + looks_like_json = text.lstrip()[:1] in ("[", "{") + if p.suffix.lower() == ".rules" and not looks_like_json: + parsed, report = parse_legacy_rules(text) if not report.ok: raise PolicyError(f"cannot parse rule file {p}: {report.errors[0]['message']}") return parsed try: - data = json.loads(p.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: + data = json.loads(text) + except json.JSONDecodeError as exc: raise PolicyError(f"cannot read rule file {p}: {exc}") from exc return _coerce_rules(data) diff --git a/src/shared/schemas/policy.py b/src/shared/schemas/policy.py index 2285635..0295aec 100644 --- a/src/shared/schemas/policy.py +++ b/src/shared/schemas/policy.py @@ -243,8 +243,11 @@ def matches( "_trace_bindings": trace_bindings or {}, } if self.condition_expr.strip(): - if not _evaluate_condition_expr(self.condition_expr, match_root, trace_window or []): - return False + # `conditions` is a flattened list of the atoms referenced inside + # `condition_expr` (kept for introspection/UI display); the boolean + # structure (AND/OR/NOT) only lives in the expression itself, so it + # is the sole source of truth here and must not be ANDed again. + return _evaluate_condition_expr(self.condition_expr, match_root, trace_window or []) for cond in self.conditions: if cond.field.startswith("trace."): if not _match_trace(cond, trace_window or []): @@ -542,6 +545,25 @@ def _eval_condition_node( return False +def extract_condition_atoms(expr: str) -> list[RuleCondition]: + """Flatten an AND/OR/NOT `condition_expr` into its atomic field conditions. + + This is used to populate `PolicyRule.conditions` for introspection/UI + purposes (e.g. "which fields does this rule reference?"). It intentionally + ignores the boolean structure -- `matches()` only evaluates `condition_expr` + itself when it is set. + """ + tokens = _tokenize_condition_expr(expr) + conditions: list[RuleCondition] = [] + for token in tokens: + if token in {"AND", "OR", "NOT", "(", ")"}: + continue + cond = _parse_expr_atom(token) + if cond is not None: + conditions.append(cond) + return conditions + + def _parse_expr_atom(expr: str) -> RuleCondition | None: parsed = re.match( r'^(?P[A-Za-z_][A-Za-z0-9_.]*)\s+' diff --git a/tests/test_attach_adapters.py b/tests/test_attach_adapters.py index b0403f6..791e7a6 100644 --- a/tests/test_attach_adapters.py +++ b/tests/test_attach_adapters.py @@ -1294,7 +1294,6 @@ async def _call_tool(self, ctx, tool, tool_input): assert _event_types(guard).count("llm_input") == 1 assert _event_types(guard).count("llm_output") == 1 assert _first_event(guard, "llm_output").metadata["label"] == "astream_chat" - assert _first_event(guard, "tool_invoke").payload.arguments == {"message": "hello"} @pytest.mark.asyncio diff --git a/tests/test_llm_dsl_generator.py b/tests/test_llm_dsl_generator.py index ba0cd7e..744cd99 100644 --- a/tests/test_llm_dsl_generator.py +++ b/tests/test_llm_dsl_generator.py @@ -119,7 +119,7 @@ def test_generate_repairs_after_validation_failure() -> None: assert len(session.attempts) == 2 assert session.attempts[0].validation.ok is False assert session.accepted_candidate is not None - assert "上一轮输出未通过校验" in client.prompts[1] + assert "current round output failed validation" in client.prompts[1] assert "unknown_tool_match" in client.prompts[1] @@ -148,8 +148,8 @@ def test_refine_uses_user_feedback_and_revalidates() -> None: assert updated.accepted_candidate is not None assert len(updated.attempts) == 2 assert updated.user_feedback_history == ["不要直接 deny，改成 LLM_CHECK 审查邮件内容"] - assert "用户修改意见" in client.prompts[1] - assert "当前已通过校验的候选规则" in client.prompts[1] + assert "user feedback" in client.prompts[1] + assert "current accepted candidate rules" in client.prompts[1] normalized = updated.accepted_candidate.validation.normalized_rules[0] assert normalized.effect.value == "require_remote_review" assert normalized.metadata["llm_prompt"].startswith("Decide allow, deny, or human_check")