butler/execution_context.py
Rama main · réplica de lectura del repositorio original.
"""Current Butler execution context shared across tool and delegate paths.
R1-10 [H] layering_violation: this module also exposes the
``is_current_turn_owner`` / ``get_current_turn_bridge`` /
``try_push_current_turn_workflow_failure`` / ``owner_required_message``
helpers — the seam that lets ``butler/tools/`` avoid importing from
``butler.gateway.owner_gate``, ``butler.gateway.outbound_bridge`` and
``butler.gateway.completion_notify``. The seam mirrors the R1-3
``EventsSink`` / R1-9 ``MetricsSink`` Protocol pattern but stays
lightweight (thin shims over the gateway module + duck-typed
orchestrator hook for future per-orchestrator policy).
The owner / bridge lookups follow the same precedence:
1. If an orchestrator is bound via :func:`use_execution_context` and
that orchestrator exposes a matching duck-typed method /
attribute (``is_owner(...)`` / ``gateway_bridge``), use it.
2. Otherwise, fall back to the canonical gateway module
(back-compat path; behaviour unchanged).
"""
from __future__ import annotations
from contextlib import contextmanager
from contextvars import ContextVar
from typing import TYPE_CHECKING, Any, Iterator, cast
from butler.contracts.gateway_registry import get_bridge_access, get_owner_gate
if TYPE_CHECKING:
from butler.orchestrator import ButlerOrchestrator
_current_orchestrator: ContextVar["ButlerOrchestrator | None"] = ContextVar(
"butler_current_orchestrator",
default=None,
)
_current_session_key: ContextVar[str] = ContextVar(
"butler_current_session_key",
default="",
)
_current_workflow_step: ContextVar[str] = ContextVar(
"butler_current_workflow_step",
default="",
)
_workflow_var_pool: ContextVar[object | None] = ContextVar(
"butler_workflow_var_pool",
default=None,
)
_current_loop_role: ContextVar[str] = ContextVar(
"butler_current_loop_role",
default="",
)
def get_current_loop_role() -> str:
"""Gateway/subagent loop role: ``butler`` | ``lead`` | ``plan`` | ``dev`` | …"""
return str(_current_loop_role.get() or "").strip().lower()
def get_current_orchestrator() -> "ButlerOrchestrator | None":
"""Return the orchestrator bound to the current AgentLoop/tool turn."""
return _current_orchestrator.get()
def get_current_session_key() -> str:
"""Return the external session key bound to the current turn, if any."""
return _current_session_key.get()
def get_current_workflow_step() -> str:
"""Active workflow step id during DAG node execution (for step-level permissions)."""
return str(_current_workflow_step.get() or "").strip()
def get_workflow_var_pool() -> object | None:
return _workflow_var_pool.get()
@contextmanager
def use_workflow_var_pool(pool: object | None) -> Iterator[None]:
token = _workflow_var_pool.set(pool)
try:
yield
finally:
_workflow_var_pool.reset(token)
@contextmanager
def use_workflow_step(step_id: str) -> Iterator[None]:
"""Bind workflow step id for tool permission checks in orchestrator nodes."""
token = _current_workflow_step.set(str(step_id or "").strip())
try:
yield
finally:
_current_workflow_step.reset(token)
_session_read_recall_gate: ContextVar[bool] = ContextVar(
"butler_session_read_recall_gate",
default=False,
)
_local_project_inventory_gate: ContextVar[bool] = ContextVar(
"butler_local_project_inventory_gate",
default=False,
)
def is_session_read_recall_gate_active() -> bool:
return bool(_session_read_recall_gate.get())
@contextmanager
def use_session_read_recall_gate(active: bool = True) -> Iterator[None]:
token = _session_read_recall_gate.set(bool(active))
try:
yield
finally:
_session_read_recall_gate.reset(token)
def is_local_project_inventory_gate_active() -> bool:
return bool(_local_project_inventory_gate.get())
@contextmanager
def use_local_project_inventory_gate(active: bool = True) -> Iterator[None]:
token = _local_project_inventory_gate.set(bool(active))
try:
yield
finally:
_local_project_inventory_gate.reset(token)
@contextmanager
def use_loop_role(role: str) -> Iterator[None]:
token = _current_loop_role.set(str(role or "").strip().lower())
try:
yield
finally:
_current_loop_role.reset(token)
@contextmanager
def use_execution_context(
orchestrator: "ButlerOrchestrator | None" = None,
*,
session_key: str = "",
loop_role: str = "",
) -> Iterator[None]:
"""Temporarily bind orchestrator and/or session key for nested tools and delegates."""
orch_token = None
role_token = None
if orchestrator is not None:
orch_token = _current_orchestrator.set(orchestrator)
session_token = _current_session_key.set(session_key)
if loop_role:
role_token = _current_loop_role.set(str(loop_role).strip().lower())
try:
yield
finally:
if role_token is not None:
_current_loop_role.reset(role_token)
_current_session_key.reset(session_token)
if orch_token is not None:
_current_orchestrator.reset(orch_token)
def get_audit_session_key(*, fallback: str = "unscoped") -> str:
"""Session key for tool audit when no external session is bound."""
key = str(get_current_session_key() or "").strip()
if key:
return key
if get_current_orchestrator() is not None:
return ""
return fallback
# ── R1-10 layering seam: owner / bridge / completion helpers ─────────────
#
# Tools call these helpers instead of importing from ``butler.gateway.*``
# directly. The helpers own the duck-typed orchestrator hook + the
# back-compat gateway fallback in one place.
#
# The orchestrator hook is ``__dict__``-checked (not bare ``getattr``)
# so :class:`unittest.mock.MagicMock` orchestrators in unit tests fall
# through to the gateway back-compat path. Real orchestrators that want
# to override must set the attribute explicitly on ``self``.
def _orchestrator_explicit_attr(orch: object, name: str) -> object | None:
"""Return ``orch.<name>`` only if the attribute is set on the instance.
Mock objects (and ``MagicMock``) auto-create attribute access; a
bare ``getattr(orch, name, None)`` would always return a child mock
and break tests that pass a ``MagicMock`` orchestrator. Checking
``__dict__`` confines the duck-typed hook to real, explicit
overrides.
"""
if orch is None:
return None
cls_attrs = getattr(type(orch), "__dict__", {})
if name in cls_attrs and not isinstance(getattr(type(orch), name, None), property):
# Class-level attribute (e.g. method) — treat as explicit.
return getattr(orch, name, None)
instance_dict = getattr(orch, "__dict__", None)
if isinstance(instance_dict, dict) and name in instance_dict:
return cast(object | None, instance_dict[name])
return None
def is_current_turn_owner(
*,
platform: str,
external_id: str | None = None,
session_key: str = "",
) -> bool:
"""Layering seam (R1-10) — owner check for the current turn.
Precedence:
1. If an orchestrator is bound and exposes a callable ``is_owner``
attribute with the same keyword signature, use it. This is the
duck-typed hook for per-orchestrator / per-tenant policy.
2. Otherwise, fall back to ``butler.gateway.owner_gate.is_gateway_owner``
(the single source of truth for owner allowlist / dev BYPASS).
Tool code MUST call this helper instead of importing
``butler.gateway.owner_gate`` directly so that the tools layer does
not have a hard reverse dependency on the gateway layer.
"""
orch = get_current_orchestrator()
is_owner = _orchestrator_explicit_attr(orch, "is_owner") if orch is not None else None
if callable(is_owner):
return bool(
is_owner(
platform=platform,
external_id=external_id,
session_key=session_key,
)
)
gate = get_owner_gate()
if gate is not None:
return bool(
gate.is_gateway_owner(
platform=platform,
external_id=external_id,
session_key=session_key,
)
)
return False
def owner_required_message() -> str:
"""Layering seam (R1-10) — owner-denied message string."""
gate = get_owner_gate()
if gate is not None:
return str(gate.owner_required_message())
return "此操作需要 Owner 权限。"
def get_current_turn_bridge() -> Any:
"""Layering seam (R1-10) — outbound bridge for the current turn.
Precedence:
1. If an orchestrator is bound and has a non-``None``
``gateway_bridge`` attribute set on the instance, return it.
2. Otherwise, use :class:`BridgeAccess` from contracts registry.
3. Otherwise, fall back to ``get_gateway_bridge_optional()``.
Tool code MUST call this helper instead of importing
``butler.gateway.outbound_bridge`` directly.
"""
orch = get_current_orchestrator()
if orch is not None:
bridge = _orchestrator_explicit_attr(orch, "gateway_bridge")
if bridge is not None:
return bridge
access = get_bridge_access()
if access is not None:
return access.get_optional_bridge()
return None
def try_push_current_turn_workflow_failure(
workflow_name: str,
error: "Exception | str",
*,
session_key: str = "",
) -> bool:
"""Layering seam (R1-10) — push a workflow-failure completion message."""
access = get_bridge_access()
if access is not None:
return bool(
access.try_push_workflow_failure(
workflow_name,
error,
session_key=session_key,
)
)
return False