mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
feat(authz): construction-enforced async decision batches (ActivationDecisions) (#6508)
* feat(authz): construction-enforced async decision batches (ActivationDecisions) The follow-up promised in the #4541 review threads. Four review rounds traced four instances of one failure family: an async execution path whose decision-map miss silently fell back to the synchronous authorize() from a worker thread — the wrong API for loop-affine providers, and under fail_closed:false a provider error there flips a denial into an allow. The carrier was a plain dict|None held together by caller discipline. ActivationDecisions (deerflow/authz/activation_decisions.py) makes the contract construction-enforced: - a covered name returns the batched decision (unchanged semantics); - a MISS resolves per the configured provider-error policy (fail-closed denies, fail-open allows) with a loud WARNING — a miss is a prepass/consumer divergence, a construction bug, not a policy question — and never touches a provider; - None remains the sync-chain signal (synchronous authorize() is correct there); - an empty candidate set still yields an empty BATCH, not None, so an unexpected consumer name fails per policy instead of silently taking the sync path; - get_or_none lets publishers distinguish 'uncovered' from False; - fail_closed travels with the batch from the resolved authorization context. Both middlewares (_activation_allowed in SkillActivationMiddleware and SkillToolPolicyMiddleware) migrate to the type; their collectors return batches; the render-decision publisher adapts. Tests 65->70+5: type unit tests (covered/miss both directions with loud-log assertions/ get_or_none/container), middleware-level miss semantics (fail-open allows with provider.sync_calls == [], production-default fail-closed denies, None sync chain still synchronous), and mutation M18 (restoring the miss's sync fallback) verified red. 252 tests green on the affected surface. * docs(authz): sync every decision-batch contract description to the new miss semantics Review round 1 on #6508 — all documentation-level: the docstrings still described the miss-as-synchronous-fallback contract this PR removes. - _activation_allowed (both middlewares): a covered name returns the batched decision; a miss resolves per the carried provider-error policy with a WARNING and never touches a provider; None (sync chain / disabled) is the only synchronous case. - _collect_activation_decisions / _resolve_policy_registry / _active_skills_for_paths: the reason not to diverge is now a policy-resolved miss (denied under the production fail-closed default; allowed without any provider consult under fail-open; logged loudly), not a silent sync fallback. - The two _REGISTRY_LOAD_FAILED sentinel comments, same rewording. - Drop unused tmp_path/monkeypatch fixtures from the two new miss tests. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
hataa
GitHub
parent
845af59301
commit
3e5bd081db
@ -20,6 +20,7 @@ from langchain.agents.middleware.types import ModelRequest, ModelResponse
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
|
||||
from deerflow.agents.middlewares.skill_usage import SKILL_USAGE_KEY, SKILL_USAGES_KEY, build_skill_usage, record_skill_usage
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
from deerflow.runtime.events.catalog import (
|
||||
MIDDLEWARE_SKILL_ACTIVATION_TAG,
|
||||
MIDDLEWARE_SKILL_SECRETS_TAG,
|
||||
@ -49,9 +50,9 @@ _SLASH_SKILL_ACTIVATION_TARGET_ID_KEY = "slash_skill_activation_target_id"
|
||||
|
||||
# Async-prepass registry load already attempted and failed. Handed down so
|
||||
# the secret-binding resolution does not silently recover with a fresh load:
|
||||
# entry names resolved from a recovered registry would miss the (empty)
|
||||
# decision map and fall back to the synchronous provider API from the worker
|
||||
# thread — for a loop-affine provider a denial becomes a fail-open allow.
|
||||
# entry names resolved from a recovered registry would surface as batch
|
||||
# misses — resolved per the carried fail-closed/fail-open policy with a
|
||||
# WARNING, never a silent synchronous fallback.
|
||||
# Entries bind nothing for that call instead (fail closed by construction).
|
||||
# Same marker pattern as SkillToolPolicyMiddleware._REGISTRY_LOAD_FAILED.
|
||||
_REGISTRY_LOAD_FAILED = object()
|
||||
@ -134,7 +135,7 @@ class SkillActivationMiddleware(AgentMiddleware):
|
||||
self._slash_source_owner_token = slash_source_owner_token
|
||||
self._skill_authorization = skill_authorization
|
||||
|
||||
def _activation_allowed(self, skill_name: str, *, activation_decisions: dict[str, bool] | None = None) -> bool:
|
||||
def _activation_allowed(self, skill_name: str, *, activation_decisions: ActivationDecisions | None = None) -> bool:
|
||||
"""Action-scoped ``skill:activate`` check for explicit slash activation.
|
||||
|
||||
``_available_skills`` is the Layer 1 *visibility* set (filter_resources,
|
||||
@ -146,20 +147,26 @@ class SkillActivationMiddleware(AgentMiddleware):
|
||||
Delegates to the shared ``skill_activation_allowed`` so the slash path,
|
||||
``describe_skill``, and the skill-file-load path cannot drift.
|
||||
|
||||
*activation_decisions* carries decisions precomputed on the event loop
|
||||
via ``aauthorize()`` by ``awrap_model_call``: the threaded handler then
|
||||
consults them instead of calling the synchronous ``authorize()`` (wrong
|
||||
API for loop-affine providers). Names absent from the map fall back to
|
||||
the synchronous check, which is the correct API for the sync
|
||||
``wrap_model_call`` path.
|
||||
*activation_decisions* carries the per-step
|
||||
:class:`~deerflow.authz.activation_decisions.ActivationDecisions`
|
||||
batch precomputed on the event loop via ``aauthorize()``: a covered
|
||||
name returns the batched decision; a miss inside the batch resolves
|
||||
per the carried provider-error policy with a loud WARNING and never
|
||||
touches a provider (wrong API from a worker thread for loop-affine
|
||||
providers). ``None`` (the sync ``wrap_model_call`` chain, or
|
||||
authorization disabled) is the only case that takes the synchronous
|
||||
check below — the correct API there.
|
||||
"""
|
||||
if activation_decisions is not None and skill_name in activation_decisions:
|
||||
return activation_decisions[skill_name]
|
||||
if activation_decisions is not None:
|
||||
# Construction-enforced: a miss inside an async batch resolves per
|
||||
# the provider-error policy with a loud log — it never falls back
|
||||
# to the synchronous authorize() from a worker thread.
|
||||
return activation_decisions.decision_for(skill_name)
|
||||
from deerflow.authz.skill_filter import skill_activation_allowed
|
||||
|
||||
return skill_activation_allowed(self._skill_authorization, skill_name)
|
||||
|
||||
async def _collect_activation_decisions(self, names) -> dict[str, bool] | None:
|
||||
async def _collect_activation_decisions(self, names) -> ActivationDecisions | None:
|
||||
"""Precompute ``skill:activate`` decisions on the event loop.
|
||||
|
||||
Async middleware paths offload the blocking handler (skill-tree reads)
|
||||
@ -173,13 +180,16 @@ class SkillActivationMiddleware(AgentMiddleware):
|
||||
return None
|
||||
candidates = {name for name in names if isinstance(name, str) and name}
|
||||
if not candidates:
|
||||
return None
|
||||
# An empty batch is still a batch: consumers that resolve a name
|
||||
# despite no candidates (a divergence) must fail per policy, not
|
||||
# silently take the sync path.
|
||||
return ActivationDecisions({}, fail_closed=self._skill_authorization.fail_closed)
|
||||
from deerflow.authz.skill_filter import skill_activation_allowed_async
|
||||
|
||||
decisions: dict[str, bool] = {}
|
||||
for name in sorted(candidates):
|
||||
decisions[name] = await skill_activation_allowed_async(self._skill_authorization, name)
|
||||
return decisions
|
||||
return ActivationDecisions(decisions, fail_closed=self._skill_authorization.fail_closed)
|
||||
|
||||
def _candidate_activation_targets(self, request: ModelRequest) -> tuple[list[str], list[str]]:
|
||||
"""Split the activation candidates into (names, entry_paths).
|
||||
@ -289,7 +299,7 @@ class SkillActivationMiddleware(AgentMiddleware):
|
||||
raise FileNotFoundError(resolved_file)
|
||||
return resolved_file.read_text(encoding="utf-8")
|
||||
|
||||
def _resolve_activation(self, text: str, *, activation_decisions: dict[str, bool] | None = None) -> _ActivationResolution | None:
|
||||
def _resolve_activation(self, text: str, *, activation_decisions: ActivationDecisions | None = None) -> _ActivationResolution | None:
|
||||
reference = parse_slash_skill_reference(text)
|
||||
if reference is None:
|
||||
return None
|
||||
@ -432,7 +442,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
"""
|
||||
return any(isinstance(message, AIMessage) for message in messages[target_index + 1 :])
|
||||
|
||||
def _find_activation_target(self, messages: list, *, run_context: dict | None = None, activation_decisions: dict[str, bool] | None = None) -> tuple[int, HumanMessage, _ActivationResolution, str] | None:
|
||||
def _find_activation_target(self, messages: list, *, run_context: dict | None = None, activation_decisions: ActivationDecisions | None = None) -> tuple[int, HumanMessage, _ActivationResolution, str] | None:
|
||||
if not messages:
|
||||
return None
|
||||
|
||||
@ -513,7 +523,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
except Exception:
|
||||
logger.warning("Failed to record slash skill activation audit event", exc_info=True)
|
||||
|
||||
def _prepare_model_request(self, request: ModelRequest, *, hook: str, activation_decisions: dict[str, bool] | None = None) -> tuple[ModelRequest | AIMessage | None, _Activation | None]:
|
||||
def _prepare_model_request(self, request: ModelRequest, *, hook: str, activation_decisions: ActivationDecisions | None = None) -> tuple[ModelRequest | AIMessage | None, _Activation | None]:
|
||||
run_context = self._run_context(request)
|
||||
target_and_resolution = self._find_activation_target(list(request.messages), run_context=run_context, activation_decisions=activation_decisions)
|
||||
if target_and_resolution is None:
|
||||
@ -559,7 +569,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
request: ModelRequest,
|
||||
*,
|
||||
hook: str,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
entry_registry: dict[str, Skill] | object | None = None,
|
||||
) -> tuple[ModelRequest | AIMessage, _Activation | None]:
|
||||
prepared, activation = self._prepare_model_request(request, hook=hook, activation_decisions=activation_decisions)
|
||||
@ -605,7 +615,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
activation: _Activation | None,
|
||||
*,
|
||||
hook: str,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
entry_registry: dict[str, Skill] | object | None = None,
|
||||
) -> None:
|
||||
"""Recompute the per-run secret injection set (binding point A+, #3861/#3914).
|
||||
@ -806,7 +816,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
request: ModelRequest,
|
||||
registry: dict[str, Skill],
|
||||
*,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
exclude_names: frozenset[str] = frozenset(),
|
||||
exclude_paths: frozenset[str] = frozenset(),
|
||||
) -> list[tuple[str, tuple[SecretRequirement, ...]]]:
|
||||
@ -920,7 +930,7 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
request: ModelRequest,
|
||||
entry_paths: list[str],
|
||||
entry_registry: dict[str, Skill] | object | None,
|
||||
activation_decisions: dict[str, bool] | None,
|
||||
activation_decisions: ActivationDecisions | None,
|
||||
) -> None:
|
||||
"""Publish per-step ``skill:activate`` decisions keyed by entry path.
|
||||
|
||||
@ -937,20 +947,24 @@ Follow this skill before choosing a general workflow. Load supporting resources
|
||||
|
||||
decisions: dict[str, bool] = {}
|
||||
registry = entry_registry if isinstance(entry_registry, dict) else {}
|
||||
collected = activation_decisions or {}
|
||||
collected = activation_decisions.get_or_none if activation_decisions is not None else (lambda _name: None)
|
||||
for path in entry_paths:
|
||||
normalized = posixpath.normpath(path)
|
||||
skill = registry.get(normalized)
|
||||
decisions[normalized] = collected.get(skill.name, False) if skill is not None else False
|
||||
decision = collected(skill.name) if skill is not None else None
|
||||
decisions[normalized] = bool(decision) if decision is not None else False
|
||||
write_skill_entry_decisions(run_context, decisions, owner_token=self._slash_source_owner_token)
|
||||
|
||||
def _collect_sync_activation_decisions(self, names: list[str]) -> dict[str, bool] | None:
|
||||
def _collect_sync_activation_decisions(self, names: list[str]) -> ActivationDecisions | None:
|
||||
"""Synchronous authorize() decisions for *names* (sync chain only)."""
|
||||
if self._skill_authorization is None or not names:
|
||||
return None
|
||||
from deerflow.authz.skill_filter import skill_activation_allowed
|
||||
|
||||
return {name: skill_activation_allowed(self._skill_authorization, name) for name in names}
|
||||
return ActivationDecisions(
|
||||
{name: skill_activation_allowed(self._skill_authorization, name) for name in names},
|
||||
fail_closed=self._skill_authorization.fail_closed,
|
||||
)
|
||||
|
||||
@override
|
||||
async def awrap_model_call(
|
||||
|
||||
@ -17,6 +17,7 @@ from langchain_core.messages import ToolMessage
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
from deerflow.runtime.secret_context import SKILL_TOOL_POLICY_DECISION_CONTEXT_KEY, read_slash_skill_source_paths
|
||||
from deerflow.skills.storage import get_or_new_skill_storage, get_or_new_user_skill_storage
|
||||
from deerflow.skills.tool_policy import ALWAYS_AVAILABLE_BUILTIN_TOOL_NAMES, allowed_tool_names_for_skills
|
||||
@ -36,9 +37,9 @@ _POLICY_SOURCES = frozenset({_POLICY_SOURCE_PASSIVE, _POLICY_SOURCE_SLASH, _POLI
|
||||
_MISSING_POLICY_DECISION = object()
|
||||
# Async-prepass registry load already attempted and failed: the async hook must
|
||||
# not let the worker-side filter silently retry storage — a successful retry
|
||||
# would resolve skills whose names are absent from the (empty) decision map and
|
||||
# fall back to the synchronous provider API from the thread, which for a
|
||||
# loop-affine provider turns a denial into a fail-open allow. Tri-state marker
|
||||
# would resolve skills whose names are absent from the (empty) decision batch —
|
||||
# a policy-resolved miss (per the carried fail-closed/fail-open policy, with a
|
||||
# WARNING), never a silent fallback. Tri-state marker
|
||||
# in the style of _MISSING_POLICY_DECISION.
|
||||
# Registry argument handed down the policy-resolution call chain: a loaded
|
||||
# snapshot (dict), the load-failure marker, or None ("load it here" — the
|
||||
@ -80,25 +81,33 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
self._skill_authorization = skill_authorization
|
||||
self._decision_owner_token = secrets.token_urlsafe(24)
|
||||
|
||||
def _activation_allowed(self, skill_name: str, *, activation_decisions: dict[str, bool] | None = None) -> bool:
|
||||
def _activation_allowed(self, skill_name: str, *, activation_decisions: ActivationDecisions | None = None) -> bool:
|
||||
"""Action-scoped ``skill:activate`` decision for one skill name.
|
||||
|
||||
Persisted ``skill_context`` entries were authorized when they were
|
||||
stamped, but the policy may have changed since — every entry is
|
||||
re-authorized before its allowed-tools declaration is applied.
|
||||
*activation_decisions* carries decisions precomputed on the event loop
|
||||
via ``aauthorize()`` (async hooks); names absent from the map fall back
|
||||
to the synchronous check, which is the correct API for the sync hooks.
|
||||
*activation_decisions* carries the per-step
|
||||
:class:`~deerflow.authz.activation_decisions.ActivationDecisions`
|
||||
batch precomputed on the event loop via ``aauthorize()`` (async
|
||||
hooks): a covered name returns the batched decision; a miss inside
|
||||
the batch resolves per the carried provider-error policy with a loud
|
||||
WARNING and never touches a provider. ``None`` (the sync hooks, or
|
||||
authorization disabled) is the only case that takes the synchronous
|
||||
check below — the correct API there.
|
||||
"""
|
||||
if self._skill_authorization is None:
|
||||
return True
|
||||
if activation_decisions is not None and skill_name in activation_decisions:
|
||||
return activation_decisions[skill_name]
|
||||
if activation_decisions is not None:
|
||||
# Construction-enforced: a miss inside an async batch resolves per
|
||||
# the provider-error policy with a loud log — it never falls back
|
||||
# to the synchronous authorize() from a worker thread.
|
||||
return activation_decisions.decision_for(skill_name)
|
||||
from deerflow.authz.skill_filter import skill_activation_allowed
|
||||
|
||||
return skill_activation_allowed(self._skill_authorization, skill_name)
|
||||
|
||||
async def _collect_activation_decisions(self, names) -> dict[str, bool] | None:
|
||||
async def _collect_activation_decisions(self, names) -> ActivationDecisions | None:
|
||||
"""Precompute ``skill:activate`` decisions on the event loop (async hooks).
|
||||
|
||||
The policy resolution itself runs in a worker thread (storage reads),
|
||||
@ -107,20 +116,22 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
resolves the policy paths through the live registry off-loop first,
|
||||
because the decision consumers (``_active_skills_for_paths``) check
|
||||
the registry skill's declared name. Authorizing a path-derived name
|
||||
here would miss the map and fall back to the synchronous provider
|
||||
call from the worker thread (wrong API for loop-affine providers).
|
||||
here would surface as a batch miss at the consumer, which resolves
|
||||
per the carried fail-closed/fail-open policy with a WARNING (denied
|
||||
under the production default; allowed without any provider consult
|
||||
under fail-open) — visible, but still a wiring bug to fix.
|
||||
"""
|
||||
if self._skill_authorization is None:
|
||||
return None
|
||||
candidates = {name for name in names if isinstance(name, str) and name}
|
||||
if not candidates:
|
||||
return None
|
||||
from deerflow.authz.skill_filter import skill_activation_allowed_async
|
||||
|
||||
decisions: dict[str, bool] = {}
|
||||
for name in sorted(candidates):
|
||||
decisions[name] = await skill_activation_allowed_async(self._skill_authorization, name)
|
||||
return decisions
|
||||
# An empty batch is still a batch — consumers resolving an uncovered
|
||||
# name must fail per policy, not silently take the sync path.
|
||||
return ActivationDecisions(decisions, fail_closed=self._skill_authorization.fail_closed)
|
||||
|
||||
def _resolve_policy_registry(self, paths: tuple[str, ...]) -> tuple[dict[str, Skill] | object, list[str]]:
|
||||
"""Load the path-keyed registry and canonicalize the policy paths.
|
||||
@ -131,8 +142,10 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
marker preserves the failure: the async caller hands it down so
|
||||
``_active_skills_for_paths`` applies its fail-closed treatment instead
|
||||
of retrying storage (a transient-failure retry that succeeds would
|
||||
resolve skills missing from the decision map and fall back to the
|
||||
synchronous provider API). Unresolvable paths yield no name:
|
||||
resolve skills whose names are absent from the decision batch — a
|
||||
policy-resolved miss: denied under the production fail-closed
|
||||
default, allowed without any provider consult under fail-open, and
|
||||
logged loudly either way). Unresolvable paths yield no name:
|
||||
``_active_skills_for_paths`` skips them before the activation check,
|
||||
so no decision is needed for them.
|
||||
"""
|
||||
@ -193,7 +206,7 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
self,
|
||||
paths: tuple[str, ...],
|
||||
*,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
registry: _RegistryArg = None,
|
||||
) -> tuple[list[Skill], bool]:
|
||||
if not paths:
|
||||
@ -250,7 +263,7 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
self,
|
||||
paths: tuple[str, ...],
|
||||
*,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
registry: _RegistryArg = None,
|
||||
) -> set[str] | None:
|
||||
active_skills, policy_failed = self._active_skills_for_paths(paths, activation_decisions=activation_decisions, registry=registry)
|
||||
@ -307,7 +320,7 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
request: ModelRequest | ToolCallRequest,
|
||||
*,
|
||||
policy: _PolicySignature | None = None,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
registry: _RegistryArg = None,
|
||||
) -> set[str] | None:
|
||||
resolved_policy = self._active_policy(request) if policy is None else policy
|
||||
@ -324,7 +337,7 @@ class SkillToolPolicyMiddleware(AgentMiddleware[AgentState]):
|
||||
*,
|
||||
policy: _PolicySignature | None = None,
|
||||
refresh_decision: bool = False,
|
||||
activation_decisions: dict[str, bool] | None = None,
|
||||
activation_decisions: ActivationDecisions | None = None,
|
||||
registry: _RegistryArg = None,
|
||||
) -> ModelRequest:
|
||||
resolved_policy = self._active_policy(request) if policy is None else policy
|
||||
|
||||
@ -0,0 +1,87 @@
|
||||
"""Typed carrier for per-step ``skill:activate`` decisions (async hooks).
|
||||
|
||||
The async middleware hooks precompute ``skill:activate`` decisions on the
|
||||
event loop with ``aauthorize()`` and hand them to the blocking handlers that
|
||||
run in worker threads. Before this type existed the carrier was a plain
|
||||
``dict | None``: ``None`` meant "sync chain — the synchronous ``authorize()``
|
||||
is the correct API", and a name *missing* from a non-None map silently fell
|
||||
back to the same synchronous call from the thread — the wrong API for
|
||||
loop-affine providers, and under ``fail_closed: false`` a provider error
|
||||
there flips a denial into an allow.
|
||||
|
||||
Four review rounds on #4541 traced four instances of that miss (wrong keys,
|
||||
failed prepass, snapshot divergence, a consumer the prepass did not cover).
|
||||
This type makes the contract construction-enforced instead of
|
||||
caller-disciplined:
|
||||
|
||||
- ``None`` (absent) — sync chain; consumers call the synchronous
|
||||
``skill_activation_allowed`` as before.
|
||||
- :class:`ActivationDecisions` (present) — an async batch. A name it covers
|
||||
returns the batched decision; a name it does NOT cover fails per the
|
||||
configured provider-error policy with a loud log and **never** calls a
|
||||
provider synchronously.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ActivationDecisions:
|
||||
"""A per-model-step batch of ``skill:activate`` decisions.
|
||||
|
||||
``fail_closed`` carries the provider-error policy from the resolved
|
||||
authorization context so a miss resolves exactly like a provider error
|
||||
would (deny when fail-closed, allow when fail-open) — the difference
|
||||
being that a miss is always a wiring bug, so it is logged loudly at
|
||||
WARNING instead of silently shaping an allow/deny.
|
||||
"""
|
||||
|
||||
__slots__ = ("_decisions", "_fail_closed")
|
||||
|
||||
def __init__(self, decisions: Mapping[str, bool], *, fail_closed: bool) -> None:
|
||||
self._decisions = dict(decisions)
|
||||
self._fail_closed = bool(fail_closed)
|
||||
|
||||
@property
|
||||
def fail_closed(self) -> bool:
|
||||
return self._fail_closed
|
||||
|
||||
def __contains__(self, name: str) -> bool:
|
||||
return name in self._decisions
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._decisions)
|
||||
|
||||
def decision_for(self, name: str) -> bool:
|
||||
"""The batched decision for *name*; a miss fails per policy, loudly.
|
||||
|
||||
A miss means the prepass failed to cover a name its consumer
|
||||
resolved — a construction bug, not a policy question. It resolves
|
||||
like a provider error (fail-closed denies, fail-open allows) and
|
||||
logs at WARNING so the divergence is visible in production logs
|
||||
instead of silently exercising the wrong API.
|
||||
"""
|
||||
try:
|
||||
return self._decisions[name]
|
||||
except KeyError:
|
||||
logger.warning(
|
||||
"skill:activate decision for '%s' missing from the async batch; failing %s (prepass/consumer divergence — not a policy denial)",
|
||||
name,
|
||||
"closed (denied)" if self._fail_closed else "open (allowed)",
|
||||
)
|
||||
return not self._fail_closed
|
||||
|
||||
def get_or_none(self, name: str) -> bool | None:
|
||||
"""The batched decision, or ``None`` when *name* is not covered.
|
||||
|
||||
For publishers that must distinguish "covered" from "miss" (e.g. the
|
||||
rendered-reminder publication treats an uncovered path as hide).
|
||||
"""
|
||||
return self._decisions.get(name)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover - debugging aid
|
||||
return f"ActivationDecisions({self._decisions!r}, fail_closed={self._fail_closed!r})"
|
||||
54
backend/tests/test_activation_decisions.py
Normal file
54
backend/tests/test_activation_decisions.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""ActivationDecisions — the typed async decision carrier (construction hardening).
|
||||
|
||||
The carrier's contract (see ``deerflow/authz/activation_decisions.py``):
|
||||
a covered name returns the batched decision; a MISS resolves per the
|
||||
provider-error policy with a loud WARNING and never touches a provider —
|
||||
``None`` remains the sync-chain signal where the synchronous authorize() is
|
||||
the correct API.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
|
||||
|
||||
def test_covered_name_returns_batched_decision():
|
||||
decisions = ActivationDecisions({"alpha": True, "beta": False}, fail_closed=True)
|
||||
|
||||
assert decisions.decision_for("alpha") is True
|
||||
assert decisions.decision_for("beta") is False
|
||||
|
||||
|
||||
def test_miss_fails_closed_with_loud_log(caplog):
|
||||
decisions = ActivationDecisions({}, fail_closed=True)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="deerflow.authz.activation_decisions"):
|
||||
assert decisions.decision_for("uncovered") is False
|
||||
|
||||
assert any("uncovered" in record.message and "denied" in record.message for record in caplog.records)
|
||||
|
||||
|
||||
def test_miss_fails_open_with_loud_log(caplog):
|
||||
decisions = ActivationDecisions({}, fail_closed=False)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="deerflow.authz.activation_decisions"):
|
||||
assert decisions.decision_for("uncovered") is True
|
||||
|
||||
assert any("uncovered" in record.message and "allowed" in record.message for record in caplog.records)
|
||||
|
||||
|
||||
def test_get_or_none_distinguishes_miss_from_false():
|
||||
decisions = ActivationDecisions({"alpha": False}, fail_closed=True)
|
||||
|
||||
assert decisions.get_or_none("alpha") is False
|
||||
assert decisions.get_or_none("uncovered") is None
|
||||
|
||||
|
||||
def test_container_protocol_and_len():
|
||||
decisions = ActivationDecisions({"alpha": True}, fail_closed=True)
|
||||
|
||||
assert "alpha" in decisions
|
||||
assert "uncovered" not in decisions
|
||||
assert len(decisions) == 1
|
||||
@ -1723,11 +1723,14 @@ def test_async_tool_policy_uses_aauthorize(monkeypatch, tmp_path):
|
||||
)
|
||||
request = SimpleNamespace(state={"skill_context": [{"name": "demo-skill", "path": skill_path}]})
|
||||
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
|
||||
async def _run():
|
||||
return await middleware._collect_activation_decisions(["demo-skill"])
|
||||
|
||||
decisions = asyncio.run(_run())
|
||||
assert decisions == {"demo-skill": True}
|
||||
assert isinstance(decisions, ActivationDecisions)
|
||||
assert decisions.decision_for("demo-skill") is True
|
||||
|
||||
# End to end through the async hook: the aauthorize-granted entry applies.
|
||||
class _ToolsRequest(SimpleNamespace):
|
||||
@ -3074,6 +3077,69 @@ def test_slash_dominance_anchors_on_path_across_midrun_rename(tmp_path, monkeypa
|
||||
assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"NEW_KEY": "new-value"}
|
||||
|
||||
|
||||
def test_async_batch_miss_never_calls_synchronous_provider():
|
||||
"""[construction hardening] A name missing from an async batch resolves
|
||||
per the provider-error policy WITHOUT consulting the synchronous
|
||||
authorize() — the wrong API from a worker thread. Under the old dict
|
||||
contract the miss silently took that path; with a loop-affine provider
|
||||
and fail_closed=false it flipped denials into allows."""
|
||||
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
|
||||
provider = _ActionAwareProvider(denied_activate=set())
|
||||
resolved = _resolved_skill_authorization(provider, fail_closed=False)
|
||||
middleware = SkillActivationMiddleware(
|
||||
available_skills={"demo-skill"},
|
||||
skill_authorization=resolved,
|
||||
slash_source_owner_token="test-token",
|
||||
)
|
||||
empty_batch = ActivationDecisions({}, fail_closed=False)
|
||||
|
||||
# Miss under fail-open allows — and the sync API is never consulted
|
||||
# (the old fallback would have recorded a sync call).
|
||||
assert middleware._activation_allowed("uncovered", activation_decisions=empty_batch) is True
|
||||
assert middleware._activation_allowed("demo-skill", activation_decisions=empty_batch) is True
|
||||
assert provider.sync_calls == []
|
||||
assert provider.async_calls == []
|
||||
|
||||
|
||||
def test_async_batch_miss_fails_closed_under_default_policy():
|
||||
"""The production default (fail_closed=True) denies a miss — loudly, and
|
||||
still without touching the synchronous provider."""
|
||||
|
||||
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
||||
from deerflow.authz.activation_decisions import ActivationDecisions
|
||||
|
||||
provider = _ActionAwareProvider(denied_activate=set())
|
||||
resolved = _resolved_skill_authorization(provider, fail_closed=True)
|
||||
middleware = SkillActivationMiddleware(
|
||||
available_skills={"demo-skill"},
|
||||
skill_authorization=resolved,
|
||||
slash_source_owner_token="test-token",
|
||||
)
|
||||
empty_batch = ActivationDecisions({}, fail_closed=True)
|
||||
|
||||
assert middleware._activation_allowed("uncovered", activation_decisions=empty_batch) is False
|
||||
assert provider.sync_calls == []
|
||||
|
||||
|
||||
def test_sync_chain_none_still_uses_synchronous_api():
|
||||
"""``None`` remains the sync-chain signal: the synchronous authorize() is
|
||||
the correct API there and stays in use."""
|
||||
provider = _ActionAwareProvider(denied_activate=set())
|
||||
resolved = _resolved_skill_authorization(provider, fail_closed=True)
|
||||
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
||||
|
||||
middleware = SkillActivationMiddleware(
|
||||
available_skills={"demo-skill"},
|
||||
skill_authorization=resolved,
|
||||
slash_source_owner_token="test-token",
|
||||
)
|
||||
|
||||
assert middleware._activation_allowed("demo-skill", activation_decisions=None) is True
|
||||
assert provider.sync_calls == ["demo-skill"]
|
||||
|
||||
|
||||
def posixpath_normpath(path: str) -> str:
|
||||
import posixpath
|
||||
|
||||
|
||||
@ -1150,6 +1150,28 @@ Phase 1 最低验证要求：
|
||||
- **复盘：** 本洞由"对刚写的修复立即执行新清单条目"抓出——第 41 条
|
||||
（重攻自身修复）的直接收益。
|
||||
|
||||
### 2026-09-28 — 决策 map 构造性加固（#4541 review 线程预告的 follow-up）
|
||||
|
||||
- **背景：** 四轮 review（R5–R9 前 #4541 合并前后）追踪到同一失败类的四个实例：
|
||||
异步执行路径上决策 map 的 miss 静默回退 worker 线程里的同步 `authorize()`——
|
||||
对 loop-affine provider 是错误 API，`fail_closed=false` 时拒绝翻成放行。根因是
|
||||
载体契约（`dict | None`）由调用方自觉维持：`None` = 同步链，miss = 回退同步。
|
||||
- **决策（类型化载体）：** `deerflow/authz/activation_decisions.py` 的
|
||||
`ActivationDecisions`：覆盖名返回批量决策；**miss 按配置的 provider-error 策略
|
||||
解析（fail-closed 拒 / fail-open 放）并以 WARNING 响亮记日志，从不触达
|
||||
provider**——miss 是构造 bug 不是策略问题。`None` 保留为同步链信号（同步
|
||||
`authorize()` 在同步链是正确 API）。空候选集返回空批而非 `None`：即便无候选，
|
||||
消费者的意外名字也必须按策略失败。`get_or_none` 供发布者区分"未覆盖"与 False。
|
||||
`fail_closed` 随载体携带（取自 resolved 授权上下文）。
|
||||
- **证据：** `tests/test_activation_decisions.py` 5 个类型单测（覆盖返回/miss
|
||||
双方向 + 响亮日志断言/get_or_none/容器协议）+ 中间件级 3 测试（miss 在
|
||||
fail-open 允许且 `provider.sync_calls == []`、生产默认 fail-closed 拒绝、
|
||||
`None` 同步链仍走同步 API）。突变 M18（恢复 miss 的同步回退）必红。
|
||||
迁移后两中间件 + 既有套件 252 passed。
|
||||
- **兼容性：** 同步链（`None`）行为逐字不变；批量内覆盖名的语义不变；仅 miss
|
||||
的处置从"静默同步回退"变为"按策略失败 + 响亮日志"——正是四轮 review 追逐
|
||||
的行为面。
|
||||
|
||||
### 新记录模板
|
||||
|
||||
```markdown
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user