`jianmu.guard`¶

适用对象：安全策略作者 / 应用集成者 / 核心维护者是否必读：按需相关模块：jianmu.execution, jianmu.tool, jianmu.skill

1. 模块职责¶

jianmu.guard 负责把预算、速率、确认、工具策略等约束整合成统一守卫层。

它的定位不是替代执行层，而是在执行前后提供可组合的策略判定与审批机制。

2. 适合查什么¶

统一约束类型：GuardConstraints、BudgetConstraints
守卫入口：GuardEnforcer
审批相关对象：ApprovalManager、ApprovalRequest
内置 checker：预算、确认、速率、工具策略

3. 使用建议¶

应用侧通常组合 Constraints + GuardEnforcer
需要人工审批流时，再引入 ApprovalManager
新策略优先通过 checker 协议扩展，而不是在节点内部散落校验逻辑

4. 最小示例¶

from jianmu.config.constraints import Constraints
from jianmu.guard.types import GuardConstraints, BudgetConstraints

constraints = Constraints(
    guard=GuardConstraints(
        max_tool_calls=5,
        tool_policy={"bash": "confirm"},
        budget=BudgetConstraints(total_tokens=20000),
    ),
)

5. 常见入口¶

想快速给 agent 加限制：看 GuardConstraints
想接人工审批：看 ApprovalManager
想自己拼 checker：看 GuardEnforcer

6. API 参考¶

决策与约束¶

Decision ¶

Bases: Enum

Possible outcomes of a policy check.

BudgetConstraints `dataclass` ¶

BudgetConstraints(
    total_tokens: Optional[int] = None,
    prompt_tokens: Optional[int] = None,
    completion_tokens: Optional[int] = None,
)

Token budget limits shared across one agent run or loop.

Any field left as None is treated as unlimited. Most applications only set total_tokens; the prompt/completion split is useful when a provider reports the components separately and you want finer control.

属性：

名称	类型	描述
`total_tokens`	`Optional[int]`	Overall token ceiling across the whole run.
`prompt_tokens`	`Optional[int]`	Optional prompt-token ceiling across all model calls.
`completion_tokens`	`Optional[int]`	Optional completion-token ceiling across all model calls.

GuardConstraints `dataclass` ¶

GuardConstraints(
    tool_policy: dict[str, str] = None,
    tool_rules: list[ToolRule] = None,
    default_policy: str = "allow",
    max_tool_calls: Optional[int] = None,
    budget: BudgetConstraints = None,
    permission: PermissionConstraints | None = None,
)

Guard-domain constraints for policy, budget, and approval decisions.

from_dict `classmethod` ¶

from_dict(data: dict | None) -> 'GuardConstraints'

Build guard constraints from a plain mapping.

参数：

名称	类型	描述	默认
`data`	`dict \| None`	The `data` value.	必需

返回：

类型	描述
`'GuardConstraints'`	The resulting `'GuardConstraints'` value.

源代码位于： jianmu/guard/types.py

@classmethod
def from_dict(cls, data: dict | None) -> "GuardConstraints":
    """Build guard constraints from a plain mapping.

    Args:
        data: The `data` value.

    Returns:
        The resulting `'GuardConstraints'` value.
    """
    payload = dict(data or {})
    budget_meta = payload.get("budget", {}) or {}
    return cls(
        tool_policy=dict(payload.get("tool_policy", {}) or {}),
        tool_rules=[_coerce_tool_rule(rule) for rule in list(payload.get("tool_rules", []) or [])],
        default_policy=str(payload.get("default_policy", "allow") or "allow"),
        max_tool_calls=payload.get("max_tool_calls"),
        budget=BudgetConstraints(
            total_tokens=budget_meta.get("total_tokens"),
            prompt_tokens=budget_meta.get("prompt_tokens"),
            completion_tokens=budget_meta.get("completion_tokens"),
        ),
        permission=(
            PermissionConstraints.from_dict(payload.get("permission"))
            if payload.get("permission") is not None
            else None
        ),
    )

守卫入口¶

GuardEnforcer ¶

GuardEnforcer(checkers: List[CheckerProtocol])

Bases: GuardProtocol

Compose multiple guard checkers into one decision point.

Most users should configure Constraints and let Jianmu build an enforcer automatically. This class is mainly useful when you need explicit checker composition or a custom enforcement pipeline.

Compose a list of guard checkers into one enforcer.

参数：

名称	类型	描述	默认
`checkers`	`List[CheckerProtocol]`	Ordered guard checkers evaluated for each action.	必需

源代码位于： jianmu/guard/enforcer.py

def __init__(self, checkers: List[CheckerProtocol]):
    """Compose a list of guard checkers into one enforcer.

    Args:
        checkers: Ordered guard checkers evaluated for each action.
    """
    self.checkers = checkers

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Run all registered checkers and stop at the first hard deny.

参数：

名称	类型	描述	默认
`action`	`str`	Action string to check.	必需
`context`	`CheckContextLike`	Typed or legacy context environment.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple containing the final Decision and reason.

源代码位于： jianmu/guard/enforcer.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Run all registered checkers and stop at the first hard deny.

    Args:
        action: Action string to check.
        context: Typed or legacy context environment.

    Returns:
        Tuple containing the final Decision and reason.
    """
    _populate_tool_analysis(action, context)
    decision: Decision = Decision.ALLOW
    reason = ""
    for checker in self.checkers:
        d, r = await checker.check(action, context)
        if d == Decision.DENY:
            return d, r
        if d == Decision.PENDING:
            return d, r
        if d == Decision.CONFIRM:
            decision, reason = Decision.CONFIRM, r
            # keep going; later checkers may DENY or ALLOW with confirmation
        if d == Decision.ALLOW and not _needs_confirm(context):
            decision = Decision.ALLOW
            if r:
                reason = r
    return decision, reason

reset ¶

reset() -> None

Reset all registered checkers.

源代码位于： jianmu/guard/enforcer.py

def reset(self) -> None:
    """Reset all registered checkers."""
    for c in self.checkers:
        if hasattr(c, "reset"):
            c.reset()

dump_resume_state ¶

dump_resume_state() -> dict

Return checker-local state for checkpoint resume.

源代码位于： jianmu/guard/enforcer.py

def dump_resume_state(self) -> dict:
    """Return checker-local state for checkpoint resume."""
    entries = []
    for checker in self.checkers:
        dump_resume_state = getattr(checker, "dump_resume_state", None)
        if not callable(dump_resume_state):
            continue
        state = dump_resume_state()
        if not isinstance(state, dict) or not state:
            continue
        entries.append(
            {
                "type": f"{checker.__class__.__module__}.{checker.__class__.__qualname__}",
                "state": state,
            }
        )
    return {"checkers": entries} if entries else {}

restore_resume_state ¶

restore_resume_state(data: dict) -> None

Restore checker-local state from checkpoint metadata.

源代码位于： jianmu/guard/enforcer.py

def restore_resume_state(self, data: dict) -> None:
    """Restore checker-local state from checkpoint metadata."""
    entries = list(dict(data or {}).get("checkers") or [])
    if not entries:
        return
    by_type: dict[str, list[dict]] = {}
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        by_type.setdefault(str(entry.get("type") or ""), []).append(entry)
    for checker in self.checkers:
        restore_resume_state = getattr(checker, "restore_resume_state", None)
        if not callable(restore_resume_state):
            continue
        checker_type = f"{checker.__class__.__module__}.{checker.__class__.__qualname__}"
        bucket = by_type.get(checker_type) or []
        entry = bucket.pop(0) if bucket else None
        if not entry:
            continue
        state = entry.get("state")
        if isinstance(state, dict):
            restore_resume_state(state)

from_guard_constraints `classmethod` ¶

from_guard_constraints(
    constraints: GuardConstraints | None,
    tool_aliases: dict[str, str] | None = None,
    approval_manager: ApprovalManager | None = None,
    permission_grant_store: PermissionGrantStoreProtocol
    | None = None,
    permission_scope_key: str = "",
) -> "GuardEnforcer"

Build a guard enforcer from guard-domain constraint settings.

参数：

名称	类型	描述	默认
`constraints`	`GuardConstraints \| None`	Guard-domain constraints specifying policies/limits.	必需
`tool_aliases`	`dict[str, str] \| None`	Optional dictionary of tool name aliases.	`None`
`approval_manager`	`ApprovalManager \| None`	Optional ApprovalManager instance.	`None`
`permission_grant_store`	`PermissionGrantStoreProtocol \| None`	Optional saved Permission V2 allow store.	`None`
`permission_scope_key`	`str`	Opaque host scope used to load saved rules.	`''`

返回：

类型	描述
`'GuardEnforcer'`	An instance of GuardEnforcer.

The resulting checker chain typically includes tool policy, call-rate limits, budget checks, and approval confirmation.

源代码位于： jianmu/guard/enforcer.py

@classmethod
def from_guard_constraints(
    cls,
    constraints: GuardConstraints | None,
    tool_aliases: dict[str, str] | None = None,
    approval_manager: ApprovalManager | None = None,
    permission_grant_store: PermissionGrantStoreProtocol | None = None,
    permission_scope_key: str = "",
) -> "GuardEnforcer":
    """Build a guard enforcer from guard-domain constraint settings.

    Args:
        constraints: Guard-domain constraints specifying policies/limits.
        tool_aliases: Optional dictionary of tool name aliases.
        approval_manager: Optional ApprovalManager instance.
        permission_grant_store: Optional saved Permission V2 allow store.
        permission_scope_key: Opaque host scope used to load saved rules.

    Returns:
        An instance of GuardEnforcer.

    The resulting checker chain typically includes tool policy, call-rate
    limits, budget checks, and approval confirmation.
    """
    if tool_aliases is None:
        tool_aliases = BuiltinToolProvider.aliases()
    guard_constraints = constraints or GuardConstraints()
    checkers: List[CheckerProtocol] = []
    if guard_constraints.permission is not None:
        if guard_constraints.tool_policy or guard_constraints.tool_rules:
            raise ValueError(
                "Guard permission V2 cannot be combined with legacy tool_policy/tool_rules. "
                "Use either guard.permission or guard.tool_policy/tool_rules."
            )
        store = permission_grant_store or getattr(approval_manager, "permission_grant_store", None)
        scope_key = permission_scope_key or str(getattr(approval_manager, "permission_scope_key", "") or "")
        checkers.append(
            PermissionRuleChecker(
                guard_constraints.permission.rules,
                guard_constraints.permission.default_effect,
                grant_store=store,
                scope_key=scope_key,
            )
        )
    elif guard_constraints.tool_policy or guard_constraints.default_policy:
        checkers.append(
            ToolPolicyChecker(
                guard_constraints.tool_policy,
                guard_constraints.default_policy or "allow",
                aliases=tool_aliases,
                tool_rules=guard_constraints.tool_rules,
            )
        )

    if guard_constraints.max_tool_calls is not None:
        checkers.append(RateLimitChecker(max_calls=guard_constraints.max_tool_calls))

    if guard_constraints.budget:
        checkers.append(BudgetChecker(guard_constraints.budget))

    if approval_manager is None:
        approval_manager = ApprovalManager()
    checkers.append(ConfirmChecker(approval_manager))

    return cls(checkers)

CheckerProtocol ¶

Bases: Protocol

Single checker that decides allow/deny/confirm for an action.

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Validate one constraint and return (decision, reason).

参数：

名称	类型	描述	默认
`action`	`str`	Action string to check.	必需
`context`	`CheckContextLike`	Typed or legacy dictionary context parameters.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple containing the Decision and a reason string.

源代码位于： jianmu/guard/base.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Validate one constraint and return ``(decision, reason)``.

    Args:
        action: Action string to check.
        context: Typed or legacy dictionary context parameters.

    Returns:
        Tuple containing the Decision and a reason string.
    """
    ...

reset ¶

reset() -> None

Reset any internal checker state.

源代码位于： jianmu/guard/base.py

def reset(self) -> None:
    """Reset any internal checker state."""
    ...

GuardProtocol ¶

Bases: Protocol

Composed guard made of multiple checkers.

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Validate the composed guard and return (decision, reason).

参数：

名称	类型	描述	默认
`action`	`str`	Action string to check.	必需
`context`	`CheckContextLike`	Typed or legacy dictionary context parameters.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple containing the Decision and a reason string.

源代码位于： jianmu/guard/base.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Validate the composed guard and return ``(decision, reason)``.

    Args:
        action: Action string to check.
        context: Typed or legacy dictionary context parameters.

    Returns:
        Tuple containing the Decision and a reason string.
    """
    ...

reset ¶

reset() -> None

Reset all underlying checker state.

源代码位于： jianmu/guard/base.py

def reset(self) -> None:
    """Reset all underlying checker state."""
    ...

检查上下文¶

ToolCheckContext `dataclass` ¶

ToolCheckContext(
    tool_name: str,
    args: dict[str, Any],
    thread_id: str = "",
    node_locator: str = "",
    tool_call_id: str = "",
    command_text: str = "",
    command_prefix: str = "",
    target_path: str = "",
    resource_kind: str = "",
    risk_tags: list[str] = list(),
    matched_rule: dict[str, Any] | None = None,
    permission: PermissionCheck | None = None,
)

Typed tool-call context consumed by guard checkers.

ApprovalCheckEffect `dataclass` ¶

ApprovalCheckEffect(
    progress_state: str = "",
    request_id: str = "",
    approval_result: Any | None = None,
    clear_progress: bool = False,
    pending: bool = False,
)

Mutable approval-check side effects produced by a checker.

ApprovalCheckContext `dataclass` ¶

ApprovalCheckContext(
    request_id: str,
    reason: str = "",
    progress_state: str = "",
    pop_resolved: Callable[[str], dict[str, Any] | None]
    | None = None,
    needs_confirm: bool = False,
    effect: ApprovalCheckEffect = ApprovalCheckEffect(),
)

Typed approval state consumed and mutated during one guard check.

GuardCheckContext `dataclass` ¶

GuardCheckContext(
    action: str,
    usage: dict[str, Any] = dict(),
    metadata: dict[str, Any] = dict(),
    tool: ToolCheckContext | None = None,
    approval: ApprovalCheckContext | None = None,
)

Typed guard context that still behaves like a legacy mapping.

get ¶

get(key: str, default: Any = None) -> Any

Return one legacy-style value from the typed context.

源代码位于： jianmu/guard/context.py

def get(self, key: str, default: Any = None) -> Any:
    """Return one legacy-style value from the typed context."""
    value = self._resolve_legacy_key(key, missing=default)
    return default if value is _MISSING else value

to_legacy_context ¶

to_legacy_context() -> dict[str, Any]

Materialize the typed context into one legacy mutable mapping.

源代码位于： jianmu/guard/context.py

def to_legacy_context(self) -> dict[str, Any]:
    """Materialize the typed context into one legacy mutable mapping."""
    payload: dict[str, Any] = copy.deepcopy(dict(self.metadata or {}))
    payload["usage"] = copy.deepcopy(dict(self.usage or {}))
    if self.tool is not None:
        payload["tool_name"] = self.tool.tool_name
        payload["args"] = copy.deepcopy(dict(self.tool.args or {}))
        payload["thread_id"] = self.tool.thread_id
        payload["node_locator"] = self.tool.node_locator
        payload["_tool_call_id"] = self.tool.tool_call_id
        payload["command_text"] = self.tool.command_text
        payload["command_prefix"] = self.tool.command_prefix
        payload["target_path"] = self.tool.target_path
        payload["resource_kind"] = self.tool.resource_kind
        payload["risk_tags"] = copy.deepcopy(list(self.tool.risk_tags or []))
        payload["matched_rule"] = copy.deepcopy(dict(self.tool.matched_rule or {})) if self.tool.matched_rule else None
        payload["permission"] = self.tool.permission
    if self.approval is not None:
        payload["request_id"] = self.approval.request_id
        payload["internal_request_id"] = self.approval.request_id
        payload["_approval_request_id"] = self.approval.effect.request_id or self.approval.request_id
        payload["_needs_confirm"] = self.approval.needs_confirm
        payload["_confirm_reason"] = self.approval.reason
        payload["_approval_progress_state"] = (
            self.approval.effect.progress_state or self.approval.progress_state
        )
        payload["_pop_resolved_approval"] = self.approval.pop_resolved
        payload["_approval_result"] = self.approval.effect.approval_result
        payload["_approval_pending"] = self.approval.effect.pending
        payload["_clear_approval_progress"] = self.approval.effect.clear_progress
    return payload

审批流¶

ApprovalManager ¶

ApprovalManager(
    callback: Optional[
        Callable[
            [ApprovalRequest],
            Awaitable[ApprovalResult | bool | str]
            | ApprovalResult
            | bool
            | str,
        ]
    ] = None,
    *,
    tool_callback: Optional[
        Callable[
            [str, dict | None],
            Awaitable[ApprovalResult | bool | str]
            | ApprovalResult
            | bool
            | str,
        ]
    ] = None,
    node_callback: Optional[
        Callable[
            [str, str, dict | None],
            Awaitable[ApprovalResult | bool | str]
            | ApprovalResult
            | bool
            | str,
        ]
    ] = None,
    audit_hook: Optional[
        Callable[[ApprovalRequest, bool], None]
    ] = None,
    runtime_event_bus: RuntimeEventBus | None = None,
    pending_store: ApprovalStoreProtocol | None = None,
    non_tty_pending_enabled: bool = False,
    permission_grant_store: PermissionGrantStoreProtocol
    | None = None,
    permission_scope_key: str = "",
)

Unified approval entry point.

Resolution order: 1) explicit callback passed to manager 2) tool/node specific fallback callback 3) built-in defaults from guard.approval.defaults

Configure generic, tool, and node-level approval callbacks.

参数：

名称	类型	描述	默认
`callback`	`Optional[Callable[[ApprovalRequest], Awaitable[ApprovalResult \| bool \| str] \| ApprovalResult \| bool \| str]]`	Primary callback for normalized approval requests.	`None`
`tool_callback`	`Optional[Callable[[str, dict \| None], Awaitable[ApprovalResult \| bool \| str] \| ApprovalResult \| bool \| str]]`	Tool-specific fallback callback.	`None`
`node_callback`	`Optional[Callable[[str, str, dict \| None], Awaitable[ApprovalResult \| bool \| str] \| ApprovalResult \| bool \| str]]`	Node-specific fallback callback.	`None`
`audit_hook`	`Optional[Callable[[ApprovalRequest, bool], None]]`	Optional observer invoked after final resolution.	`None`
`runtime_event_bus`	`RuntimeEventBus \| None`	Optional bus for approval lifecycle events.	`None`
`pending_store`	`ApprovalStoreProtocol \| None`	Durable store for request and result records.	`None`
`non_tty_pending_enabled`	`bool`	Whether non-interactive defaults may return `"pending"` instead of auto-denying.	`False`

源代码位于： jianmu/guard/approval/manager.py

def __init__(
    self,
    callback: Optional[Callable[[ApprovalRequest], Awaitable[ApprovalResult | bool | str] | ApprovalResult | bool | str]] = None,
    *,
    tool_callback: Optional[Callable[[str, dict | None], Awaitable[ApprovalResult | bool | str] | ApprovalResult | bool | str]] = None,
    node_callback: Optional[Callable[[str, str, dict | None], Awaitable[ApprovalResult | bool | str] | ApprovalResult | bool | str]] = None,
    audit_hook: Optional[Callable[[ApprovalRequest, bool], None]] = None,
    runtime_event_bus: RuntimeEventBus | None = None,
    pending_store: ApprovalStoreProtocol | None = None,
    non_tty_pending_enabled: bool = False,
    permission_grant_store: PermissionGrantStoreProtocol | None = None,
    permission_scope_key: str = "",
):
    """Configure generic, tool, and node-level approval callbacks.

    Args:
        callback: Primary callback for normalized approval requests.
        tool_callback: Tool-specific fallback callback.
        node_callback: Node-specific fallback callback.
        audit_hook: Optional observer invoked after final resolution.
        runtime_event_bus: Optional bus for approval lifecycle events.
        pending_store: Durable store for request and result records.
        non_tty_pending_enabled: Whether non-interactive defaults may
            return `"pending"` instead of auto-denying.
    """
    self.callback = callback
    self.tool_callback = tool_callback
    self.node_callback = node_callback
    self.audit_hook = audit_hook
    self.runtime_event_bus = runtime_event_bus
    self.pending_store = pending_store or InMemoryApprovalStore()
    self.non_tty_pending_enabled = bool(non_tty_pending_enabled)
    self.permission_grant_store = permission_grant_store
    self.permission_scope_key = str(permission_scope_key or "")
    self._resolution_subscribers: dict[str, list[Callable[[ApprovalResult], None]]] = {}
    self._resolution_lock = threading.Lock()

request `async` ¶

request(req: ApprovalRequest) -> ApprovalRecord

Run the configured approval callback for a generic request.

参数：

名称	类型	描述	默认
`req`	`ApprovalRequest`	The ApprovalRequest instance to process.	必需

返回：

类型	描述
`ApprovalRecord`	Durable approval record reflecting the current lifecycle state.

引发：

类型	描述
`ValueError`	If `req.request_id` is empty.

源代码位于： jianmu/guard/approval/manager.py

async def request(self, req: ApprovalRequest) -> ApprovalRecord:
    """Run the configured approval callback for a generic request.

    Args:
        req: The ApprovalRequest instance to process.

    Returns:
        Durable approval record reflecting the current lifecycle state.

    Raises:
        ValueError: If `req.request_id` is empty.
    """
    if not str(req.request_id or "").strip():
        raise ValueError("ApprovalRequest.request_id must be set before calling ApprovalManager.request().")
    created_at = time.time()
    self.pending_store.upsert_record(
        ApprovalRecord(
            request_id=req.request_id,
            status=ApprovalStatus.REQUESTED,
            request=ApprovalRequest(
                kind=req.kind,
                target=req.target,
                request_id=req.request_id,
                description=req.description,
                context=dict(req.context or {}),
                permission=req.permission,
            ),
            result=None,
            created_at=created_at,
            updated_at=created_at,
        )
    )
    self._emit_runtime_event("approval.requested", req=req, status=ApprovalStatus.REQUESTED)
    try:
        if self.callback is not None:
            result = self.callback(req)
        elif req.kind == "tool":
            cb = self.tool_callback or default_confirm_callback
            if self._supports_non_tty_pending(cb):
                result = cb(
                    req.target,
                    req.context.get("args"),
                    non_tty_pending=self.non_tty_pending_enabled,
                )
            else:
                result = cb(
                    req.target,
                    req.context.get("args"),
                )
        elif req.kind == "node":
            cb = self.node_callback or default_node_approval_callback
            if self._supports_non_tty_pending(cb):
                result = cb(
                    req.target,
                    req.description,
                    req.context,
                    non_tty_pending=self.non_tty_pending_enabled,
                )
            else:
                result = cb(
                    req.target,
                    req.description,
                    req.context,
                )
        else:
            logger.warning("ApprovalManager received unsupported request kind '{}'", req.kind)
            normalized = ApprovalResult(
                approved=False,
                request_id=req.request_id,
                reason=req.description,
            )
            self._audit(req, normalized.approved)
            record = ApprovalRecord(
                request_id=req.request_id,
                status=ApprovalStatus.REJECTED,
                request=req,
                result=normalized,
                created_at=(self.pending_store.get_record(req.request_id) or ApprovalRecord(
                    request_id=req.request_id,
                    status=ApprovalStatus.REQUESTED,
                    request=req,
                )).created_at,
                updated_at=time.time(),
            )
            self.pending_store.upsert_record(record)
            self._emit_runtime_event("approval.resolved", req=req, status=record.status, result=record.result)
            return record

        if inspect.isawaitable(result):
            result = await result
        resolved_status, normalized = self._normalize_result(result, request_id=req.request_id, reason=req.description)
    except Exception as exc:
        logger.warning("ApprovalManager request failed for {} '{}': {}", req.kind, req.target, exc)
        resolved_status = ApprovalStatus.REJECTED
        normalized = ApprovalResult(
            approved=False,
            request_id=req.request_id,
            reason=req.description or str(exc),
        )

    if resolved_status == ApprovalStatus.PENDING:
        self.pending_store.put_request(req)
        record = self.pending_store.get_record(req.request_id) or ApprovalRecord(
            request_id=req.request_id,
            status=ApprovalStatus.PENDING,
            request=req,
            created_at=created_at,
            updated_at=time.time(),
        )
    else:
        record = ApprovalRecord(
            request_id=req.request_id,
            status=resolved_status,
            request=ApprovalRequest(
                kind=req.kind,
                target=req.target,
                request_id=req.request_id,
                description=req.description,
                context=dict(req.context or {}),
                permission=req.permission,
            ),
            result=normalized,
            created_at=(self.pending_store.get_record(req.request_id) or ApprovalRecord(
                request_id=req.request_id,
                status=ApprovalStatus.REQUESTED,
                request=req,
            )).created_at,
            updated_at=time.time(),
        )
        self.pending_store.upsert_record(record)
    self._emit_runtime_event("approval.resolved", req=req, status=record.status, result=record.result)
    if record.result is not None:
        self._audit(req, record.result.approved)
    return record

confirm_tool `async` ¶

confirm_tool(
    tool_name: str,
    params: dict | None = None,
    *,
    reason: str = "",
    context: dict[str, Any] | None = None,
    request_id: str = "",
) -> ApprovalRecord

Request approval for a tool invocation.

参数：

名称	类型	描述	默认
`tool_name`	`str`	Name of the tool.	必需
`params`	`dict \| None`	Optional dictionary of arguments passed to the tool.	`None`
`reason`	`str`	Explanation for the approval request.	`''`
`context`	`dict[str, Any] \| None`	Context dictionary.	`None`

返回：

类型	描述
`ApprovalRecord`	Durable approval record for the tool request.

引发：

类型	描述
`ValueError`	If durable tool approval is requested without a precomputed `request_id`.

源代码位于： jianmu/guard/approval/manager.py

async def confirm_tool(
    self,
    tool_name: str,
    params: dict | None = None,
    *,
    reason: str = "",
    context: dict[str, Any] | None = None,
    request_id: str = "",
) -> ApprovalRecord:
    """Request approval for a tool invocation.

    Args:
        tool_name: Name of the tool.
        params: Optional dictionary of arguments passed to the tool.
        reason: Explanation for the approval request.
        context: Context dictionary.

    Returns:
        Durable approval record for the tool request.

    Raises:
        ValueError: If durable tool approval is requested without a
            precomputed `request_id`.
    """
    payload = dict(context or {})
    payload.setdefault("args", params or {})
    if reason:
        payload.setdefault("reason", reason)
    internal_request_id = str(request_id or "").strip()
    if not internal_request_id:
        if payload.get("durable_tool_approval"):
            raise ValueError(
                "Durable tool approval requests must provide a precomputed request_id from tool_context."
            )
        internal_request_id = tool_request_id(
            thread_id=str(payload.get("thread_id") or "default_thread"),
            node_locator=str(payload.get("node_locator") or ""),
            tool_name=str(payload.get("_approval_target") or tool_name),
            tool_call_id=str(payload.get("_tool_call_id") or ""),
        )
    payload.setdefault("internal_request_id", internal_request_id)
    raw_permission = payload.get("permission")
    permission = (
        raw_permission
        if isinstance(raw_permission, PermissionCheck)
        else permission_check_from_dict(raw_permission if isinstance(raw_permission, dict) else None)
    )
    req = ApprovalRequest(
        kind="tool",
        target=tool_name,
        request_id=request_id,
        description=reason,
        context=payload,
        permission=permission,
    )
    req.request_id = internal_request_id
    return await self.request(req)

approve_node `async` ¶

approve_node(
    node_name: str,
    description: str = "",
    context: dict[str, Any] | None = None,
    request_id: str = "",
) -> ApprovalRecord

Request approval for a node-level action.

参数：

名称	类型	描述	默认
`node_name`	`str`	Name of the node.	必需
`description`	`str`	Description of the node action.	`''`
`context`	`dict[str, Any] \| None`	Context dictionary.	`None`

返回：

类型	描述
`ApprovalRecord`	Durable approval record for the node request.

源代码位于： jianmu/guard/approval/manager.py

async def approve_node(
    self,
    node_name: str,
    description: str = "",
    context: dict[str, Any] | None = None,
    request_id: str = "",
) -> ApprovalRecord:
    """Request approval for a node-level action.

    Args:
        node_name: Name of the node.
        description: Description of the node action.
        context: Context dictionary.

    Returns:
        Durable approval record for the node request.
    """
    payload = dict(context or {})
    internal_request_id = str(request_id or "").strip()
    if not internal_request_id:
        internal_request_id = approval_request_id(
            thread_id=str(payload.get("thread_id") or "default_thread"),
            approval_kind=ApprovalKind.NODE.value,
            approval_phase=str(payload.get("approval_phase") or payload.get("phase") or ApprovalPhase.BEFORE.value),
            node_locator=str(payload.get("node_locator") or ""),
            approval_target=str(payload.get("approval_target") or node_name),
        )
        payload.setdefault("internal_request_id", internal_request_id)
    req = ApprovalRequest(kind="node", target=node_name, request_id=internal_request_id, description=description, context=payload)
    return await self.request(req)

resolve_request ¶

resolve_request(
    request_id: str,
    approved: bool,
    *,
    reason: str = "",
    payload: dict[str, Any] | None = None,
) -> ApprovalResult | None

Resolve one pending request stored by the manager.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for request.	必需
`approved`	`bool`	The `approved` value.	必需
`reason`	`str`	The `reason` value.	`''`
`payload`	`dict[str, Any] \| None`	Payload data for the operation.	`None`

返回：

类型	描述
`ApprovalResult \| None`	The resolved value, or `None` when no value is available.

源代码位于： jianmu/guard/approval/manager.py

def resolve_request(
    self,
    request_id: str,
    approved: bool,
    *,
    reason: str = "",
    payload: dict[str, Any] | None = None,
) -> ApprovalResult | None:
    """Resolve one pending request stored by the manager.

    Args:
        request_id: Identifier for request.
        approved: The `approved` value.
        reason: The `reason` value.
        payload: Payload data for the operation.

    Returns:
        The resolved value, or `None` when no value is available.
    """
    resolved = self.pending_store.resolve_request(request_id, approved, reason=reason, payload=payload)
    if resolved is None:
        return None
    req, result = resolved
    self._emit_runtime_event(
        "approval.resolved",
        req=req,
        status=ApprovalStatus.APPROVED if result.approved else ApprovalStatus.REJECTED,
        result=result,
    )
    self._notify_resolution_subscribers(request_id, result)
    return result

reply_request ¶

reply_request(
    request_id: str,
    reply: PermissionReply | str,
    *,
    reason: str = "",
    feedback: str = "",
    scope_key: str = "",
) -> ApprovalResult | None

Resolve a pending permission request as once, always, or reject.

源代码位于： jianmu/guard/approval/manager.py

def reply_request(
    self,
    request_id: str,
    reply: PermissionReply | str,
    *,
    reason: str = "",
    feedback: str = "",
    scope_key: str = "",
) -> ApprovalResult | None:
    """Resolve a pending permission request as once, always, or reject."""
    normalized = reply if isinstance(reply, PermissionReply) else PermissionReply(str(reply or "").strip().lower())
    request = self.get_pending_request(request_id)
    if request is None:
        return None
    saved_rule_ids: list[str] = []
    if normalized == PermissionReply.ALWAYS:
        store = self.permission_grant_store
        resolved_scope = str(scope_key or self.permission_scope_key or "")
        if store is None:
            raise RuntimeError("always permission reply requires a PermissionGrantStore")
        if not resolved_scope:
            raise RuntimeError("always permission reply requires a non-empty scope_key")
        permission = request.permission
        if permission is None or not permission.save:
            raise ValueError("approval request does not provide saveable permission resources")
        rules = [
            PermissionRule(
                action=permission.action,
                resource=pattern,
                effect=PermissionEffect.ALLOW,
                description=f"Saved from approval {request_id}",
                source="saved",
            )
            for pattern in permission.save
        ]
        saved_rule_ids = store.save_rules(
            resolved_scope,
            rules,
            source_request_id=request_id,
        )
    approved = normalized != PermissionReply.REJECT
    result = self.resolve_request(
        request_id,
        approved,
        reason=reason or feedback or normalized.value,
        payload={
            "approved": approved,
            "reply": normalized.value,
            "feedback": feedback,
            "saved_rule_ids": list(saved_rule_ids),
        },
    )
    if normalized == PermissionReply.ALWAYS:
        self._resolve_pending_satisfied_by_saved(scope_key=str(scope_key or self.permission_scope_key or ""))
    return result

get_pending_request ¶

get_pending_request(
    request_id: str,
) -> ApprovalRequest | None

Return one pending request stored by the manager.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalRequest \| None`	The pending approval request, or `None` when not found.

源代码位于： jianmu/guard/approval/manager.py

def get_pending_request(self, request_id: str) -> ApprovalRequest | None:
    """Return one pending request stored by the manager.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The pending approval request, or `None` when not found.
    """
    return self.pending_store.get_request(request_id)

get_result ¶

get_result(request_id: str) -> ApprovalResult | None

Return one resolved result stored by the manager without consuming it.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalResult \| None`	The resolved approval result, or `None` when unavailable.

源代码位于： jianmu/guard/approval/manager.py

def get_result(self, request_id: str) -> ApprovalResult | None:
    """Return one resolved result stored by the manager without consuming it.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The resolved approval result, or `None` when unavailable.
    """
    return self.pending_store.get_result(request_id)

get_record ¶

get_record(request_id: str) -> ApprovalRecord | None

Return one durable approval record by request id.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalRecord \| None`	The durable approval record, or `None` when not found.

源代码位于： jianmu/guard/approval/manager.py

def get_record(self, request_id: str) -> ApprovalRecord | None:
    """Return one durable approval record by request id.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The durable approval record, or `None` when not found.
    """
    return self.pending_store.get_record(request_id)

ack_result ¶

ack_result(request_id: str) -> ApprovalResult | None

Mark one resolved result as consumed but not yet pruned.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalResult \| None`	The acknowledged result, or `None` when no result exists.

源代码位于： jianmu/guard/approval/manager.py

def ack_result(self, request_id: str) -> ApprovalResult | None:
    """Mark one resolved result as consumed but not yet pruned.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The acknowledged result, or `None` when no result exists.
    """
    return self.pending_store.ack_result(request_id)

prune_result ¶

prune_result(request_id: str) -> ApprovalResult | None

Delete one resolved result after checkpoint durability is established.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalResult \| None`	The pruned result, or `None` when no result exists.

源代码位于： jianmu/guard/approval/manager.py

def prune_result(self, request_id: str) -> ApprovalResult | None:
    """Delete one resolved result after checkpoint durability is established.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The pruned result, or `None` when no result exists.
    """
    return self.pending_store.prune_result(request_id)

drain_result ¶

drain_result(request_id: str) -> ApprovalResult | None

Backward-compatible alias that ack/prunes one resolved result.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需

返回：

类型	描述
`ApprovalResult \| None`	The drained result, or `None` when no result exists.

源代码位于： jianmu/guard/approval/manager.py

def drain_result(self, request_id: str) -> ApprovalResult | None:
    """Backward-compatible alias that ack/prunes one resolved result.

    Args:
        request_id: Identifier for the approval request.

    Returns:
        The drained result, or `None` when no result exists.
    """
    result = self.pending_store.ack_result(request_id)
    if result is None:
        return None
    self.pending_store.prune_result(request_id)
    return result

subscribe_resolution ¶

subscribe_resolution(
    request_id: str,
    callback: Callable[[ApprovalResult], None],
) -> Callable[[], None]

Subscribe one callback to one internal approval request id.

The subscription is race-safe for resolve-before-subscribe: if the result already exists, the callback is delivered immediately and the returned unsubscribe is a no-op.

参数：

名称	类型	描述	默认
`request_id`	`str`	Identifier for the approval request.	必需
`callback`	`Callable[[ApprovalResult], None]`	Listener invoked with the final approval result.	必需

返回：

类型	描述
`Callable[[], None]`	A callable that removes the subscription when invoked.

源代码位于： jianmu/guard/approval/manager.py

def subscribe_resolution(
    self,
    request_id: str,
    callback: Callable[[ApprovalResult], None],
) -> Callable[[], None]:
    """Subscribe one callback to one internal approval request id.

    The subscription is race-safe for resolve-before-subscribe: if the
    result already exists, the callback is delivered immediately and the
    returned unsubscribe is a no-op.

    Args:
        request_id: Identifier for the approval request.
        callback: Listener invoked with the final approval result.

    Returns:
        A callable that removes the subscription when invoked.
    """
    cleaned_request_id = str(request_id or "").strip()
    if not cleaned_request_id:
        def _noop() -> None:
            return
        return _noop
    with self._resolution_lock:
        existing = self.pending_store.get_result(cleaned_request_id)
        if existing is not None:
            callback(existing)
            def _noop() -> None:
                return
            return _noop
        self._resolution_subscribers.setdefault(cleaned_request_id, []).append(callback)

    def _unsubscribe() -> None:
        with self._resolution_lock:
            subscribers = self._resolution_subscribers.get(cleaned_request_id)
            if not subscribers:
                return
            try:
                subscribers.remove(callback)
            except ValueError:
                return
            if not subscribers:
                self._resolution_subscribers.pop(cleaned_request_id, None)

    return _unsubscribe

ApprovalRequest `dataclass` ¶

ApprovalRequest(
    kind: str,
    target: str,
    request_id: str = "",
    description: str = "",
    context: dict[str, Any] = dict(),
    permission: PermissionCheck | None = None,
)

Normalized approval request shared by node/tool HITL flows.

属性：

名称	类型	描述
`kind`	`str`	Approval category such as `tool` or `node`.
`target`	`str`	Unique target identifier, usually a tool or node name.
`request_id`	`str`	Framework-generated internal request id used for suspend, store, resume, and listener correlation.
`description`	`str`	Human-readable summary shown to approvers.
`context`	`dict[str, Any]`	Extra structured payload needed by the approval UI or policy.

ApprovalResult `dataclass` ¶

ApprovalResult(
    approved: bool,
    request_id: str = "",
    reason: str = "",
    payload: dict[str, Any] = dict(),
)

Normalized final approval resolution.

Approval ¶

Approval(
    *,
    child: Behaviour,
    description: str = "",
    phase: str = "before",
    callback: Optional[
        Callable[[ApprovalRequest], Awaitable[bool] | bool]
    ] = None,
    context_fn: Optional[Callable[[], dict | None]] = None,
    on_reject: str = "failure",
    name: Optional[str] = None,
)

Bases: Decorator

Node-level human approval gate.

phase

"before": approve before child executes
"after": approve after child finishes
"both": approve before and after

Configure a human-approval gate around a child behaviour.

源代码位于： jianmu/guard/approval/decorator.py

def __init__(
    self,
    *,
    child: behaviour.Behaviour,
    description: str = "",
    phase: str = "before",
    callback: Optional[Callable[[ApprovalRequest], Awaitable[bool] | bool]] = None,
    context_fn: Optional[Callable[[], dict | None]] = None,
    on_reject: str = "failure",
    name: Optional[str] = None,
):
    """Configure a human-approval gate around a child behaviour."""
    gate_name = name or f"Approval({child.name})"
    super().__init__(name=gate_name, child=child)
    if phase not in (ApprovalPhase.BEFORE.value, ApprovalPhase.AFTER.value, "both"):
        raise ValueError(f"Unsupported phase: {phase}")
    if on_reject not in ("failure", "success_skip"):
        raise ValueError(f"Unsupported on_reject: {on_reject}")

    self.description = description
    self.phase = phase
    self.callback = callback
    self.context_fn = context_fn
    self.on_reject = on_reject

    self._default_manager: Optional[ApprovalManager] = None
    self._callback_manager: Optional[ApprovalManager] = None

    self._pending_phase: str | None = None
    self._pending_task: asyncio.Task | None = None
    self._last_child_status: Status = Status.INVALID
    self._pre_approved: bool = False
    self._post_approved: bool = False
    self._child_completed: bool = False

initialise ¶

initialise() -> None

Reset decorator-local approval state before a fresh tick cycle.

源代码位于： jianmu/guard/approval/decorator.py

def initialise(self) -> None:
    """Reset decorator-local approval state before a fresh tick cycle."""
    self._pending_phase = None
    self._pending_task = None
    self._last_child_status = Status.INVALID
    self._pre_approved = False
    self._post_approved = False
    self._child_completed = False

terminate ¶

terminate(new_status: Status) -> None

Cancel any pending approval task when the decorator stops.

参数：

名称	类型	描述	默认
`new_status`	`Status`	The execution status the node is transitioning to.	必需

源代码位于： jianmu/guard/approval/decorator.py

def terminate(self, new_status: Status) -> None:
    """Cancel any pending approval task when the decorator stops.

    Args:
        new_status: The execution status the node is transitioning to.
    """
    if new_status != Status.RUNNING:
        if self._pending_task and not self._pending_task.done():
            self._pending_task.cancel()
        self._pending_task = None
        self._pending_phase = None

update ¶

update() -> Status

Return the decorated child status for py_trees compatibility.

返回：

类型	描述
`Status`	The execution Status of the decorated child node.

源代码位于： jianmu/guard/approval/decorator.py

def update(self) -> Status:
    """Return the decorated child status for py_trees compatibility.

    Returns:
        The execution Status of the decorated child node.
    """
    # Not used because tick() is overridden, but required by Behaviour.
    return self.decorated.status

tick ¶

tick()

Drive pre/post approval phases around the decorated child tick.

源代码位于： jianmu/guard/approval/decorator.py

def tick(self):
    """Drive pre/post approval phases around the decorated child tick."""
    self.logger.debug(f"{self.__class__.__name__}.tick()")
    # Re-entering after SUCCESS/FAILURE should always start a fresh approval cycle.
    if self.status != Status.RUNNING:
        self.initialise()
    self._sync_progress_cache()

    if self._needs_pre() and not self._pre_approved:
        status = self._consume_approval(ApprovalPhase.BEFORE.value)
        if status is not None:
            if status != Status.RUNNING:
                self._clear_progress()
            self.stop(status)
            self.status = status
            yield self
            return

    if not self._child_completed:
        for node in self.decorated.tick():
            yield node
        self._last_child_status = self.decorated.status

        if self.decorated.status == Status.RUNNING:
            self._update_progress(child_completed=False, last_child_status=self._last_child_status)
            self.status = Status.RUNNING
            yield self
            return
        self._child_completed = True
        self._update_progress(child_completed=True, last_child_status=self._last_child_status)

    if self._needs_post() and not self._post_approved:
        status = self._consume_approval(ApprovalPhase.AFTER.value)
        if status is not None:
            if status != Status.RUNNING:
                self._clear_progress()
                self.stop(status)
            self.status = status
            yield self
            return

    new_status = self.decorated.status
    if new_status != Status.RUNNING:
        self._clear_progress()
        self.stop(new_status)
    self.status = new_status
    yield self

内置 Checker¶

ToolPolicyChecker ¶

ToolPolicyChecker(
    policy: dict[str, str],
    default: str = "allow",
    aliases: dict[str, str] | None = None,
    tool_rules: list[ToolRule] | None = None,
)

Bases: CheckerProtocol

Policy-based tool checker.

policy: mapping tool_name -> "allow" | "deny" | "confirm" default: fallback policy aliases: optional mapping input_name -> canonical name

Normalize tool policy rules and alias mappings.

源代码位于： jianmu/guard/checkers/tool_policy.py

def __init__(
    self,
    policy: dict[str, str],
    default: str = "allow",
    aliases: dict[str, str] | None = None,
    tool_rules: list[ToolRule] | None = None,
):
    """Normalize tool policy rules and alias mappings."""
    self.policy = {k: v.lower() for k, v in policy.items()}
    self.default = (default or "allow").lower()
    self.aliases = aliases or {}
    self.tool_rules = list(tool_rules or [])

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Apply allow/deny/confirm policy rules to a tool action.

参数：

名称	类型	描述	默认
`action`	`str`	The action string to check.	必需
`context`	`CheckContextLike`	Context dictionary representing the execution environment.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple of Decision and reason string.

源代码位于： jianmu/guard/checkers/tool_policy.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Apply allow/deny/confirm policy rules to a tool action.

    Args:
        action: The action string to check.
        context: Context dictionary representing the execution environment.

    Returns:
        Tuple of Decision and reason string.
    """
    if action != "tool_call":
        return Decision.ALLOW, ""

    name = _tool_name(context)
    canonical = self.aliases.get(name, name)

    matched = _match_tool_rule(context, name, canonical, self.tool_rules)
    if matched is not None:
        rule, reason = matched
        _mark_matched_rule(context, rule)
        if rule.decision == "deny":
            return Decision.DENY, reason
        if rule.decision == "confirm":
            _mark_needs_confirm(context, reason)
            return Decision.CONFIRM, reason
        return Decision.ALLOW, reason

    for candidate in (name, canonical):
        pol = self.policy.get(candidate, self.default)
        if pol == "deny":
            return Decision.DENY, f"Tool '{candidate}' is denied by policy"
        if pol == "confirm":
            # mark for confirmation, let later checker handle
            reason = f"Tool '{candidate}' requires confirmation"
            _mark_needs_confirm(context, reason)
            return Decision.CONFIRM, reason

    return Decision.ALLOW, ""

reset ¶

reset() -> None

Reset tool policy checker state.

源代码位于： jianmu/guard/checkers/tool_policy.py

def reset(self) -> None:
    """Reset tool policy checker state."""
    pass

RateLimitChecker ¶

RateLimitChecker(max_calls: int)

Bases: CheckerProtocol

Limit tool_call action count.

Configure the maximum number of permitted tool calls.

源代码位于： jianmu/guard/checkers/rate_limit.py

def __init__(self, max_calls: int):
    """Configure the maximum number of permitted tool calls."""
    self.max_calls = max_calls
    self.call_count = 0

check `async` ¶

check(
    action: str, context: dict[str, Any]
) -> Tuple[Decision, str]

Enforce a simple per-run tool call count limit.

参数：

名称	类型	描述	默认
`action`	`str`	The action string to check.	必需
`context`	`dict[str, Any]`	Context dictionary.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple of Decision and reason string.

源代码位于： jianmu/guard/checkers/rate_limit.py

async def check(self, action: str, context: dict[str, Any]) -> Tuple[Decision, str]:
    """Enforce a simple per-run tool call count limit.

    Args:
        action: The action string to check.
        context: Context dictionary.

    Returns:
        Tuple of Decision and reason string.
    """
    if action != "tool_call":
        return Decision.ALLOW, ""
    if self.max_calls is None:
        return Decision.ALLOW, ""
    external_count = context.get("_tool_task_count", None)
    if external_count is not None:
        try:
            current_count = max(0, int(external_count))
        except Exception:
            current_count = self.call_count
        if current_count >= self.max_calls:
            return Decision.DENY, f"Max tool calls ({self.max_calls}) exceeded"
        return Decision.ALLOW, ""
    if self.call_count >= self.max_calls:
        return Decision.DENY, f"Max tool calls ({self.max_calls}) exceeded"
    self.call_count += 1
    return Decision.ALLOW, ""

reset ¶

reset() -> None

Clear accumulated call count.

源代码位于： jianmu/guard/checkers/rate_limit.py

def reset(self) -> None:
    """Clear accumulated call count."""
    self.call_count = 0

dump_resume_state ¶

dump_resume_state() -> dict[str, int]

Return checker-local state for checkpoint resume.

源代码位于： jianmu/guard/checkers/rate_limit.py

def dump_resume_state(self) -> dict[str, int]:
    """Return checker-local state for checkpoint resume."""
    return {"call_count": int(self.call_count)}

restore_resume_state ¶

restore_resume_state(data: dict[str, Any]) -> None

Restore checker-local state from checkpoint metadata.

源代码位于： jianmu/guard/checkers/rate_limit.py

def restore_resume_state(self, data: dict[str, Any]) -> None:
    """Restore checker-local state from checkpoint metadata."""
    try:
        self.call_count = max(0, int(dict(data or {}).get("call_count") or 0))
    except Exception:
        self.call_count = 0

BudgetChecker ¶

BudgetChecker(budget: BudgetConstraints)

Bases: CheckerProtocol

Placeholder budget checker.

Expects context to carry usage dict {'prompt_tokens': int, 'completion_tokens': int, 'total_tokens': int}.

Bind the checker to a fixed token budget specification.

源代码位于： jianmu/guard/checkers/budget.py

def __init__(self, budget: BudgetConstraints):
    """Bind the checker to a fixed token budget specification."""
    self.budget = budget

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Reject requests that would exceed the configured token budget.

参数：

名称	类型	描述	默认
`action`	`str`	The action string to check.	必需
`context`	`CheckContextLike`	Context dictionary containing usage details.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple of Decision and reason string.

源代码位于： jianmu/guard/checkers/budget.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Reject requests that would exceed the configured token budget.

    Args:
        action: The action string to check.
        context: Context dictionary containing usage details.

    Returns:
        Tuple of Decision and reason string.
    """
    if action not in ("llm_call", "tool_call"):
        return Decision.ALLOW, ""
    usage = _usage(context)
    for key in ("total_tokens", "prompt_tokens", "completion_tokens"):
        limit = getattr(self.budget, key, None)
        if limit is None:
            continue
        used = int(usage.get(key, 0))
        if used > limit:
            return Decision.DENY, f"Budget exceeded for {key}: {used}>{limit}"
    return Decision.ALLOW, ""

reset ¶

reset() -> None

Reset budget checker state.

源代码位于： jianmu/guard/checkers/budget.py

def reset(self) -> None:
    """Reset budget checker state."""
    pass

ConfirmChecker ¶

ConfirmChecker(approval_manager: ApprovalManager)

Bases: CheckerProtocol

Perform user confirmation when context marks _needs_confirm.

Bind the checker to an approval manager.

源代码位于： jianmu/guard/checkers/confirm.py

def __init__(self, approval_manager: ApprovalManager):
    """Bind the checker to an approval manager."""
    self.approval_manager = approval_manager

check `async` ¶

check(
    action: str, context: CheckContextLike
) -> Tuple[Decision, str]

Reject risky actions until explicit confirmation is provided.

参数：

名称	类型	描述	默认
`action`	`str`	The action string to check.	必需
`context`	`CheckContextLike`	Context dictionary representing the execution environment.	必需

返回：

类型	描述
`Tuple[Decision, str]`	Tuple of Decision and reason string.

源代码位于： jianmu/guard/checkers/confirm.py

async def check(self, action: str, context: CheckContextLike) -> Tuple[Decision, str]:
    """Reject risky actions until explicit confirmation is provided.

    Args:
        action: The action string to check.
        context: Context dictionary representing the execution environment.

    Returns:
        Tuple of Decision and reason string.
    """
    if isinstance(context, GuardCheckContext):
        return await self._check_typed(action, context)
    return await self._check_legacy(action, context)

reset ¶

reset() -> None

Reset confirmation checker state.

源代码位于： jianmu/guard/checkers/confirm.py

def reset(self) -> None:
    """Reset confirmation checker state."""
    pass

jianmu.guard¶

1. 模块职责¶

2. 适合查什么¶

3. 使用建议¶

4. 最小示例¶

5. 常见入口¶

6. API 参考¶

决策与约束¶

Decision ¶

BudgetConstraints dataclass ¶

GuardConstraints dataclass ¶

from_dict classmethod ¶

守卫入口¶

GuardEnforcer ¶

check async ¶

reset ¶

dump_resume_state ¶

restore_resume_state ¶

from_guard_constraints classmethod ¶

CheckerProtocol ¶

check async ¶

reset ¶

GuardProtocol ¶

check async ¶

reset ¶

检查上下文¶

ToolCheckContext dataclass ¶

ApprovalCheckEffect dataclass ¶

ApprovalCheckContext dataclass ¶

GuardCheckContext dataclass ¶

get ¶

to_legacy_context ¶

审批流¶

ApprovalManager ¶

request async ¶

confirm_tool async ¶

approve_node async ¶

resolve_request ¶

reply_request ¶

get_pending_request ¶

get_result ¶

get_record ¶

ack_result ¶

prune_result ¶

drain_result ¶

subscribe_resolution ¶

ApprovalRequest dataclass ¶

ApprovalResult dataclass ¶

Approval ¶

initialise ¶

terminate ¶

update ¶

tick ¶

内置 Checker¶

ToolPolicyChecker ¶

check async ¶

reset ¶

RateLimitChecker ¶

check async ¶

reset ¶

dump_resume_state ¶

restore_resume_state ¶

BudgetChecker ¶

check async ¶

reset ¶

ConfirmChecker ¶

check async ¶

reset ¶

`jianmu.guard`¶

BudgetConstraints `dataclass` ¶

GuardConstraints `dataclass` ¶

from_dict `classmethod` ¶

check `async` ¶

from_guard_constraints `classmethod` ¶

check `async` ¶

check `async` ¶

ToolCheckContext `dataclass` ¶

ApprovalCheckEffect `dataclass` ¶

ApprovalCheckContext `dataclass` ¶

GuardCheckContext `dataclass` ¶

request `async` ¶

confirm_tool `async` ¶

approve_node `async` ¶

ApprovalRequest `dataclass` ¶

ApprovalResult `dataclass` ¶

check `async` ¶

check `async` ¶

check `async` ¶

check `async` ¶