class handler_wrapper:
"""
Descriptor that replaces the original function.
Tracks invocations and writes serialized params/result to storage.
Supports retry, return-type validation, and storage injection.
"""
def __init__(
self,
func: Callable,
prefix: str,
storage: StoragePlugin,
serializer: Serialize,
retry: Optional[Union[int, RetryPolicy]] = None,
durability: Union[str, Durability] = Durability.AT_MOST_ONCE,
middleware: Optional[MiddlewareStack] = None,
exception_registry: Optional[ExceptionHandlerRegistry] = None,
authorizer: Optional[Authorizer] = None,
dependency_overrides: Optional[Mapping[Callable, Callable]] = None,
approval: Union[bool, str] = False,
idempotency_lease_s: Optional[float] = None,
):
self.func = func
self.prefix = prefix
# Advisory, for agents: "a human should clear this call before it runs".
# Carried in the capability manifest; enforcement is the caller's gate
# (see istos.agent.approval), not this wrapper — the authorizer is what
# stops a peer that ignores the flag.
self.approval = approval
self.storage = storage
self.serializer = serializer
self.calls = 0
self._middleware = middleware
self._authorizer = authorizer
self._exception_registry = exception_registry or get_default_registry()
self._logger = get_logger("handler")
self.idempotency_lease_s = (
DEFAULT_CLAIM_LEASE_S if idempotency_lease_s is None else idempotency_lease_s
)
if retry is None:
self.retry_policy = RetryPolicy(max_retries=0)
elif isinstance(retry, int):
self.retry_policy = RetryPolicy.from_int(retry)
else:
self.retry_policy = retry
_params = inspect.signature(func).parameters
self._inject_db = "db" in _params
self._depends_params = {n for n, p in _params.items() if extract_depends(p) is not None}
self._has_depends = bool(self._depends_params)
# db / Depends(...) are framework-injected — not network params.
self._injected_params = set(self._depends_params)
if self._inject_db:
self._injected_params.add("db")
self._dependency_overrides = dependency_overrides if dependency_overrides is not None else {}
if isinstance(durability, str):
self.durability = Durability(durability)
else:
self.durability = durability
hints = get_type_hints(func)
self._return_type = hints.get("return", None)
@staticmethod
def _make_idempotency_key(prefix: str, params: dict, identity: str = "") -> str:
"""
Deterministic key from prefix + caller + sorted params.
Positional and keyword arguments are bound to the same names first, so
``move(10)`` and ``move(distance=10)`` match and ``move(10)`` does not
match ``move(20)``. ``identity`` is the principal (or a hash of the
token) so two callers with the same params do not share a result.
"""
raw = f"{prefix}\n{identity}\n{_json.dumps(params, sort_keys=True, default=str)}"
return hashlib.sha256(raw.encode()).hexdigest()
def _bound_params(self, args: Tuple[Any, ...], kwargs: dict) -> dict:
"""Arguments the handler was actually called with, minus injected ones."""
sig = inspect.signature(self.func)
try:
bound = sig.bind_partial(*args, **kwargs)
except TypeError:
raw = {f"arg{i}": value for i, value in enumerate(args)}
raw.update(kwargs)
return {
k: v for k, v in raw.items()
if k not in self._injected_params and k != "self"
}
params: dict = {}
for name, value in bound.arguments.items():
if name == "self" or name in self._injected_params:
continue
param = sig.parameters[name]
if param.kind is inspect.Parameter.VAR_KEYWORD and isinstance(value, dict):
params.update(value)
elif param.kind is inspect.Parameter.VAR_POSITIONAL and isinstance(value, tuple):
for index, item in enumerate(value):
params[f"arg{index}"] = item
else:
params[name] = value
return params
@staticmethod
def _caller_identity() -> str:
"""Who is calling, for the exactly-once key. Empty when anonymous."""
ctx = peek_request_context()
if ctx is None:
return ""
principal = ctx.principal
if isinstance(principal, str) and principal:
return "p:" + principal
ident = getattr(principal, "id", None) if principal is not None else None
if ident:
return "p:" + str(ident)
token = ctx.token
if token:
return "t:" + hashlib.sha256(token.encode()).hexdigest()[:16]
return ""
def _validate_return(self, result: Any) -> Any:
"""Validate the return value against the function's return type hint."""
if self._return_type is None or not HAS_PYDANTIC:
return result
if self._return_type is type(None):
return result
try:
adapter = TypeAdapter(self._return_type)
return adapter.validate_python(result)
except ValidationError as e:
raise SchemaValidationError(
e.errors(),
message=f"Return type validation failed for '{self.func.__name__}'"
) from e
async def __call__(self, *args: Any, **kwargs: Any) -> Any:
self.calls += 1
idemp_params = self._bound_params(args, kwargs)
idemp_key = self._make_idempotency_key(
self.prefix, idemp_params, self._caller_identity()
)
claimed = False
if self.durability == Durability.EXACTLY_ONCE:
# Claim before executing: a check first and a mark afterwards leaves
# the whole handler body between the two, so every concurrent
# redelivery passes the check and runs the side effects.
claim = await self._claim(idemp_key)
if claim.state is ClaimState.DONE:
return claim.result
if claim.state is ClaimState.IN_FLIGHT:
raise ConflictError(
f"'{self.prefix}' is already executing this exact request; "
"retry to collect its result",
details={"idempotency_key": idemp_key},
)
claimed = True
try:
return await self._invoke(args, kwargs, idemp_params, idemp_key)
except BaseException:
# The work did not complete, so the key must not stay claimed or a
# failed call is never retryable. Cancellation counts: a cancelled
# handler has no result either.
if claimed:
await self._release(idemp_key)
raise
async def _claim(self, idemp_key: str) -> Claim:
"""Claim the idempotency key, atomically where the backend can."""
claim_processed = getattr(self.storage, "claim_processed", None)
if claim_processed is not None:
return cast(Claim, await claim_processed(idemp_key, lease_s=self.idempotency_lease_s))
# A plugin written against the pre-claim protocol. Check-then-act is all
# that is left, and it does not hold under concurrency — say so rather
# than let it pass as exactly-once.
self._warn_no_claim_support()
cached = await self.storage.check_processed(idemp_key)
if cached is not None:
return Claim(ClaimState.DONE, cached)
return Claim(ClaimState.CLAIMED)
async def _release(self, idemp_key: str) -> None:
release_claim = getattr(self.storage, "release_claim", None)
if release_claim is None:
return
try:
await release_claim(idemp_key)
except Exception:
# The lease expires anyway, so this costs a delay, not correctness.
# Never mask the error that got us here.
self._logger.warning(
"failed to release idempotency claim; it expires in %ss",
self.idempotency_lease_s,
exc_info=True,
)
def _warn_no_claim_support(self) -> None:
backend = type(self.storage).__name__
if backend in _WARNED_NO_CLAIM:
return
_WARNED_NO_CLAIM.add(backend)
warnings.warn(
f"{backend} does not implement claim_processed(), so "
"durability='exactly_once' degrades to at-least-once with a result "
"cache: concurrent duplicates of one request can all execute. "
"Implement claim_processed()/release_claim() (see StoragePlugin) or "
"use a built-in storage backend.",
RuntimeWarning,
stacklevel=2,
)
async def _invoke(
self,
args: Tuple[Any, ...],
kwargs: dict,
idemp_params: dict,
idemp_key: str,
) -> Any:
"""Resolve dependencies, run the handler, and record the outcome."""
# Exit stack tears down yield-deps after the handler finishes.
async with AsyncExitStack() as di_stack:
call_kwargs = dict(kwargs)
if self._inject_db and "db" not in call_kwargs:
call_kwargs["db"] = self.storage
if self._has_depends:
call_kwargs = await resolve_dependencies(
self.func,
call_kwargs,
di_stack,
cache={},
overrides=self._dependency_overrides,
)
async def _execute():
async def _handler(scope: RequestScope) -> Any:
if inspect.iscoroutinefunction(self.func):
return await self.func(*args, **call_kwargs)
return await asyncio.to_thread(self.func, *args, **call_kwargs)
if self._middleware is not None:
scope = RequestScope(
prefix=self.prefix,
operation="handle",
params=idemp_params,
)
# middleware.invoke() installs a fresh context — copy
outer = peek_request_context()
if outer is not None:
scope.context.principal = outer.principal
scope.context.attachment = outer.attachment
scope.context.correlation_id = outer.correlation_id
scope.context.traceparent = outer.traceparent
return await self._middleware.invoke(scope, _handler)
return await _handler(RequestScope(prefix=self.prefix, operation="handle"))
result = await execute_with_retry(_execute, self.retry_policy)
result = self._validate_return(result)
metadata = {
"func_name": self.func.__name__,
"total_calls": self.calls,
"timestamp": time.time(),
"status": "ok",
}
try:
serialized = self.serializer.serialize(metadata)
await self.storage.put(self.prefix, serialized)
if self.durability in (Durability.AT_LEAST_ONCE, Durability.EXACTLY_ONCE):
await self.storage.log(self.prefix, serialized, idempotency_key=idemp_key)
except Exception:
pass # storage write must never crash the handler
# Complete the claim last: log() skips keys already marked done, so
# the event log must land while the key is still only claimed. Not
# best-effort like the writes above — swallowing this leaves the key
# claimed until the lease lapses and the work runs twice.
if self.durability == Durability.EXACTLY_ONCE:
await self.storage.mark_processed(idemp_key, result)
return result
@staticmethod
def _extract_attachment(query: zenoh.Query) -> Optional[bytes]:
"""Best-effort read of a query's attachment as raw bytes (for auth tokens)."""
raw = getattr(query, "attachment", None)
if raw is None:
return None
try:
return bytes(raw)
except (TypeError, ValueError):
return None
async def on_query(self, query: zenoh.Query) -> None:
ctx = None
token = None
try:
key = str(query.selector.key_expr)
# zenoh.Parameters has no .items(); iterates as (key, value).
# stub omits __iter__, but it is iterable at runtime.
params: dict = {}
if hasattr(query.selector, "parameters") and query.selector.parameters:
params = decode_params(
dict(cast(Iterable[Tuple[str, str]], query.selector.parameters))
)
# Network gate only — TestClient / in-process __call__ skips this.
attachment = self._extract_attachment(query)
ctx = ingress_context(
prefix=self.prefix, operation="handle", attachment=attachment,
)
token = push_request_context(ctx)
try:
principal = await check_authorized(
self._authorizer,
AuthContext(
prefix=self.prefix,
key_expr=key,
params=params,
attachment=attachment,
),
)
except UnauthorizedError as e:
error = self._exception_registry.resolve(e)
error.correlation_id = ctx.correlation_id
self._logger.warning(
"Unauthorized request on %s: %s", self.prefix, e,
extra={"prefix": self.prefix, "error": str(e)},
)
try:
query.reply(key, self.serializer.serialize(error.to_dict()))
except Exception:
pass
return
ctx.principal = principal
try:
validated_params = validate_params(
self.func, params, skip_params=self._injected_params
)
validated_params.pop("db", None)
except SchemaValidationError as e:
error = self._exception_registry.resolve(e)
error.correlation_id = ctx.correlation_id
self._logger.warning(
"Validation error on %s: %s", self.prefix, e,
extra={"prefix": self.prefix, "error": str(e)},
)
query.reply(key, self.serializer.serialize(error.to_dict()))
return
result = await self(**validated_params)
if result is not None:
payload = self.serializer.serialize(result)
query.reply(key, payload)
except Exception as e:
error = self._exception_registry.resolve(e)
error.correlation_id = ctx.correlation_id if ctx is not None else None
self._logger.error(
"Handler error on %s: %s", self.prefix, e,
exc_info=True,
extra={"prefix": self.prefix},
)
try:
query.reply(
str(query.selector.key_expr),
self.serializer.serialize(error.to_dict()),
)
except Exception:
pass
finally:
if token is not None:
pop_request_context(token)
def __get__(self, instance: Any, owner: Any) -> Any:
if instance is None:
return self
return bound_handler_wrapper(self, instance)