forked from HAL9000/cleveragents-core
aditya
2764fcef5c
Address review-driven fixes across actor schema, preflight guardrails, docs/examples, and Behave/Robot coverage: unify preflight warning behavior with shared role-warning logic, resolve actor-name to config payloads in production preflight flow, harden response_format validation/coercion edge cases, extract duplicated helper logic, and expand negative-path test coverage. Also fix cross-scenario patcher leakage in step modules to eliminate full-run-only coverage failures.
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""Role-aware compatibility helpers for actor configurations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from enum import StrEnum
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RoleHint(StrEnum):
|
|
"""Optional role hint used for role-aware compatibility checks."""
|
|
|
|
STRATEGY = "strategy"
|
|
EXECUTION = "execution"
|
|
ESTIMATION = "estimation"
|
|
INVARIANT_RECONCILIATION = "invariant_reconciliation"
|
|
REVIEW = "review"
|
|
|
|
|
|
def _coerce_role_hint(value: object) -> RoleHint | None:
|
|
"""Coerce role hint values from raw payloads into ``RoleHint``.
|
|
|
|
Accepts enum values directly and string inputs case-insensitively.
|
|
Returns ``None`` for non-string/unrecognized values.
|
|
"""
|
|
if isinstance(value, RoleHint):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
return RoleHint(value.lower())
|
|
except ValueError:
|
|
logger.warning("Unrecognized role_hint value in actor config: %r", value)
|
|
return None
|
|
return None
|
|
|
|
|
|
def _coerce_context_view(value: object) -> str | None:
|
|
"""Coerce context_view values while preserving unrecognized strings."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, str):
|
|
normalized = value.lower()
|
|
if normalized in {"strategist", "executor", "reviewer", "full"}:
|
|
return normalized
|
|
return value
|
|
enum_value = getattr(value, "value", None)
|
|
if isinstance(enum_value, str):
|
|
return enum_value.lower()
|
|
return None
|
|
|
|
|
|
def actor_role_warnings(config: object) -> list[str]:
|
|
"""Return non-fatal role compatibility warnings for actor configs."""
|
|
if isinstance(config, dict):
|
|
role_hint = _coerce_role_hint(config.get("role_hint"))
|
|
context_view = _coerce_context_view(config.get("context_view"))
|
|
|
|
response_format = config.get("response_format")
|
|
if not isinstance(response_format, dict):
|
|
nested = config.get("config")
|
|
if isinstance(nested, dict) and isinstance(
|
|
nested.get("response_format"), dict
|
|
):
|
|
response_format = nested.get("response_format")
|
|
else:
|
|
response_format = None
|
|
else:
|
|
role_hint = _coerce_role_hint(getattr(config, "role_hint", None))
|
|
context_view = _coerce_context_view(getattr(config, "context_view", None))
|
|
response_format = getattr(config, "response_format", None)
|
|
|
|
if role_hint != RoleHint.ESTIMATION:
|
|
return []
|
|
|
|
warnings: list[str] = []
|
|
if not isinstance(response_format, dict) or not response_format:
|
|
warnings.append(
|
|
"Estimation actors should define 'response_format' for structured output."
|
|
)
|
|
if context_view not in (None, "strategist"):
|
|
warnings.append(
|
|
"Estimation actors should use context_view 'strategist' "
|
|
"for planning context."
|
|
)
|
|
return warnings
|
|
|
|
|
|
__all__ = ["RoleHint", "actor_role_warnings"]
|