From 2e86917081559ee3e1d0970c0ecbe310fdd0b69c Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Tue, 1 Sep 2026 17:09:40 +0800 Subject: [PATCH 1/2] refactor: extract shared runtime foundation --- README.md | 8 +- agent_core/events.py | 24 ++++ agent_core/execution_context.py | 146 ++++++++++++++++++++++++ agent_core/runtime/events.py | 55 +++++++++ agent_core/runtime/tool_permission.py | 110 ++++++++++++++++++ agent_core/types.py | 63 +++++++++++ tests/test_runtime_foundation.py | 156 ++++++++++++++++++++++++++ 7 files changed, 561 insertions(+), 1 deletion(-) create mode 100644 agent_core/events.py create mode 100644 agent_core/execution_context.py create mode 100644 agent_core/runtime/events.py create mode 100644 agent_core/runtime/tool_permission.py create mode 100644 agent_core/types.py create mode 100644 tests/test_runtime_foundation.py diff --git a/README.md b/README.md index e569daa..24f18c2 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,10 @@ Version `0.1.x` contains the converged foundation layer: - message trimming; - bounded LLM summaries, projected-token triggers, and tiered compaction; - session-isolated, content-addressed spill storage for recoverable tool output; +- product-neutral identity/status types, kernel events, and task-local execution + context propagation; +- failure-isolated async event dispatch and composable fail-closed tool + permission policy; - provider-neutral LLM response, stream, and client contracts; - loop configuration, lifecycle contexts, observer protocol, intervention merging, and observer dispatch helpers. @@ -126,7 +130,9 @@ edit both products' core copies, that is evidence it belongs here. execution, and `agent_loop` orchestration. 5. **Context management Phase 2** (complete in AgentCore): calibrated trigger, bounded LLM summary, tier selection, observability, and spill storage. -6. Remove product compatibility facades after downstream imports have moved to +6. **Runtime foundation** (complete in AgentCore): task identity/status, + execution ContextVars, kernel events, event dispatch, and tool permissions. +7. Remove product compatibility facades after downstream imports have moved to `agent_core`. Each slice must leave product CI green and must not depend on a floating branch. diff --git a/agent_core/events.py b/agent_core/events.py new file mode 100644 index 0000000..d136221 --- /dev/null +++ b/agent_core/events.py @@ -0,0 +1,24 @@ +"""Product-neutral kernel event identifiers.""" + +from __future__ import annotations + +from enum import StrEnum + + +class EventType(StrEnum): + """Framework-mechanical events shared by all products.""" + + TASK_CREATED = "task_created" + TASK_STATUS_CHANGED = "task_status_changed" + PHASE_TRANSITION = "phase_transition" + AGENT_ACTION = "agent_action" + AGENT_MESSAGE = "agent_message" + AGENT_TOOL_CALL = "agent_tool_call" + TOOL_CALLED = "tool_called" + TOOL_RESULT = "tool_result" + ERROR = "error" + REPORT_GENERATED = "report_generated" + WORKING_MEMORY_SNAPSHOT = "working_memory_snapshot" + + +__all__ = ["EventType"] diff --git a/agent_core/execution_context.py b/agent_core/execution_context.py new file mode 100644 index 0000000..d72fa0d --- /dev/null +++ b/agent_core/execution_context.py @@ -0,0 +1,146 @@ +"""Task-local execution identity shared across LLM and tool calls.""" + +from __future__ import annotations + +from collections.abc import Generator +from contextlib import contextmanager +from contextvars import ContextVar, Token +from dataclasses import dataclass, field +from typing import Any, cast + +from agent_core.types import new_prompt_id, new_session_id, new_step_id + + +@dataclass +class ExecutionScope: + """Runtime execution scope for the current phase.""" + + task_id: str = "" + phase_id: str = "" + role_id: str = "" + metadata: dict[str, Any] = field(default_factory=dict[str, Any]) + + +_CURRENT_SCOPE: ContextVar[ExecutionScope | None] = ContextVar( + "agent_core_execution_scope", default=None +) +_CURRENT_TOOL_CALL_ID: ContextVar[str] = ContextVar( + "agent_core_current_tool_call_id", default="" +) +_CURRENT_TOOL_BUDGET: ContextVar[float | None] = ContextVar( + "agent_core_current_tool_budget", default=None +) +_CHAIN_FALLBACK_ACTIVE: ContextVar[bool] = ContextVar( + "agent_core_chain_fallback_active", default=False +) + + +def normalize_execution_context(value: Any) -> dict[str, Any]: + """Return an independent mutable execution-context mapping.""" + if isinstance(value, dict): + return dict(cast("dict[str, Any]", value)) + return {} + + +def build_execution_scope( + *, + task_id: str, + phase_id: str, + role_id: str, + state: dict[str, Any] | None = None, +) -> ExecutionScope: + """Build a scope from task identity and optional persisted state.""" + persisted = state.get("execution_context") if state is not None else None + metadata = normalize_execution_context(persisted) + metadata.setdefault("agent_id", role_id) + return ExecutionScope( + task_id=task_id, + phase_id=phase_id, + role_id=role_id, + metadata=metadata, + ) + + +def set_current_execution_scope(scope: ExecutionScope) -> Token[ExecutionScope | None]: + return _CURRENT_SCOPE.set(scope) + + +def get_current_execution_scope() -> ExecutionScope | None: + return _CURRENT_SCOPE.get() + + +def reset_current_execution_scope(token: Token[ExecutionScope | None]) -> None: + _CURRENT_SCOPE.reset(token) + + +def set_current_tool_call_id(tool_call_id: str) -> Token[str]: + return _CURRENT_TOOL_CALL_ID.set(tool_call_id) + + +def get_current_tool_call_id() -> str: + return _CURRENT_TOOL_CALL_ID.get() + + +def reset_current_tool_call_id(token: Token[str]) -> None: + _CURRENT_TOOL_CALL_ID.reset(token) + + +def set_current_tool_budget(seconds: float | None) -> Token[float | None]: + """Publish the outer loop budget for the active tool task.""" + return _CURRENT_TOOL_BUDGET.set(seconds) + + +def get_current_tool_budget() -> float | None: + return _CURRENT_TOOL_BUDGET.get() + + +def reset_current_tool_budget(token: Token[float | None]) -> None: + _CURRENT_TOOL_BUDGET.reset(token) + + +def chain_fallback_active() -> bool: + """Whether an outer provider-chain runner can rotate this attempt.""" + return _CHAIN_FALLBACK_ACTIVE.get() + + +@contextmanager +def chain_fallback_scope() -> Generator[None, None, None]: + """Mark this context as running beneath an outer provider chain.""" + token = _CHAIN_FALLBACK_ACTIVE.set(True) + try: + yield + finally: + _CHAIN_FALLBACK_ACTIVE.reset(token) + + +def ensure_trace_metadata( + metadata: dict[str, Any], + *, + default_step_id: str | None = None, + refresh_prompt_id: bool = False, +) -> dict[str, Any]: + """Ensure stable session, step, and prompt identifiers exist.""" + metadata.setdefault("session_id", str(new_session_id())) + metadata.setdefault("step_id", default_step_id or str(new_step_id())) + if refresh_prompt_id or not metadata.get("prompt_id"): + metadata["prompt_id"] = str(new_prompt_id()) + return metadata + + +__all__ = [ + "ExecutionScope", + "build_execution_scope", + "chain_fallback_active", + "chain_fallback_scope", + "ensure_trace_metadata", + "get_current_execution_scope", + "get_current_tool_budget", + "get_current_tool_call_id", + "normalize_execution_context", + "reset_current_execution_scope", + "reset_current_tool_budget", + "reset_current_tool_call_id", + "set_current_execution_scope", + "set_current_tool_budget", + "set_current_tool_call_id", +] diff --git a/agent_core/runtime/events.py b/agent_core/runtime/events.py new file mode 100644 index 0000000..e3673d5 --- /dev/null +++ b/agent_core/runtime/events.py @@ -0,0 +1,55 @@ +"""Async pub/sub for portable runtime events.""" + +from __future__ import annotations + +import asyncio +import logging +from collections import defaultdict +from collections.abc import Awaitable, Callable +from enum import Enum +from typing import Any + +logger = logging.getLogger(__name__) + +Handler = Callable[[dict[str, Any]], Awaitable[None]] + + +def _coerce_key(event_type: str | Enum) -> str: + if isinstance(event_type, Enum): + value = getattr(event_type, "value", None) + if isinstance(value, str): + return value + return str(value) + return str(event_type) + + +class EventBus: + """Small failure-isolated async event bus.""" + + def __init__(self) -> None: + self._handlers: dict[str, list[Handler]] = defaultdict(list) + self._global_handlers: list[Handler] = [] + + def subscribe(self, event_type: str | Enum, handler: Handler) -> None: + self._handlers[_coerce_key(event_type)].append(handler) + + def subscribe_all(self, handler: Handler) -> None: + self._global_handlers.append(handler) + + async def publish(self, event_type: str | Enum, payload: dict[str, Any]) -> None: + key = _coerce_key(event_type) + handlers = [*self._handlers.get(key, ()), *self._global_handlers] + if not handlers: + return + event_data = {"event_type": key, **payload} + await asyncio.gather(*(self._safe_call(handler, event_data) for handler in handlers)) + + @staticmethod + async def _safe_call(handler: Handler, data: dict[str, Any]) -> None: + try: + await handler(data) + except Exception: + logger.exception("Event handler failed: %s", getattr(handler, "__name__", repr(handler))) + + +__all__ = ["EventBus", "Handler"] diff --git a/agent_core/runtime/tool_permission.py b/agent_core/runtime/tool_permission.py new file mode 100644 index 0000000..e36a5bf --- /dev/null +++ b/agent_core/runtime/tool_permission.py @@ -0,0 +1,110 @@ +"""Composable, fail-closed tool permission policy.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, cast + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ToolPermissionContext: + """Allowlist and denylist constraints for tool names.""" + + allow_names: frozenset[str] | None = None + deny_names: frozenset[str] = frozenset() + deny_prefixes: tuple[str, ...] = () + + def blocks(self, tool_name: str) -> bool: + lowered = tool_name.lower() + return lowered in self.deny_names or any( + lowered.startswith(prefix) for prefix in self.deny_prefixes + ) + + def allows(self, tool_name: str) -> bool: + lowered = tool_name.lower() + if self.blocks(lowered): + return False + return self.allow_names is None or lowered in self.allow_names + + def filter(self, tool_names: set[str]) -> set[str]: + return {name for name in tool_names if self.allows(name)} + + def is_empty(self) -> bool: + return self.allow_names is None and not self.deny_names and not self.deny_prefixes + + def merge(self, other: ToolPermissionContext) -> ToolPermissionContext: + """Combine two contexts so every restriction remains effective.""" + if self.allow_names is None: + allow = other.allow_names + elif other.allow_names is None: + allow = self.allow_names + else: + allow = self.allow_names & other.allow_names + return ToolPermissionContext( + allow_names=allow, + deny_names=self.deny_names | other.deny_names, + deny_prefixes=tuple(dict.fromkeys((*self.deny_prefixes, *other.deny_prefixes))), + ) + + @classmethod + def from_iterables( + cls, + *, + allow_names: set[str] | list[str] | tuple[str, ...] | None = None, + deny_names: set[str] | list[str] | tuple[str, ...] = (), + deny_prefixes: tuple[str, ...] | list[str] = (), + ) -> ToolPermissionContext: + normalized_allow = None + if allow_names is not None: + normalized_allow = frozenset(name.lower() for name in allow_names) + return cls( + allow_names=normalized_allow, + deny_names=frozenset(name.lower() for name in deny_names), + deny_prefixes=tuple(prefix.lower() for prefix in deny_prefixes), + ) + + +def from_config_map(mapping: Any | None) -> ToolPermissionContext: + """Build a policy from a ``{tool_name: bool}`` configuration map.""" + if not isinstance(mapping, dict) or not mapping: + return ToolPermissionContext() + + allow: set[str] = set() + deny: set[str] = set() + entries = cast("dict[object, object]", mapping) + for name, enabled in entries.items(): + if not isinstance(name, str): + continue + if enabled is True: + allow.add(name) + elif enabled is False: + deny.add(name) + else: + logger.warning( + "Tool policy entry %r=%r ignored: value must be a bare bool " + "(true/false), got %s — the tool is left UNCHANGED.", + name, + enabled, + type(enabled).__name__, + ) + return ToolPermissionContext.from_iterables( + allow_names=allow or None, + deny_names=deny, + ) + + +def from_execution_policy(policy: Any | None) -> ToolPermissionContext: + """Build a policy from an object exposing execution-policy attributes.""" + if policy is None: + return ToolPermissionContext() + return ToolPermissionContext.from_iterables( + allow_names=getattr(policy, "allow_tools", None), + deny_names=getattr(policy, "deny_tools", ()), + deny_prefixes=getattr(policy, "deny_tool_prefixes", ()), + ) + + +__all__ = ["ToolPermissionContext", "from_config_map", "from_execution_policy"] diff --git a/agent_core/types.py b/agent_core/types.py new file mode 100644 index 0000000..5147bcf --- /dev/null +++ b/agent_core/types.py @@ -0,0 +1,63 @@ +"""Product-neutral identity and task lifecycle types.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import NewType +from uuid import uuid4 + +TaskId = NewType("TaskId", str) +EventId = NewType("EventId", str) +SessionId = NewType("SessionId", str) +AgentSessionId = NewType("AgentSessionId", str) +PromptId = NewType("PromptId", str) +StepId = NewType("StepId", str) +AgentRoleId = NewType("AgentRoleId", str) + + +def new_task_id() -> TaskId: + return TaskId(uuid4().hex[:12]) + + +def new_event_id() -> EventId: + return EventId(uuid4().hex[:16]) + + +def new_session_id() -> SessionId: + return SessionId(uuid4().hex[:12]) + + +def new_prompt_id() -> PromptId: + return PromptId(uuid4().hex[:12]) + + +def new_step_id() -> StepId: + return StepId(uuid4().hex[:10]) + + +class TaskStatus(StrEnum): + """Portable task lifecycle states.""" + + CREATED = "created" + RUNNING = "running" + SUSPENDED = "suspended" + COMPLETED = "completed" + FAILED = "failed" + ABORTED = "aborted" + + +__all__ = [ + "AgentRoleId", + "AgentSessionId", + "EventId", + "PromptId", + "SessionId", + "StepId", + "TaskId", + "TaskStatus", + "new_event_id", + "new_prompt_id", + "new_session_id", + "new_step_id", + "new_task_id", +] diff --git a/tests/test_runtime_foundation.py b/tests/test_runtime_foundation.py new file mode 100644 index 0000000..986834f --- /dev/null +++ b/tests/test_runtime_foundation.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +import pytest + +from agent_core.events import EventType +from agent_core.execution_context import ( + ExecutionScope, + build_execution_scope, + chain_fallback_active, + chain_fallback_scope, + ensure_trace_metadata, + get_current_execution_scope, + get_current_tool_budget, + get_current_tool_call_id, + reset_current_execution_scope, + reset_current_tool_budget, + reset_current_tool_call_id, + set_current_execution_scope, + set_current_tool_budget, + set_current_tool_call_id, +) +from agent_core.runtime.events import EventBus +from agent_core.runtime.tool_permission import ( + ToolPermissionContext, + from_config_map, + from_execution_policy, +) +from agent_core.types import TaskStatus, new_event_id, new_prompt_id, new_session_id, new_step_id + + +def test_identity_shapes_and_string_event_contract() -> None: + assert len(new_event_id()) == 16 + assert len(new_session_id()) == 12 + assert len(new_prompt_id()) == 12 + assert len(new_step_id()) == 10 + assert TaskStatus.RUNNING == "running" + assert EventType.TOOL_RESULT == "tool_result" + + +def test_execution_scope_build_and_trace_metadata() -> None: + state = {"execution_context": {"session_id": "session", "custom": 1}} + scope = build_execution_scope(task_id="t", phase_id="p", role_id="r", state=state) + assert scope.metadata == {"session_id": "session", "custom": 1, "agent_id": "r"} + assert scope.metadata is not state["execution_context"] + + ensure_trace_metadata(scope.metadata, default_step_id="step") + first_prompt = scope.metadata["prompt_id"] + ensure_trace_metadata(scope.metadata, refresh_prompt_id=True) + assert scope.metadata["session_id"] == "session" + assert scope.metadata["step_id"] == "step" + assert scope.metadata["prompt_id"] != first_prompt + + +def test_execution_context_tokens_restore_nested_values() -> None: + assert get_current_execution_scope() is None + scope_token = set_current_execution_scope(ExecutionScope(task_id="task")) + call_token = set_current_tool_call_id("call") + budget_token = set_current_tool_budget(12.5) + try: + assert get_current_execution_scope() == ExecutionScope(task_id="task") + assert get_current_tool_call_id() == "call" + assert get_current_tool_budget() == 12.5 + assert not chain_fallback_active() + with chain_fallback_scope(): + assert chain_fallback_active() + assert not chain_fallback_active() + finally: + reset_current_tool_budget(budget_token) + reset_current_tool_call_id(call_token) + reset_current_execution_scope(scope_token) + + assert get_current_execution_scope() is None + assert get_current_tool_call_id() == "" + assert get_current_tool_budget() is None + + +@pytest.mark.asyncio +async def test_execution_context_is_isolated_between_tasks() -> None: + async def read(call_id: str) -> str: + token = set_current_tool_call_id(call_id) + try: + await asyncio.sleep(0) + return get_current_tool_call_id() + finally: + reset_current_tool_call_id(token) + + assert await asyncio.gather(read("a"), read("b")) == ["a", "b"] + + +@pytest.mark.asyncio +async def test_event_bus_dispatches_typed_and_global_handlers() -> None: + bus = EventBus() + seen: list[tuple[str, int]] = [] + + async def handler(event: dict[str, object]) -> None: + seen.append((str(event["event_type"]), int(event["value"]))) + + bus.subscribe(EventType.TOOL_RESULT, handler) + bus.subscribe_all(handler) + await bus.publish(EventType.TOOL_RESULT, {"value": 3}) + assert seen == [("tool_result", 3), ("tool_result", 3)] + + +@pytest.mark.asyncio +async def test_event_bus_isolates_handler_failures() -> None: + bus = EventBus() + completed = False + + async def broken(_event: dict[str, object]) -> None: + raise RuntimeError("boom") + + async def healthy(_event: dict[str, object]) -> None: + nonlocal completed + completed = True + + bus.subscribe("event", broken) + bus.subscribe("event", healthy) + await bus.publish("event", {}) + assert completed + + +def test_tool_permission_merge_is_fail_closed() -> None: + profile = ToolPermissionContext.from_iterables( + allow_names={"bash", "web_search"}, deny_prefixes=("internal_",) + ) + request = ToolPermissionContext.from_iterables( + allow_names={"bash", "python"}, deny_names={"bash"} + ) + merged = profile.merge(request) + assert merged.allow_names == frozenset({"bash"}) + assert not merged.allows("bash") + assert not merged.allows("internal_admin") + + +def test_config_map_requires_bare_booleans(caplog: pytest.LogCaptureFixture) -> None: + context = from_config_map({"bash": True, "web_search": False, "python": "false"}) + assert context.allow_names == frozenset({"bash"}) + assert context.deny_names == frozenset({"web_search"}) + assert "ignored" in caplog.text + + +@dataclass +class _Policy: + allow_tools: tuple[str, ...] = ("bash",) + deny_tools: tuple[str, ...] = ("web_search",) + deny_tool_prefixes: tuple[str, ...] = ("internal_",) + + +def test_execution_policy_is_structurally_consumed() -> None: + context = from_execution_policy(_Policy()) + assert context.allows("bash") + assert not context.allows("web_search") + assert not context.allows("internal_admin") From 72503db7b89bea5656798968b0d547b6432b3ddc Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Tue, 1 Sep 2026 17:44:21 +0800 Subject: [PATCH 2/2] fix(runtime): close the review findings on the runtime foundation tool_permission: - from_execution_policy now tolerates None-valued deny attributes, which crashed policy construction for the common `deny_tools: tuple | None` shape (getattr only falls back when the attribute is absent). - Normalize case in __post_init__ so direct construction no longer silently denies every tool a caller listed with mixed case. - Document that a single True entry in a config map makes the map an exhaustive allowlist, and lock that semantic with a test. - Widen filter() to Iterable[str]. execution_context: - normalize_execution_context deep-copies, so mutating nested scope metadata no longer writes through to persisted checkpoint state; it degrades to a shallow copy with a warning if a value is uncopyable. - role_id is authoritative for agent_id, so a stale value persisted by a previous phase no longer misattributes this role's work. runtime/events: - Give each handler its own payload copy (they interleave at every await) and let the published event_type win over one carried in the payload. - Add unsubscribe/unsubscribe_all/clear so a long-lived bus no longer retains every finished task's handler. Also re-export the new symbols from the package inits, matching the convention set by SpillStore/LLMClient, and add 11 regression tests. Co-Authored-By: Claude Opus 5 (1M context) --- agent_core/__init__.py | 7 ++ agent_core/execution_context.py | 35 ++++++- agent_core/runtime/__init__.py | 15 ++- agent_core/runtime/events.py | 49 ++++++++- agent_core/runtime/tool_permission.py | 71 ++++++++++--- tests/test_runtime_foundation.py | 145 +++++++++++++++++++++++++- 6 files changed, 297 insertions(+), 25 deletions(-) diff --git a/agent_core/__init__.py b/agent_core/__init__.py index 7c1287f..c182b4e 100644 --- a/agent_core/__init__.py +++ b/agent_core/__init__.py @@ -1,6 +1,8 @@ """Product-neutral building blocks for Apodex agent runtimes.""" from agent_core.errors import AgentCoreError +from agent_core.events import EventType +from agent_core.execution_context import ExecutionScope, build_execution_scope from agent_core.llm import LLMClient, LLMResponse, StreamDelta from agent_core.messages import ( Message, @@ -10,15 +12,20 @@ tool_msg, user_msg, ) +from agent_core.types import TaskStatus __all__ = [ "AgentCoreError", + "EventType", + "ExecutionScope", "LLMClient", "LLMResponse", "Message", "StreamDelta", + "TaskStatus", "ToolCall", "assistant_msg", + "build_execution_scope", "system_msg", "tool_msg", "user_msg", diff --git a/agent_core/execution_context.py b/agent_core/execution_context.py index d72fa0d..78aa42c 100644 --- a/agent_core/execution_context.py +++ b/agent_core/execution_context.py @@ -2,6 +2,8 @@ from __future__ import annotations +import copy +import logging from collections.abc import Generator from contextlib import contextmanager from contextvars import ContextVar, Token @@ -10,6 +12,8 @@ from agent_core.types import new_prompt_id, new_session_id, new_step_id +logger = logging.getLogger(__name__) + @dataclass class ExecutionScope: @@ -36,10 +40,27 @@ class ExecutionScope: def normalize_execution_context(value: Any) -> dict[str, Any]: - """Return an independent mutable execution-context mapping.""" - if isinstance(value, dict): - return dict(cast("dict[str, Any]", value)) - return {} + """Return a fully independent mutable execution-context mapping. + + The copy is deep: callers mutate scope metadata freely (including nested + dicts and lists), and persisted checkpoint state it was loaded from must + not change underneath them. + """ + if not isinstance(value, dict): + return {} + mapping = cast("dict[str, Any]", value) + try: + return copy.deepcopy(mapping) + except Exception: + # Persisted state is expected to be plain data, but an in-memory state + # dict may carry something uncopyable. Degrading to a shallow copy is + # better than failing to build the scope at all. + logger.warning( + "Execution context could not be deep-copied; falling back to a " + "shallow copy, so nested values stay shared with the source state.", + exc_info=True, + ) + return dict(mapping) def build_execution_scope( @@ -52,7 +73,11 @@ def build_execution_scope( """Build a scope from task identity and optional persisted state.""" persisted = state.get("execution_context") if state is not None else None metadata = normalize_execution_context(persisted) - metadata.setdefault("agent_id", role_id) + # ``role_id`` is authoritative for this scope: a resumed task may carry an + # ``agent_id`` written during a different phase, and keeping it would + # attribute this role's work to the previous agent. + if role_id or "agent_id" not in metadata: + metadata["agent_id"] = role_id return ExecutionScope( task_id=task_id, phase_id=phase_id, diff --git a/agent_core/runtime/__init__.py b/agent_core/runtime/__init__.py index 2801880..50b41d7 100644 --- a/agent_core/runtime/__init__.py +++ b/agent_core/runtime/__init__.py @@ -1,5 +1,18 @@ """Shared runtime implementations.""" +from agent_core.runtime.events import EventBus, Handler from agent_core.runtime.spill import SpillStore +from agent_core.runtime.tool_permission import ( + ToolPermissionContext, + from_config_map, + from_execution_policy, +) -__all__ = ["SpillStore"] +__all__ = [ + "EventBus", + "Handler", + "SpillStore", + "ToolPermissionContext", + "from_config_map", + "from_execution_policy", +] diff --git a/agent_core/runtime/events.py b/agent_core/runtime/events.py index e3673d5..b2dfbd0 100644 --- a/agent_core/runtime/events.py +++ b/agent_core/runtime/events.py @@ -24,7 +24,13 @@ def _coerce_key(event_type: str | Enum) -> str: class EventBus: - """Small failure-isolated async event bus.""" + """Small failure-isolated async event bus. + + Subscriptions are held strongly and never expire on their own. A bus shared + across tasks must be paired with :meth:`unsubscribe` (or :meth:`clear`) when + a subscriber goes away, or every finished task leaves its handler — and the + state that handler closes over — reachable forever. + """ def __init__(self) -> None: self._handlers: dict[str, list[Handler]] = defaultdict(list) @@ -36,13 +42,50 @@ def subscribe(self, event_type: str | Enum, handler: Handler) -> None: def subscribe_all(self, handler: Handler) -> None: self._global_handlers.append(handler) + def unsubscribe(self, event_type: str | Enum, handler: Handler) -> bool: + """Remove one typed subscription. Returns whether it was registered.""" + key = _coerce_key(event_type) + registered = self._handlers.get(key) + if not registered: + return False + try: + registered.remove(handler) + except ValueError: + return False + if not registered: + del self._handlers[key] + return True + + def unsubscribe_all(self, handler: Handler) -> bool: + """Remove one catch-all subscription. Returns whether it was registered.""" + try: + self._global_handlers.remove(handler) + except ValueError: + return False + return True + + def clear(self) -> None: + """Drop every subscription.""" + self._handlers.clear() + self._global_handlers.clear() + async def publish(self, event_type: str | Enum, payload: dict[str, Any]) -> None: key = _coerce_key(event_type) handlers = [*self._handlers.get(key, ()), *self._global_handlers] if not handlers: return - event_data = {"event_type": key, **payload} - await asyncio.gather(*(self._safe_call(handler, event_data) for handler in handlers)) + if "event_type" in payload and payload["event_type"] != key: + logger.warning( + "Event payload carried event_type=%r for a %r publish; the published type wins.", + payload["event_type"], + key, + ) + # The published type is authoritative, so it is applied last. + event_data = {**payload, "event_type": key} + # Each handler gets its own top-level copy: handlers run concurrently + # and interleave at every await, so a handler that mutates the payload + # must not change what its peers observe. + await asyncio.gather(*(self._safe_call(handler, dict(event_data)) for handler in handlers)) @staticmethod async def _safe_call(handler: Handler, data: dict[str, Any]) -> None: diff --git a/agent_core/runtime/tool_permission.py b/agent_core/runtime/tool_permission.py index e36a5bf..492785e 100644 --- a/agent_core/runtime/tool_permission.py +++ b/agent_core/runtime/tool_permission.py @@ -3,20 +3,41 @@ from __future__ import annotations import logging +from collections.abc import Iterable from dataclasses import dataclass from typing import Any, cast logger = logging.getLogger(__name__) +def _normalize_names(names: Iterable[str] | None) -> frozenset[str]: + return frozenset(name.lower() for name in names or ()) + + @dataclass(frozen=True) class ToolPermissionContext: - """Allowlist and denylist constraints for tool names.""" + """Allowlist and denylist constraints for tool names. + + Names are matched case-insensitively: every constraint is lower-cased on + construction, so building a context directly is as safe as going through + :meth:`from_iterables`. + """ allow_names: frozenset[str] | None = None deny_names: frozenset[str] = frozenset() deny_prefixes: tuple[str, ...] = () + def __post_init__(self) -> None: + # Normalizing here (rather than only in the factories) keeps direct + # construction from silently denying every tool the caller listed with + # mixed case, and tolerates any iterable being passed in. + if self.allow_names is not None: + object.__setattr__(self, "allow_names", _normalize_names(self.allow_names)) + object.__setattr__(self, "deny_names", _normalize_names(self.deny_names)) + object.__setattr__( + self, "deny_prefixes", tuple(prefix.lower() for prefix in self.deny_prefixes or ()) + ) + def blocks(self, tool_name: str) -> bool: lowered = tool_name.lower() return lowered in self.deny_names or any( @@ -29,7 +50,7 @@ def allows(self, tool_name: str) -> bool: return False return self.allow_names is None or lowered in self.allow_names - def filter(self, tool_names: set[str]) -> set[str]: + def filter(self, tool_names: Iterable[str]) -> set[str]: return {name for name in tool_names if self.allows(name)} def is_empty(self) -> bool: @@ -53,22 +74,32 @@ def merge(self, other: ToolPermissionContext) -> ToolPermissionContext: def from_iterables( cls, *, - allow_names: set[str] | list[str] | tuple[str, ...] | None = None, - deny_names: set[str] | list[str] | tuple[str, ...] = (), - deny_prefixes: tuple[str, ...] | list[str] = (), + allow_names: Iterable[str] | None = None, + deny_names: Iterable[str] | None = (), + deny_prefixes: Iterable[str] | None = (), ) -> ToolPermissionContext: - normalized_allow = None - if allow_names is not None: - normalized_allow = frozenset(name.lower() for name in allow_names) return cls( - allow_names=normalized_allow, - deny_names=frozenset(name.lower() for name in deny_names), - deny_prefixes=tuple(prefix.lower() for prefix in deny_prefixes), + allow_names=None if allow_names is None else _normalize_names(allow_names), + deny_names=_normalize_names(deny_names), + deny_prefixes=tuple(prefix.lower() for prefix in deny_prefixes or ()), ) def from_config_map(mapping: Any | None) -> ToolPermissionContext: - """Build a policy from a ``{tool_name: bool}`` configuration map.""" + """Build a policy from a ``{tool_name: bool}`` configuration map. + + The two boolean values are **not** symmetric toggles: + + * ``False`` adds the tool to the denylist and leaves every unlisted tool + alone. + * ``True`` adds the tool to the *allowlist* — and because an allowlist is + exhaustive by definition, a single ``True`` entry anywhere in the map + denies every tool the map does not name. + + So ``{"bash": True}`` means "bash and nothing else", not "additionally + enable bash". This is the fail-closed reading: to enable one tool without + revoking the rest, list the tools to remove with ``False`` instead. + """ if not isinstance(mapping, dict) or not mapping: return ToolPermissionContext() @@ -90,6 +121,12 @@ def from_config_map(mapping: Any | None) -> ToolPermissionContext: enabled, type(enabled).__name__, ) + if allow: + logger.debug( + "Tool policy map enabled %d tool(s) explicitly; this is an allowlist, " + "so every unlisted tool is now denied.", + len(allow), + ) return ToolPermissionContext.from_iterables( allow_names=allow or None, deny_names=deny, @@ -97,13 +134,17 @@ def from_config_map(mapping: Any | None) -> ToolPermissionContext: def from_execution_policy(policy: Any | None) -> ToolPermissionContext: - """Build a policy from an object exposing execution-policy attributes.""" + """Build a policy from an object exposing execution-policy attributes. + + Attributes that are absent *or* ``None`` fall back to "unconstrained", + since ``allow_tools: tuple[str, ...] | None = None`` is a common shape. + """ if policy is None: return ToolPermissionContext() return ToolPermissionContext.from_iterables( allow_names=getattr(policy, "allow_tools", None), - deny_names=getattr(policy, "deny_tools", ()), - deny_prefixes=getattr(policy, "deny_tool_prefixes", ()), + deny_names=getattr(policy, "deny_tools", None) or (), + deny_prefixes=getattr(policy, "deny_tool_prefixes", None) or (), ) diff --git a/tests/test_runtime_foundation.py b/tests/test_runtime_foundation.py index 986834f..95dd479 100644 --- a/tests/test_runtime_foundation.py +++ b/tests/test_runtime_foundation.py @@ -1,7 +1,9 @@ from __future__ import annotations import asyncio +import threading from dataclasses import dataclass +from typing import cast import pytest @@ -15,6 +17,7 @@ get_current_execution_scope, get_current_tool_budget, get_current_tool_call_id, + normalize_execution_context, reset_current_execution_scope, reset_current_tool_budget, reset_current_tool_call_id, @@ -96,7 +99,7 @@ async def test_event_bus_dispatches_typed_and_global_handlers() -> None: seen: list[tuple[str, int]] = [] async def handler(event: dict[str, object]) -> None: - seen.append((str(event["event_type"]), int(event["value"]))) + seen.append((str(event["event_type"]), cast("int", event["value"]))) bus.subscribe(EventType.TOOL_RESULT, handler) bus.subscribe_all(handler) @@ -154,3 +157,143 @@ def test_execution_policy_is_structurally_consumed() -> None: assert context.allows("bash") assert not context.allows("web_search") assert not context.allows("internal_admin") + + +@dataclass +class _NullPolicy: + """The common shape where every constraint is an optional tuple.""" + + allow_tools: tuple[str, ...] | None = None + deny_tools: tuple[str, ...] | None = None + deny_tool_prefixes: tuple[str, ...] | None = None + + +def test_execution_policy_tolerates_none_valued_attributes() -> None: + context = from_execution_policy(_NullPolicy()) + assert context.is_empty() + assert context.allows("bash") + + +def test_direct_construction_normalizes_case() -> None: + context = ToolPermissionContext( + allow_names=frozenset({"Bash", "WEB_SEARCH"}), + deny_names=frozenset({"Web_Search"}), + deny_prefixes=("Internal_",), + ) + assert context.allows("Bash") + assert context.allows("bash") + assert not context.allows("web_search") + assert not context.allows("INTERNAL_admin") + + +def test_permission_filter_accepts_any_iterable() -> None: + context = ToolPermissionContext.from_iterables(deny_names=["bash"]) + assert context.filter(frozenset({"bash", "python"})) == {"python"} + assert context.filter(["bash", "python"]) == {"python"} + + +def test_config_map_true_entry_is_an_exhaustive_allowlist() -> None: + """A single ``True`` denies every unlisted tool; ``False`` alone does not.""" + enabling = from_config_map({"bash": True}) + assert enabling.allows("bash") + assert not enabling.allows("web_search") + + disabling = from_config_map({"web_search": False}) + assert disabling.allows("bash") + assert not disabling.allows("web_search") + + +def test_execution_scope_metadata_is_deeply_independent() -> None: + persisted: dict[str, object] = {"labels": {"run": "first"}, "steps": [1]} + state = {"execution_context": persisted} + scope = build_execution_scope(task_id="t", phase_id="p", role_id="r", state=state) + + labels = cast("dict[str, object]", scope.metadata["labels"]) + labels["run"] = "second" + steps = cast("list[object]", scope.metadata["steps"]) + steps.append(2) + + assert persisted == {"labels": {"run": "first"}, "steps": [1]} + + +def test_execution_scope_role_id_overrides_stale_persisted_agent_id() -> None: + state = {"execution_context": {"agent_id": "previous_role"}} + scope = build_execution_scope( + task_id="t", phase_id="phase_2", role_id="current_role", state=state + ) + assert scope.metadata["agent_id"] == "current_role" + assert scope.metadata["agent_id"] == scope.role_id + + # With no authoritative role to apply, a persisted value is left alone. + kept = build_execution_scope( + task_id="t", phase_id="phase_2", role_id="", state={"execution_context": {"agent_id": "x"}} + ) + assert kept.metadata["agent_id"] == "x" + + +def test_normalize_execution_context_survives_uncopyable_values() -> None: + lock = threading.Lock() + normalized = normalize_execution_context({"lock": lock, "count": 1}) + assert normalized["count"] == 1 + assert normalized["lock"] is lock + + +@pytest.mark.asyncio +async def test_event_payload_is_not_shared_between_handlers() -> None: + bus = EventBus() + observed: list[dict[str, object]] = [] + + async def mutator(event: dict[str, object]) -> None: + await asyncio.sleep(0) + event.pop("value", None) + event["injected"] = True + + async def observer(event: dict[str, object]) -> None: + await asyncio.sleep(0) + observed.append(dict(event)) + + bus.subscribe("event", mutator) + bus.subscribe("event", observer) + await bus.publish("event", {"value": 3}) + + assert observed == [{"event_type": "event", "value": 3}] + + +@pytest.mark.asyncio +async def test_published_event_type_wins_over_payload() -> None: + bus = EventBus() + seen: list[str] = [] + + async def handler(event: dict[str, object]) -> None: + seen.append(str(event["event_type"])) + + bus.subscribe(EventType.TOOL_RESULT, handler) + await bus.publish(EventType.TOOL_RESULT, {"event_type": "spoofed"}) + assert seen == ["tool_result"] + + +@pytest.mark.asyncio +async def test_event_bus_subscriptions_can_be_removed() -> None: + bus = EventBus() + calls = 0 + + async def handler(_event: dict[str, object]) -> None: + nonlocal calls + calls += 1 + + bus.subscribe("event", handler) + bus.subscribe_all(handler) + await bus.publish("event", {}) + assert calls == 2 + + assert bus.unsubscribe("event", handler) + assert not bus.unsubscribe("event", handler) + assert bus.unsubscribe_all(handler) + assert not bus.unsubscribe_all(handler) + await bus.publish("event", {}) + assert calls == 2 + + bus.subscribe("event", handler) + bus.clear() + await bus.publish("event", {}) + assert calls == 2