Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-skills/src/lexigram/ai/skills/executor/core.py: 29%
66 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""SkillExecutor — executes skills with retry, caching, and permission checks."""
3from __future__ import annotations
5import asyncio
6import time
7from typing import TYPE_CHECKING, Any, Protocol, cast
9from lexigram.ai.skills.exceptions import (
10 SkillNotFoundError,
11 SkillPermissionDeniedError,
12 SkillTimeoutError,
13 SkillValidationError,
14)
15from lexigram.contracts.ai.skills import SkillError, SkillResult
16from lexigram.logging import (
17 get_logger,
18)
19from lexigram.result import Err, Ok, Result
21if TYPE_CHECKING:
22 from lexigram.contracts.ai.skills import SkillRegistryProtocol
23 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
26class PermissionCheckerProtocol(Protocol):
27 """Protocol for permission checking on skill execution.
29 Defines the interface for verifying user access to skill execution
30 based on required permissions.
31 """
33 def check(self, user_id: str, permissions: set[str]) -> bool:
34 """Check if user has required permissions.
36 Args:
37 user_id: The user identifier.
38 permissions: Set of required permission strings.
40 Returns:
41 True if user has all required permissions, False otherwise.
42 """
43 ...
46logger = get_logger(__name__)
49class SkillExecutor:
50 """Executes skills with the full production lifecycle.
52 Handles skill resolution, permission checking, parameter validation,
53 result caching, retry with exponential backoff, timeout enforcement,
54 and observability logging.
55 """
57 def __init__(
58 self,
59 registry: SkillRegistryProtocol,
60 *,
61 cache: CacheBackendProtocol | None = None,
62 permission_checker: PermissionCheckerProtocol | None = None,
63 semaphore: asyncio.Semaphore | None = None,
64 ) -> None:
65 """Initialise the executor.
67 Args:
68 registry: SkillRegistryProtocol implementation.
69 cache: Optional CacheBackendProtocol for caching deterministic results.
70 permission_checker: Optional PermissionCheckerProtocol for access control.
71 semaphore: Optional semaphore limiting concurrent executions.
72 """
73 self._registry = registry
74 self._cache = cache
75 self._permissions = permission_checker
76 self._semaphore = semaphore
78 async def execute(
79 self,
80 skill_name: str,
81 params: dict[str, Any],
82 user_id: str | None = None,
83 session_id: str | None = None,
84 ) -> Result[SkillResult, SkillError]:
85 """Execute a skill with full lifecycle management.
87 Steps: resolve → check permissions → validate → check cache →
88 execute with retry → store cache → return.
90 Args:
91 skill_name: Registered name of the skill to execute.
92 params: Skill parameters.
93 user_id: Optional caller identity for permission checks.
94 session_id: Optional session context (passed through to metadata).
96 Returns:
97 Result wrapping SkillResult on success or SkillError on failure.
98 """
99 start = time.monotonic()
101 # 1. Resolve skill
102 skill = self._registry.get(skill_name)
103 if skill is None:
104 return Err(SkillNotFoundError(skill_name))
106 defn = skill.definition
108 # 2. Permission check
109 if self._permissions and defn.permissions and user_id is not None:
110 allowed = self._permissions.check(
111 user_id,
112 set(defn.permissions),
113 )
114 if not allowed:
115 return Err(SkillPermissionDeniedError(skill_name, defn.permissions))
117 # 3. Validate parameters
118 errors = skill.validate(params)
119 if errors:
120 return Err(SkillValidationError(skill_name, errors))
122 # 4. Cache lookup
123 if defn.cacheable and self._cache is not None:
124 cache_result = cast(
125 "SkillResult | None",
126 await self._cache.get(skill_name, params), # type: ignore[call-arg]
127 )
128 if cache_result is not None:
129 logger.debug("skill_cache_hit", skill=skill_name)
130 return Ok(
131 SkillResult(
132 skill_name=skill_name,
133 success=True,
134 output=cache_result.output,
135 cached=True,
136 duration_ms=0.0,
137 metadata={"session_id": session_id},
138 )
139 )
141 # 5. Execute with retry
142 last_error: SkillError | None = None
143 for attempt in range(defn.max_retries + 1):
144 try:
145 coro = skill.execute(**params)
146 if self._semaphore is not None:
147 async with self._semaphore:
148 result = await asyncio.wait_for(
149 coro, timeout=defn.timeout_seconds
150 )
151 else:
152 result = await asyncio.wait_for(coro, timeout=defn.timeout_seconds)
154 if result.is_ok():
155 sr = result.unwrap()
156 duration_ms = (time.monotonic() - start) * 1000
157 final = SkillResult(
158 skill_name=sr.skill_name,
159 success=sr.success,
160 output=sr.output,
161 error=sr.error,
162 duration_ms=duration_ms,
163 cached=sr.cached,
164 metadata={
165 **sr.metadata,
166 "session_id": session_id,
167 "attempt": attempt,
168 },
169 )
170 if defn.cacheable and self._cache is not None:
171 await self._cache.set(skill_name, params, final) # type: ignore[arg-type]
172 logger.debug(
173 "skill_executed",
174 skill=skill_name,
175 attempt=attempt,
176 duration_ms=duration_ms,
177 )
178 return Ok(final)
180 last_error = result.unwrap_err()
182 except TimeoutError:
183 last_error = SkillTimeoutError(skill_name, defn.timeout_seconds)
184 except SkillError as exc:
185 last_error = exc
186 except Exception as exc: # noqa: BLE001
187 last_error = SkillError(
188 f"Unexpected error: {exc}", skill_name=skill_name
189 )
191 if attempt < defn.max_retries:
192 backoff = min(2**attempt, 30)
193 logger.debug(
194 "skill_retry",
195 skill=skill_name,
196 attempt=attempt,
197 backoff=backoff,
198 error=str(last_error),
199 )
200 await asyncio.sleep(backoff)
202 duration_ms = (time.monotonic() - start) * 1000
203 logger.warning(
204 "skill_failed",
205 skill=skill_name,
206 duration_ms=duration_ms,
207 error=str(last_error),
208 )
209 return Err(last_error or SkillError("Unknown error", skill_name=skill_name))
212__all__ = ["SkillExecutor"]