mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-26 17:38:36 +00:00
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""Context-local state for delegate_task child execution.
|
|
|
|
The parent Hermes process may itself be a Kanban dispatcher worker with
|
|
HERMES_KANBAN_* variables in process env. delegate_task children run inside the
|
|
same Python process, but they are not dispatcher-owned Kanban workers. This
|
|
module lets code paths that resolve tool schemas or spawn subprocesses fail
|
|
closed for delegated children without mutating global os.environ for the parent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from contextvars import ContextVar
|
|
from typing import Iterator, Mapping, MutableMapping
|
|
|
|
_DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar(
|
|
"hermes_delegated_child_context",
|
|
default=False,
|
|
)
|
|
|
|
KANBAN_ENV_KEYS: tuple[str, ...] = (
|
|
"HERMES_KANBAN_TASK",
|
|
"HERMES_KANBAN_RUN_ID",
|
|
"HERMES_KANBAN_WORKSPACE",
|
|
"HERMES_KANBAN_WORKSPACES_ROOT",
|
|
"HERMES_KANBAN_CLAIM_LOCK",
|
|
"HERMES_KANBAN_BOARD",
|
|
"HERMES_KANBAN_DB",
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def delegated_child_context() -> Iterator[None]:
|
|
"""Mark the current execution context as a delegate_task child."""
|
|
token = _DELEGATED_CHILD_CONTEXT.set(True)
|
|
try:
|
|
yield
|
|
finally:
|
|
_DELEGATED_CHILD_CONTEXT.reset(token)
|
|
|
|
|
|
def is_delegated_child_context() -> bool:
|
|
"""Return True while code is running for a delegate_task child."""
|
|
return bool(_DELEGATED_CHILD_CONTEXT.get())
|
|
|
|
|
|
def scrub_kanban_env(env: Mapping[str, str] | MutableMapping[str, str]) -> dict[str, str]:
|
|
"""Return *env* with dispatcher-only Kanban variables removed."""
|
|
cleaned = dict(env)
|
|
for key in KANBAN_ENV_KEYS:
|
|
cleaned.pop(key, None)
|
|
return cleaned
|