cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(cli): add --skill flag to actor run command #971

Merged
hurui200320 merged 1 commits from feature/m4-actor-run-skill into master 2026-03-19 09:30:36 +00:00
Conversation 13 Commits 1 Files Changed 20
+2027 -493
20 changed files with 2027 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+7
View File
@@ -69,6 +69,13 @@
exist, causing `AttributeError` at runtime. (#745)
- Added Google/Gemini API key pattern (`AIzaSy...`) to secret redaction in
`redaction.py`. (#745)
- Added `--skill <SKILL>` repeatable flag to `agents actor run` and
`actor-run` CLI commands. The flag resolves named skills from the
Skill Registry at runtime and merges their tools into agents that
already have configured tools, enabling ad-hoc skill injection
without modifying YAML configuration. Skill resolution uses the
DI-provided `SkillService` singleton; unknown or invalid skill names
produce a clear error and exit code 2. (#887)
- Added `--execution-env-priority` flag to `agents plan use` command, accepting
`fallback` (default) or `override` to control execution environment routing
precedence per ADR-043. Includes `ExecutionEnvPriority` StrEnum on the domain
benchmarks/skill_cli_bench.py
+5 -3
View File
@@ -29,8 +29,10 @@ importlib.reload(cleveragents)
from typer.testing import CliRunner # noqa: E402
from cleveragents.application.services.skill_service import SkillService # noqa: E402
from cleveragents.cli.commands import skill as skill_mod # noqa: E402
from cleveragents.cli.commands.skill import app as skill_app # noqa: E402
from cleveragents.cli.commands.skill import ( # noqa: E402
_reset_skill_service,
app as skill_app,
)
from cleveragents.domain.models.core.skill import Skill # noqa: E402
_VALID_YAML = """\
@@ -48,7 +50,7 @@ _runner = CliRunner()
def _fresh_service() -> SkillService:
"""Create and install a fresh service for benchmarking."""
svc = SkillService()
skill_mod._service = svc
_reset_skill_service(svc)
return svc
features/actor_cli_coverage.feature
+30
View File
@@ -277,3 +277,33 @@ Feature: Actor CLI coverage
And I have an actor JSON config file
When I run actor run with clever agents error
Then the actor run should exit with error code 2
@coverage
Scenario: Run actor with single skill flag
Given an actor CLI runner
And I have an actor JSON config file
When I run actor run with a single skill flag
Then the actor run should pass skill names to the runtime
@coverage
Scenario: Run actor with multiple skill flags
Given an actor CLI runner
And I have an actor JSON config file
When I run actor run with multiple skill flags
Then the actor run should pass all skill names to the runtime
@coverage
Scenario: Run actor with unknown skill flag
Given an actor CLI runner
And I have an actor JSON config file
When I run actor run with an unknown skill flag
Then the actor run should exit with skill not found error
@coverage
Scenario: Run actor with skill and context flags combined
Given an actor CLI runner
And I have an actor JSON config file
When I run actor run with skill and context flags
Then the actor run should pass skill names to the runtime
And the context manager should have been instantiated
And the context manager exists should have been called
features/actor_run_cli_coverage.feature
+37
View File
@@ -47,3 +47,40 @@ Feature: Actor run command coverage
And I have an actor JSON config file
When I invoke the actor-run command with a clever agents exception
Then the actor run should exit with error code 2
@coverage
Scenario: Actor run command with single skill flag
Given an actor CLI runner
And I have an actor JSON config file
When I invoke the actor-run command with a single skill flag
Then the actor run should pass skill names to the runtime
@coverage
Scenario: Actor run command with multiple skill flags
Given an actor CLI runner
And I have an actor JSON config file
When I invoke the actor-run command with multiple skill flags
Then the actor run should pass all skill names to the runtime
@coverage
Scenario: Actor run command with unknown skill flag
Given an actor CLI runner
And I have an actor JSON config file
When I invoke the actor-run command with an unknown skill flag
Then the actor run should exit with skill not found error
@coverage
Scenario: Actor run command with skill and context flags combined
Given an actor CLI runner
And I have an actor JSON config file
When I invoke the actor-run command with skill and context flags
Then the actor run should pass skill names to the runtime
And the context manager should have been instantiated
And the context manager exists should have been called
@coverage
Scenario: Actor run command with duplicate skill names
Given an actor CLI runner
And I have an actor JSON config file
When I invoke the actor-run command with duplicate skill flags
Then the actor run should pass duplicate skill names to the runtime
features/application_container_coverage_boost.feature
+20
View File
@@ -10,12 +10,14 @@ Feature: Application Container coverage boost for _build_checkpoint_service, _bu
# _build_checkpoint_service (lines 187-195)
# -----------------------------------------------------------------
@coverage
Scenario: Build checkpoint service with in-memory database and no lifecycle service
When I build a checkpoint service with an in-memory database URL
Then the result should be a CheckpointService instance
And the checkpoint service should have a repository
And the checkpoint service should have no plan lifecycle service
@coverage
Scenario: Build checkpoint service with an explicit plan lifecycle service
Given a mock plan lifecycle service
When I build a checkpoint service with the mock plan lifecycle service
@@ -26,12 +28,14 @@ Feature: Application Container coverage boost for _build_checkpoint_service, _bu
# _build_trace_service (lines 204-211)
# -----------------------------------------------------------------
@coverage
Scenario: Build trace service with in-memory database using default settings
When I build a trace service with an in-memory database URL and no explicit settings
Then the result should be a TraceService instance
And the trace service should have a repository
And the trace service should have resolved settings from defaults
@coverage
Scenario: Build trace service with explicit settings
Given explicit application settings for trace service
When I build a trace service with the explicit settings
@@ -55,3 +59,19 @@ Feature: Application Container coverage boost for _build_checkpoint_service, _bu
When I resolve session_factory from the container with an in-memory database URL
Then the resolved session factory should be callable
And calling the resolved session factory should produce a Session
# -----------------------------------------------------------------
# _build_skill_service (happy path + fallback)
# -----------------------------------------------------------------
@coverage
Scenario: Build skill service with in-memory database succeeds
When I build a skill service with an in-memory database URL
Then the result should be a SkillService instance
And the skill service should have a repository
@coverage
Scenario: Build skill service falls back to in-memory on DB error
When I build a skill service with an invalid database URL
Then the result should be a SkillService instance
And the skill service should have no repository
features/reactive_application_coverage_boost.feature
+88
View File
@@ -3,34 +3,122 @@ Feature: Reactive application coverage boost
I want to exercise the remaining uncovered lines in reactive/application.py
So that code coverage is maximized
@coverage
Scenario: Logging adds a handler when root logger has no handlers
Given a reactive app with all root logger handlers removed
When I configure logging with verbose level 2
Then the root logger should have at least one handler
And the handler format should include the module name pattern
@coverage
Scenario: Ensure agent registered creates instance from config when agent not already present
Given a reactive app with a route referencing an agent not yet registered but present in config
When I register agents from config triggering deferred registration
Then the deferred config agent should be registered in the stream router
And the deferred config agent alias should also be registered
@coverage
Scenario: Initialize graph context refreshes stage order when stages are missing
Given a reactive app with a global context that has a partial stage order list
When I initialize the graph context
Then the stage order should be refreshed to the full default list
@coverage
Scenario: Initialize graph context resets writing stage when current value is invalid
Given a reactive app with a global context that has an invalid writing stage
When I initialize the graph context
Then the writing stage should be reset to intro
@coverage
Scenario: Follow chained edges returns immediately when next node is end
Given a reactive app configured for chained edge traversal to end node
When I follow chained edges starting from the end node
Then the chained edge result should be the current message with should_return true
@coverage
Scenario: Follow chained edges falls through when next node becomes falsy
Given a reactive app configured for chained edge traversal to a falsy node
When I follow chained edges starting from a node that chains to a falsy node
Then the chained edge result should be the current message with should_return false
@coverage
Scenario: Reactive app resolves skill names and stores tools
Given a reactive app with skill names resolved via a mock skill service
When I check the resolved skill tools
Then the app should have resolved skill tool entries
And the skill names property should match the input
@coverage
Scenario: Reactive app raises error for unknown skill name
Given a reactive app configured with an unknown skill name
When I attempt to create the app with the unknown skill
Then a CleverAgentsException should be raised with skill not found message
@coverage
Scenario: Reactive app merges skill tools into agent tool list
Given a reactive app with skill tools resolved and a config with agents
When agents are registered from config with skill tools
Then the agents should have skill tools merged into their tool list
@coverage
Scenario: Reactive app works without skill names
Given a reactive app created without any skill names
When I check the skill related properties
Then the skill names should be empty
And the resolved skill tools should be empty
@coverage
Scenario: Reactive app resolves skill with tool overrides
Given a reactive app with skill names resolved including overrides
When I check the resolved skill tools
Then the resolved tool entry should include overrides
@coverage
Scenario: Reactive app deduplicates skill names
Given a reactive app with duplicate skill names resolved via a mock skill service
When I check the resolved skill tools
Then the skill names should be deduplicated
And resolve_tools should be called once per unique skill
@coverage
Scenario: Reactive app raises error for empty string skill name
Given a reactive app configured with an empty string skill name
When I attempt to create the app with the empty skill
Then a CleverAgentsException should be raised for resolution failure
@coverage
Scenario: Reactive app skill injection skips LLM agents without tools
Given a reactive app with skill tools and a config with an LLM agent
When agents are registered from config with skill tools present
Then the LLM agent should remain a SimpleLLMAgent not a SimpleToolAgent
@coverage
Scenario: Reactive app raises error for skill resolution ValueError
Given a reactive app configured with a skill that triggers a ValueError
When I attempt to create the app with the ValueError skill
Then a CleverAgentsException should be raised with resolution failed message
@coverage
Scenario: Reactive app handles skill that resolves to zero tools
Given a reactive app with a skill that resolves to zero tools
When I check the resolved skill tools after zero tool resolution
Then the resolved skill tools should be empty
And the skill names should contain the zero-tool skill
@coverage
Scenario: Reactive app rejects skill name exceeding max length
Given a reactive app configured with an overly long skill name
When I attempt to create the app with the long skill name
Then a CleverAgentsException should be raised for invalid name format
@coverage
Scenario: Reactive app strips ANSI escape codes from skill names
Given a reactive app configured with a skill name containing ANSI codes
When I attempt to create the app with the ANSI skill name
Then a CleverAgentsException should be raised for invalid name format
@coverage
Scenario: Reactive app rejects skill name with disallowed characters
Given a reactive app configured with a skill name containing special characters
When I attempt to create the app with the special char skill name
Then a CleverAgentsException should be raised for invalid name format
features/skill_cli_coverage_boost.feature
+7 -22
View File
@@ -8,33 +8,18 @@ Feature: Skill CLI coverage boost
Given boost- a fresh skill CLI service
# -------------------------------------------------------------------
# _get_skill_service: DB initialization path (lines 82-100)
# _get_skill_service: delegates to DI container
# -------------------------------------------------------------------
Scenario: _get_skill_service creates service with DB when container is available
Given boost- the module-level _service is set to None
And boost- the DI container and DB components are mocked successfully
Scenario: _get_skill_service delegates to DI container skill_service
Given boost- the DI container returns a DB-backed SkillService
When boost- I call _get_skill_service
Then boost- the returned service should have a skill_repo
And boost- create_engine should have been called with the mock database URL
# -------------------------------------------------------------------
# _get_skill_service: exception fallback (lines 101-103)
# -------------------------------------------------------------------
Scenario: _get_skill_service falls back to in-memory when DB setup fails
Given boost- the module-level _service is set to None
And boost- the DI container import will raise an exception
When boost- I call _get_skill_service
Then boost- the returned service should be an in-memory SkillService
And boost- the returned service should have no skill_repo
Scenario: _get_skill_service falls back when create_engine raises
Given boost- the module-level _service is set to None
And boost- create_engine will raise an OperationalError
When boost- I call _get_skill_service
Then boost- the returned service should be an in-memory SkillService
And boost- the returned service should have no skill_repo
Scenario: _get_skill_service always delegates to container (no module cache)
Given boost- the DI container returns a DB-backed SkillService
When boost- I call _get_skill_service twice
Then boost- the container skill_service should be called twice
# -------------------------------------------------------------------
# tools non-rich format: agent_skill: source type (line 889)
features/steps/actor_cli_run_steps.py
+380
View File
@@ -658,3 +658,383 @@ def step_actor_run_exit_code_two(context):
def step_actor_run_allow_rxpy_flag(context):
call_kwargs = context.app_exec.run_single_shot.call_args.kwargs
assert call_kwargs.get("allow_rxpy_in_run_mode") is True
# ---------------------------------------------------------------------------
# --skill flag: actor.py run command
# ---------------------------------------------------------------------------
@when("I run actor run with a single skill flag")
def step_run_actor_with_single_skill(context):
context.prompt = "skill-single"
context.run_result = "skill response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
with patch(
"cleveragents.cli.commands.actor.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls:
context.result = context.runner.invoke(
actor_app,
[
"run",
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.expected_skill_names = ["local/web-tools"]
@when("I run actor run with multiple skill flags")
def step_run_actor_with_multiple_skills(context):
context.prompt = "skill-multi"
context.run_result = "multi skill response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
with patch(
"cleveragents.cli.commands.actor.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls:
context.result = context.runner.invoke(
actor_app,
[
"run",
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
"--skill",
"local/db-tools",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.expected_skill_names = ["local/web-tools", "local/db-tools"]
@when("I run actor run with an unknown skill flag")
def step_run_actor_with_unknown_skill(context):
"""Exercise the real error chain: ReactiveCleverAgentsApp.__init__
calls _resolve_skills(), which obtains SkillService from the DI
container. We mock only the container so the real constructor and
_resolve_skills() execute end-to-end.
"""
context.prompt = "skill-unknown"
mock_service = MagicMock()
mock_service.resolve_tools.side_effect = KeyError(
"Skill 'local/nonexistent' is not registered"
)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.result = context.runner.invoke(
actor_app,
[
"run",
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/nonexistent",
],
)
# ---------------------------------------------------------------------------
# --skill flag: actor_run.py run command
# ---------------------------------------------------------------------------
@when("I invoke the actor-run command with a single skill flag")
def step_invoke_actor_run_with_single_skill(context):
context.prompt = "actor-run skill-single"
context.run_result = "actor-run skill response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
with patch(
"cleveragents.cli.commands.actor_run.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls:
context.result = context.runner.invoke(
actor_run_app,
[
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.expected_skill_names = ["local/web-tools"]
@when("I invoke the actor-run command with multiple skill flags")
def step_invoke_actor_run_with_multiple_skills(context):
context.prompt = "actor-run skill-multi"
context.run_result = "actor-run multi skill response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
with patch(
"cleveragents.cli.commands.actor_run.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls:
context.result = context.runner.invoke(
actor_run_app,
[
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
"--skill",
"local/db-tools",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.expected_skill_names = ["local/web-tools", "local/db-tools"]
@when("I invoke the actor-run command with an unknown skill flag")
def step_invoke_actor_run_with_unknown_skill(context):
"""Exercise the real error chain: ReactiveCleverAgentsApp.__init__
calls _resolve_skills(), which obtains SkillService from the DI
container. We mock only the container so the real constructor and
_resolve_skills() execute end-to-end.
"""
context.prompt = "actor-run skill-unknown"
mock_service = MagicMock()
mock_service.resolve_tools.side_effect = KeyError(
"Skill 'local/nonexistent' is not registered"
)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.result = context.runner.invoke(
actor_run_app,
[
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/nonexistent",
],
)
# ---------------------------------------------------------------------------
# --skill flag: shared assertions
# ---------------------------------------------------------------------------
@then("the actor run should pass skill names to the runtime")
@then("the actor run should pass all skill names to the runtime")
def step_actor_run_pass_skill_names(context):
assert context.result.exit_code == 0
call_kwargs = context.mock_cls.call_args.kwargs
assert call_kwargs.get("skill_names") == context.expected_skill_names
@then("the actor run should exit with skill not found error")
def step_actor_run_skill_not_found(context):
assert context.result.exit_code == 2
assert "Error: Skill 'local/nonexistent' not found in registry" in (
context.result.output
)
@then("the context manager should have been instantiated")
def step_context_manager_instantiated(context):
context.mock_ctx_cls.assert_called_once()
@then("the context manager exists should have been called")
def step_context_manager_exists_called(context):
context.ctx_mgr.exists.assert_called_once()
# ---------------------------------------------------------------------------
# --skill + --context combined flag test
# ---------------------------------------------------------------------------
@when("I run actor run with skill and context flags")
def step_run_actor_with_skill_and_context(context):
context.prompt = "skill-context-combo"
context.run_result = "skill context response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
ctx_mgr = _make_context_manager(
global_context={"cached": "value"},
exists=True,
)
with (
patch(
"cleveragents.cli.commands.actor.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls,
patch(
"cleveragents.cli.commands.actor.ContextManager",
return_value=ctx_mgr,
) as mock_ctx_cls,
):
context.result = context.runner.invoke(
actor_app,
[
"run",
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
"--context",
"test-session",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.mock_ctx_cls = mock_ctx_cls
context.ctx_mgr = ctx_mgr
context.expected_skill_names = ["local/web-tools"]
# ---------------------------------------------------------------------------
# --skill + --context combined flag test for actor_run.py (M6)
# ---------------------------------------------------------------------------
@when("I invoke the actor-run command with skill and context flags")
def step_invoke_actor_run_with_skill_and_context(context):
context.prompt = "actor-run skill-context-combo"
context.run_result = "actor-run skill context response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
ctx_mgr = _make_context_manager(
global_context={"cached": "value"},
exists=True,
)
with (
patch(
"cleveragents.cli.commands.actor_run.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls,
patch(
"cleveragents.cli.commands.actor_run.ContextManager",
return_value=ctx_mgr,
) as mock_ctx_cls,
):
context.result = context.runner.invoke(
actor_run_app,
[
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
"--context",
"test-session",
],
)
context.app_exec = app_exec
context.mock_cls = mock_cls
context.mock_ctx_cls = mock_ctx_cls
context.ctx_mgr = ctx_mgr
context.expected_skill_names = ["local/web-tools"]
# ---------------------------------------------------------------------------
# Duplicate skill deduplication test (M7)
# ---------------------------------------------------------------------------
@when("I invoke the actor-run command with duplicate skill flags")
def step_invoke_actor_run_with_duplicate_skills(context):
context.prompt = "actor-run skill-dedup"
context.run_result = "actor-run dedup response"
app_exec = _make_app(
result=context.run_result,
config_global_context={"existing": "value"},
)
with patch(
"cleveragents.cli.commands.actor_run.ReactiveCleverAgentsApp",
return_value=app_exec,
) as mock_cls:
context.result = context.runner.invoke(
actor_run_app,
[
"--config",
str(context.actor_config_path),
"--prompt",
context.prompt,
"--skill",
"local/web-tools",
"--skill",
"local/web-tools",
],
)
context.mock_cls = mock_cls
@then("the actor run should pass duplicate skill names to the runtime")
def step_actor_run_pass_duplicate_skill_names(context):
assert context.result.exit_code == 0
call_kwargs = context.mock_cls.call_args.kwargs
# CLI passes duplicate names through unchanged; deduplication
# happens inside ReactiveCleverAgentsApp.__init__ via dict.fromkeys.
assert call_kwargs.get("skill_names") == [
"local/web-tools",
"local/web-tools",
]
features/steps/application_container_coverage_boost_steps.py
+53
View File
@@ -17,10 +17,12 @@ from behave import given, then, when
from cleveragents.application.container import (
_build_checkpoint_service,
_build_session_factory,
_build_skill_service,
_build_trace_service,
reset_container,
)
from cleveragents.application.services.checkpoint_service import CheckpointService
from cleveragents.application.services.skill_service import SkillService
from cleveragents.application.services.trace_service import TraceService
from cleveragents.infrastructure.database.llm_trace_repository import (
LLMTraceRepository,
@@ -266,3 +268,54 @@ def step_verify_resolved_factory_produces_session(context: Any) -> None:
)
finally:
session.close()
# -------------------------------------------------------------------
# _build_skill_service scenarios
# -------------------------------------------------------------------
@when("I build a skill service with an in-memory database URL")
def step_build_skill_service_default(context):
"""Call _build_skill_service with a valid in-memory database URL."""
context.boost_skill_svc = _build_skill_service(_IN_MEMORY_URL)
@when("I build a skill service with an invalid database URL")
def step_build_skill_service_invalid(context):
"""Mock create_engine to raise OperationalError, triggering the fallback."""
from unittest.mock import patch
from sqlalchemy.exc import OperationalError
with patch(
"sqlalchemy.create_engine",
side_effect=OperationalError("mock", {}, Exception("DB unavailable")),
):
context.boost_skill_svc = _build_skill_service(
"sqlite:////nonexistent/path/db.sqlite"
)
@then("the result should be a SkillService instance")
def step_verify_skill_service_type(context):
"""Assert the returned object is a SkillService."""
assert isinstance(context.boost_skill_svc, SkillService), (
f"Expected SkillService, got {type(context.boost_skill_svc).__name__}"
)
@then("the skill service should have a repository")
def step_verify_skill_has_repository(context):
"""Assert the service was initialised with a SkillRepository."""
assert context.boost_skill_svc._skill_repo is not None, (
"Expected a SkillRepository, got None"
)
@then("the skill service should have no repository")
def step_verify_skill_no_repository(context):
"""Assert the fallback service has no DB-backed repository."""
assert context.boost_skill_svc._skill_repo is None, (
"Expected no SkillRepository for in-memory fallback"
)
features/steps/reactive_application_coverage_boost_steps.py
+537
View File
@@ -13,6 +13,7 @@ from __future__ import annotations
import logging
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
@@ -301,3 +302,539 @@ def step_chained_falsy_result(context: Context) -> None:
f"Expected 'test_message' but got '{context.chained_message}'"
)
assert context.chained_should_return is False, "Expected should_return=False"
# ---------------------------------------------------------------------------
# Scenario: Reactive app resolves skill names and stores tools
# ---------------------------------------------------------------------------
@given("a reactive app with skill names resolved via a mock skill service")
def step_app_with_skill_names_resolved(context: Context) -> None:
from cleveragents.domain.models.core.skill import ResolvedToolEntry
mock_service = MagicMock()
mock_skill = MagicMock()
mock_entries = [
ResolvedToolEntry(
name="tool-a",
source_skill="local/web-tools",
is_inline=False,
),
ResolvedToolEntry(
name="tool-b",
source_skill="local/web-tools",
is_inline=True,
),
]
mock_service.resolve_tools.return_value = (mock_skill, mock_entries)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/web-tools"],
)
context.mock_service = mock_service
@when("I check the resolved skill tools")
def step_check_resolved_skill_tools(context: Context) -> None:
context.resolved_tools = context.app.resolved_skill_tools
context.skill_names_result = context.app.skill_names
@then("the app should have resolved skill tool entries")
def step_app_has_resolved_tools(context: Context) -> None:
assert len(context.resolved_tools) == 2, (
f"Expected 2 resolved tools but got {len(context.resolved_tools)}"
)
assert context.resolved_tools[0]["name"] == "tool-a"
assert context.resolved_tools[0]["source_skill"] == "local/web-tools"
assert context.resolved_tools[0]["is_inline"] is False
assert context.resolved_tools[0]["operation"] == "identity"
assert context.resolved_tools[0]["skill_source"] == "local/web-tools"
assert context.resolved_tools[1]["name"] == "tool-b"
assert context.resolved_tools[1]["is_inline"] is True
assert context.resolved_tools[1]["skill_source"] == "local/web-tools"
@then("the skill names property should match the input")
def step_skill_names_match(context: Context) -> None:
assert context.skill_names_result == ["local/web-tools"]
# ---------------------------------------------------------------------------
# Scenario: Reactive app raises error for unknown skill name
# ---------------------------------------------------------------------------
@given("a reactive app configured with an unknown skill name")
def step_app_with_unknown_skill(context: Context) -> None:
mock_service = MagicMock()
mock_service.resolve_tools.side_effect = KeyError(
"Skill 'local/bad' is not registered"
)
context.mock_service = mock_service
@when("I attempt to create the app with the unknown skill")
def step_attempt_create_app_with_unknown_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
mock_container = MagicMock()
mock_container.skill_service.return_value = context.mock_service
context.raised_exception = None
try:
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
ReactiveCleverAgentsApp(skill_names=["local/bad"])
except CleverAgentsException as exc:
context.raised_exception = exc
@then("a CleverAgentsException should be raised with skill not found message")
def step_exception_raised_with_message(context: Context) -> None:
assert context.raised_exception is not None, "Expected CleverAgentsException"
assert "local/bad" in str(context.raised_exception)
assert "not found" in str(context.raised_exception)
# ---------------------------------------------------------------------------
# Scenario: Reactive app merges skill tools into agent tool list
# ---------------------------------------------------------------------------
@given("a reactive app with skill tools resolved and a config with agents")
def step_app_with_skill_tools_and_config(context: Context) -> None:
from cleveragents.domain.models.core.skill import ResolvedToolEntry
mock_service = MagicMock()
mock_skill = MagicMock()
mock_entries = [
ResolvedToolEntry(
name="skill-tool-x",
source_skill="local/test-skill",
is_inline=False,
),
]
mock_service.resolve_tools.return_value = (mock_skill, mock_entries)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/test-skill"],
)
@when("agents are registered from config with skill tools")
def step_register_agents_with_skill_tools(context: Context) -> None:
agent_cfg = AgentConfig(
name="test_actor",
type="custom",
config={"tools": [{"operation": "uppercase"}]},
)
context.app.config = ReactiveConfig(
agents={"test_actor": agent_cfg},
routes={},
)
context.app._register_agents_from_config() # pylint: disable=protected-access
@then("the agents should have skill tools merged into their tool list")
def step_agents_have_merged_tools(context: Context) -> None:
from cleveragents.reactive.stream_router import SimpleToolAgent
agent = context.app.stream_router.agents.get("test_actor")
assert agent is not None, "Expected test_actor to be registered"
assert isinstance(agent, SimpleToolAgent), "Expected SimpleToolAgent instance"
# Original tool + 1 skill tool = 2
assert len(agent.tools) == 2, f"Expected 2 tools but got {len(agent.tools)}"
assert agent.tools[0] == {"operation": "uppercase"}
assert agent.tools[1]["name"] == "skill-tool-x"
assert agent.tools[1]["skill_source"] == "local/test-skill"
# ---------------------------------------------------------------------------
# Scenario: Reactive app works without skill names
# ---------------------------------------------------------------------------
@given("a reactive app created without any skill names")
def step_app_without_skills(context: Context) -> None:
context.app = ReactiveCleverAgentsApp()
@when("I check the skill related properties")
def step_check_skill_properties(context: Context) -> None:
context.skill_names_result = context.app.skill_names
context.resolved_tools_result = context.app.resolved_skill_tools
@then("the skill names should be empty")
def step_skill_names_empty(context: Context) -> None:
assert context.skill_names_result == [], (
f"Expected empty skill names but got {context.skill_names_result}"
)
@then("the resolved skill tools should be empty")
def step_resolved_tools_empty(context: Context) -> None:
assert context.resolved_tools_result == [], (
f"Expected empty resolved tools but got {context.resolved_tools_result}"
)
# ---------------------------------------------------------------------------
# Scenario: Reactive app resolves skill with tool overrides
# ---------------------------------------------------------------------------
@given("a reactive app with skill names resolved including overrides")
def step_app_with_overrides(context: Context) -> None:
from cleveragents.domain.models.core.skill import ResolvedToolEntry
mock_service = MagicMock()
mock_skill = MagicMock()
mock_entries = [
ResolvedToolEntry(
name="tool-with-overrides",
source_skill="local/override-skill",
is_inline=False,
overrides={"timeout": 600, "retries": 3},
),
]
mock_service.resolve_tools.return_value = (mock_skill, mock_entries)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/override-skill"],
)
@then("the resolved tool entry should include overrides")
def step_resolved_tool_has_overrides(context: Context) -> None:
assert len(context.resolved_tools) == 1, (
f"Expected 1 resolved tool but got {len(context.resolved_tools)}"
)
tool = context.resolved_tools[0]
assert tool["name"] == "tool-with-overrides"
assert tool["skill_source"] == "local/override-skill"
assert "overrides" in tool, "Expected 'overrides' key in tool dict"
assert tool["overrides"] == {"timeout": 600, "retries": 3}
# ---------------------------------------------------------------------------
# Scenario: Reactive app deduplicates skill names (M7)
# ---------------------------------------------------------------------------
@given("a reactive app with duplicate skill names resolved via a mock skill service")
def step_app_with_duplicate_skills(context: Context) -> None:
from cleveragents.domain.models.core.skill import ResolvedToolEntry
mock_service = MagicMock()
mock_skill = MagicMock()
mock_entries = [
ResolvedToolEntry(
name="tool-a",
source_skill="local/web-tools",
is_inline=False,
),
]
mock_service.resolve_tools.return_value = (mock_skill, mock_entries)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/web-tools", "local/web-tools", "local/web-tools"],
)
context.mock_service = mock_service
@then("the skill names should be deduplicated")
def step_skill_names_deduplicated(context: Context) -> None:
assert context.app.skill_names == ["local/web-tools"], (
f"Expected deduplicated list but got {context.app.skill_names}"
)
@then("resolve_tools should be called once per unique skill")
def step_resolve_called_once(context: Context) -> None:
context.mock_service.resolve_tools.assert_called_once_with("local/web-tools")
# ---------------------------------------------------------------------------
# Scenario: Reactive app raises error for empty string skill name (m3)
# ---------------------------------------------------------------------------
@given("a reactive app configured with an empty string skill name")
def step_app_with_empty_string_skill(context: Context) -> None:
# No mock service needed — the empty string is rejected by
# _sanitize_skill_name before _resolve_skills is even called.
pass
@when("I attempt to create the app with the empty skill")
def step_attempt_create_app_with_empty_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
context.raised_exception = None
try:
ReactiveCleverAgentsApp(skill_names=[""])
except CleverAgentsException as exc:
context.raised_exception = exc
@then("a CleverAgentsException should be raised for resolution failure")
def step_exception_raised_for_resolution_failure(context: Context) -> None:
assert context.raised_exception is not None, "Expected CleverAgentsException"
msg = str(context.raised_exception)
assert "Invalid skill name" in msg, f"Expected 'Invalid skill name' in: {msg}"
# ---------------------------------------------------------------------------
# Scenario: Reactive app skill injection skips LLM agents (C1)
# ---------------------------------------------------------------------------
@given("a reactive app with skill tools and a config with an LLM agent")
def step_app_with_skill_tools_and_llm_agent(context: Context) -> None:
from cleveragents.domain.models.core.skill import ResolvedToolEntry
mock_service = MagicMock()
mock_skill = MagicMock()
mock_entries = [
ResolvedToolEntry(
name="skill-tool-x",
source_skill="local/test-skill",
is_inline=False,
),
]
mock_service.resolve_tools.return_value = (mock_skill, mock_entries)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/test-skill"],
)
@when("agents are registered from config with skill tools present")
def step_register_agents_with_skill_tools_and_llm(context: Context) -> None:
llm_agent_cfg = AgentConfig(
name="llm_actor",
type="llm",
config={"system_prompt": "You are helpful.", "model": "test"},
)
tool_agent_cfg = AgentConfig(
name="tool_actor",
type="custom",
config={"tools": [{"operation": "uppercase"}]},
)
context.app.config = ReactiveConfig(
agents={"llm_actor": llm_agent_cfg, "tool_actor": tool_agent_cfg},
routes={},
)
context.app._register_agents_from_config() # pylint: disable=protected-access
@then("the LLM agent should remain a SimpleLLMAgent not a SimpleToolAgent")
def step_llm_agent_not_converted(context: Context) -> None:
from cleveragents.reactive.stream_router import SimpleLLMAgent, SimpleToolAgent
llm_agent = context.app.stream_router.agents.get("llm_actor")
assert llm_agent is not None, "Expected llm_actor to be registered"
assert isinstance(llm_agent, SimpleLLMAgent), (
f"Expected SimpleLLMAgent but got {type(llm_agent).__name__}"
)
tool_agent = context.app.stream_router.agents.get("tool_actor")
assert tool_agent is not None, "Expected tool_actor to be registered"
assert isinstance(tool_agent, SimpleToolAgent), (
f"Expected SimpleToolAgent but got {type(tool_agent).__name__}"
)
# Tool agent should have original tool + skill tool
assert len(tool_agent.tools) == 2, (
f"Expected 2 tools but got {len(tool_agent.tools)}"
)
# ---------------------------------------------------------------------------
# Scenario: Reactive app raises error for skill resolution ValueError
# Targets: except ValueError branch in _resolve_skills()
# ---------------------------------------------------------------------------
@given("a reactive app configured with a skill that triggers a ValueError")
def step_app_with_valueerror_skill(context: Context) -> None:
mock_service = MagicMock()
mock_service.resolve_tools.side_effect = ValueError(
"Cycle detected in skill includes"
)
context.mock_service = mock_service
@when("I attempt to create the app with the ValueError skill")
def step_attempt_create_app_with_valueerror_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
mock_container = MagicMock()
mock_container.skill_service.return_value = context.mock_service
context.raised_exception = None
try:
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
ReactiveCleverAgentsApp(skill_names=["local/cycle-skill"])
except CleverAgentsException as exc:
context.raised_exception = exc
@then("a CleverAgentsException should be raised with resolution failed message")
def step_exception_raised_with_resolution_failed(context: Context) -> None:
assert context.raised_exception is not None, "Expected CleverAgentsException"
msg = str(context.raised_exception)
assert "resolution failed" in msg, f"Expected 'resolution failed' in: {msg}"
assert "local/cycle-skill" in msg, f"Expected 'local/cycle-skill' in: {msg}"
# ---------------------------------------------------------------------------
# Scenario: Reactive app handles skill that resolves to zero tools
# Targets: zero-tool warning path (lines 116-124)
# ---------------------------------------------------------------------------
@given("a reactive app with a skill that resolves to zero tools")
def step_app_with_zero_tool_skill(context: Context) -> None:
mock_service = MagicMock()
mock_skill = MagicMock()
mock_service.resolve_tools.return_value = (mock_skill, [])
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
context.app = ReactiveCleverAgentsApp(
skill_names=["local/empty-skill"],
)
@when("I check the resolved skill tools after zero tool resolution")
def step_check_resolved_skill_tools_zero(context: Context) -> None:
context.resolved_tools_result = context.app.resolved_skill_tools
context.skill_names_result = context.app.skill_names
@then("the skill names should contain the zero-tool skill")
def step_skill_names_contain_zero_tool(context: Context) -> None:
assert context.skill_names_result == ["local/empty-skill"], (
f"Expected ['local/empty-skill'] but got {context.skill_names_result}"
)
# ---------------------------------------------------------------------------
# Scenario: Reactive app rejects skill name exceeding max length
# Targets: _sanitize_skill_name edge case — name > 127+1+127 = 255 chars
# ---------------------------------------------------------------------------
@given("a reactive app configured with an overly long skill name")
def step_app_with_long_skill_name(context: Context) -> None:
# 128 chars in namespace segment exceeds 127 limit
context.long_name = "a" * 128 + "/" + "b" * 10
@when("I attempt to create the app with the long skill name")
def step_attempt_create_app_with_long_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
context.raised_exception = None
try:
ReactiveCleverAgentsApp(skill_names=[context.long_name])
except CleverAgentsException as exc:
context.raised_exception = exc
@then("a CleverAgentsException should be raised for invalid name format")
def step_exception_raised_for_invalid_name(context: Context) -> None:
assert context.raised_exception is not None, "Expected CleverAgentsException"
msg = str(context.raised_exception)
assert "Invalid skill name" in msg, f"Expected 'Invalid skill name' in: {msg}"
# ---------------------------------------------------------------------------
# Scenario: Reactive app strips ANSI escape codes from skill names
# ---------------------------------------------------------------------------
@given("a reactive app configured with a skill name containing ANSI codes")
def step_app_with_ansi_skill_name(context: Context) -> None:
# ANSI escape code \x1b[31m is a control character sequence
context.ansi_name = "\x1b[31mlocal/bad\x1b[0m"
@when("I attempt to create the app with the ANSI skill name")
def step_attempt_create_app_with_ansi_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
context.raised_exception = None
try:
ReactiveCleverAgentsApp(skill_names=[context.ansi_name])
except CleverAgentsException as exc:
context.raised_exception = exc
# ---------------------------------------------------------------------------
# Scenario: Reactive app rejects skill name with disallowed characters
# ---------------------------------------------------------------------------
@given("a reactive app configured with a skill name containing special characters")
def step_app_with_special_char_skill_name(context: Context) -> None:
context.special_name = "local/bad@skill!"
@when("I attempt to create the app with the special char skill name")
def step_attempt_create_app_with_special_char_skill(context: Context) -> None:
from cleveragents.core.exceptions import CleverAgentsException
context.raised_exception = None
try:
ReactiveCleverAgentsApp(skill_names=[context.special_name])
except CleverAgentsException as exc:
context.raised_exception = exc
features/steps/skill_cli_coverage_boost_steps.py
+28 -84
View File
@@ -80,76 +80,30 @@ def step_boost_background(context: Context) -> None:
context.boost_guard_aborted = False
# ── Given: _get_skill_service DB path ───────────────────────
# ── Given: _get_skill_service delegates to container ────────
@given("boost- the module-level _service is set to None")
def step_boost_set_service_none(context: Context) -> None:
"""Set the module-level _service to None so _get_skill_service re-initialises."""
skill_mod._service = None
"""No-op — _get_skill_service always delegates to the container now."""
@given("boost- the DI container and DB components are mocked successfully")
def step_boost_mock_db_components(context: Context) -> None:
"""Mock container, create_engine, sessionmaker, SkillRepository.
Since the function uses lazy imports (``from sqlalchemy import create_engine``
etc.), we must patch at the source module level (e.g. ``sqlalchemy.create_engine``).
"""
mock_container = MagicMock()
mock_container.database_url.return_value = "sqlite:///test_boost.db"
mock_engine = MagicMock()
mock_session_factory = MagicMock()
@given("boost- the DI container returns a DB-backed SkillService")
def step_boost_mock_container_db_service(context: Context) -> None:
"""Mock get_container().skill_service() to return a SkillService with a repo."""
mock_skill_repo = MagicMock()
mock_service = SkillService(skill_repo=mock_skill_repo)
mock_container = MagicMock()
mock_container.skill_service.return_value = mock_service
# Patch at the source modules since the function uses lazy imports
p1 = patch(
"cleveragents.cli.commands.skill.get_container",
return_value=mock_container,
)
p2 = patch(
"sqlalchemy.create_engine",
return_value=mock_engine,
)
p3 = patch(
"sqlalchemy.orm.sessionmaker",
return_value=mock_session_factory,
)
p4 = patch(
"cleveragents.infrastructure.database.repositories.SkillRepository",
return_value=mock_skill_repo,
)
p1.start()
context.boost_mock_engine_fn = p2.start()
p3.start()
p4.start()
context.boost_mock_db_url = "sqlite:///test_boost.db"
context.boost_patches.extend([p1, p2, p3, p4])
@given("boost- the DI container import will raise an exception")
def step_boost_container_raises(context: Context) -> None:
"""Patch get_container to raise so the except branch is hit."""
p1 = patch(
"cleveragents.cli.commands.skill.get_container",
side_effect=RuntimeError("DB unavailable"),
)
p1.start()
context.boost_patches.append(p1)
@given("boost- create_engine will raise an OperationalError")
def step_boost_create_engine_raises(context: Context) -> None:
"""Patch create_engine to raise so the except branch catches it."""
p1 = patch(
"sqlalchemy.create_engine",
side_effect=Exception("Could not connect to database"),
)
p1.start()
context.boost_patches.append(p1)
context.boost_mock_container = mock_container
# ── Given: tools with agent_skill entries ───────────────────
@@ -195,6 +149,16 @@ def step_boost_call_get_skill_service(context: Context) -> None:
_stop_patches(context)
@when("boost- I call _get_skill_service twice")
def step_boost_call_get_skill_service_twice(context: Context) -> None:
"""Call _get_skill_service twice to test caching."""
try:
context.boost_returned_service = _get_skill_service()
_get_skill_service() # second call should use cached
finally:
_stop_patches(context)
# ── When: tools command ─────────────────────────────────────
@@ -258,36 +222,16 @@ def step_boost_service_has_repo(context: Context) -> None:
)
@then("boost- create_engine should have been called with the mock database URL")
def step_boost_create_engine_called(context: Context) -> None:
"""Assert create_engine was called with the expected DB URL."""
context.boost_mock_engine_fn.assert_called_once()
call_args = context.boost_mock_engine_fn.call_args
assert call_args[0][0] == context.boost_mock_db_url, (
f"Expected create_engine called with '{context.boost_mock_db_url}', "
f"got {call_args}"
)
@then("boost- the container skill_service should be called once")
def step_boost_container_called_once(context: Context) -> None:
"""Assert get_container().skill_service() was called only once (caching)."""
context.boost_mock_container.skill_service.assert_called_once()
@then("boost- the returned service should be an in-memory SkillService")
def step_boost_service_is_inmemory(context: Context) -> None:
"""Assert the returned service is a SkillService instance."""
svc = context.boost_returned_service
assert svc is not None, "No service was returned"
assert isinstance(svc, SkillService), (
f"Expected SkillService, got {type(svc).__name__}"
)
@then("boost- the returned service should have no skill_repo")
def step_boost_service_no_repo(context: Context) -> None:
"""Assert the fallback service has no DB repo."""
svc = context.boost_returned_service
assert svc is not None, "No service was returned"
assert svc._skill_repo is None, (
f"Expected skill_repo to be None (in-memory fallback), "
f"but got {svc._skill_repo}"
)
@then("boost- the container skill_service should be called twice")
def step_boost_container_called_twice(context: Context) -> None:
"""Assert get_container().skill_service() is called on every access (no cache)."""
assert context.boost_mock_container.skill_service.call_count == 2
# ── Then: CLI exit code ─────────────────────────────────────
robot/helper_skill_actor_run.py
+143
View File
@@ -0,0 +1,143 @@
"""Helper script for skill_actor_run.robot smoke tests.
Tests the ``--skill`` flag on the ``actor-run run`` CLI command.
Each subcommand is a self-contained check that prints a sentinel on success.
"""
from __future__ import annotations
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from typer.testing import CliRunner # noqa: E402
from cleveragents.cli.commands.actor_run import app as actor_run_app # noqa: E402
runner = CliRunner()
_SIMPLE_YAML = """\
name: smoke-actor
type: custom
tools:
- operation: identity
"""
def _write_yaml(content: str) -> str:
fd, path = tempfile.mkstemp(suffix=".yaml")
with os.fdopen(fd, "w") as fh:
fh.write(content)
return path
def unknown_skill_exits_with_error() -> None:
"""Verify that ``--skill`` with an unknown skill name exits with code 2."""
path = _write_yaml(_SIMPLE_YAML)
try:
result = runner.invoke(
actor_run_app,
[
"--config",
path,
"--prompt",
"hello",
"--skill",
"local/nonexistent-skill",
],
)
if result.exit_code == 2 and "not found in registry" in result.output:
print("skill-actor-run-unknown-skill-ok")
else:
print(
f"FAIL: expected exit code 2 with 'not found in registry', "
f"got code={result.exit_code}",
file=sys.stderr,
)
print(result.output, file=sys.stderr)
if result.exception:
import traceback
traceback.print_exception(
type(result.exception),
result.exception,
result.exception.__traceback__,
file=sys.stderr,
)
sys.exit(1)
finally:
Path(path).unlink(missing_ok=True)
def skill_flag_accepted() -> None:
"""Verify that ``--skill`` with a valid skill name is accepted by the CLI."""
from cleveragents.application.services.skill_service import SkillService
from cleveragents.domain.models.core.skill import Skill
# Build a real in-memory SkillService with a registered skill
svc = SkillService()
skill = Skill(
name="local/smoke-skill",
description="Smoke test skill",
tool_refs=["builtin/read_file"],
)
svc._skills["local/smoke-skill"] = skill
mock_container = MagicMock()
mock_container.skill_service.return_value = svc
path = _write_yaml(_SIMPLE_YAML)
try:
with patch(
"cleveragents.reactive.application.get_container",
return_value=mock_container,
):
result = runner.invoke(
actor_run_app,
[
"--config",
path,
"--prompt",
"hello",
"--skill",
"local/smoke-skill",
],
)
# The command may fail for other reasons (no real LLM, etc.) but
# the important thing is it did NOT fail with "not found in registry"
# and the skill was accepted.
if "not found in registry" in (result.output or ""):
print("FAIL: skill was rejected as not found", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
print("skill-actor-run-flag-accepted-ok")
finally:
Path(path).unlink(missing_ok=True)
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS = {
"unknown-skill": unknown_skill_exits_with_error,
"skill-flag-accepted": skill_flag_accepted,
}
def main() -> None:
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
sys.exit(2)
_COMMANDS[sys.argv[1]]()
if __name__ == "__main__":
main()
robot/helper_skill_cli.py
+7 -3
View File
@@ -19,8 +19,12 @@ if _SRC not in sys.path:
from typer.testing import CliRunner # noqa: E402
from cleveragents.application.services.skill_service import SkillService # noqa: E402
from cleveragents.cli.commands import skill as skill_mod # noqa: E402
from cleveragents.cli.commands.skill import app as skill_app # noqa: E402
from cleveragents.cli.commands.skill import ( # noqa: E402
_reset_skill_service,
)
from cleveragents.cli.commands.skill import ( # noqa: E402
app as skill_app,
)
from cleveragents.domain.models.core.skill import Skill # noqa: E402
runner = CliRunner()
@@ -53,7 +57,7 @@ def _write_yaml(content: str) -> str:
def _fresh_service() -> SkillService:
"""Create and install a fresh SkillService."""
svc = SkillService()
skill_mod._service = svc
_reset_skill_service(svc)
return svc
robot/skill_actor_run.robot
+27
View File
@@ -0,0 +1,27 @@
*** Settings ***
Documentation Smoke tests for the --skill flag on the actor-run run command
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_skill_actor_run.py
*** Test Cases ***
Actor Run With Unknown Skill Exits With Error
[Documentation] Verify that ``actor-run run --skill local/nonexistent`` exits
... with code 2 and an error message containing "not found in registry".
${result}= Run Process ${PYTHON} ${HELPER} unknown-skill cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-actor-run-unknown-skill-ok
Actor Run Skill Flag Accepted
[Documentation] Verify that ``actor-run run --skill local/smoke-skill`` is
... accepted by the CLI when the skill exists in the registry.
${result}= Run Process ${PYTHON} ${HELPER} skill-flag-accepted cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-actor-run-flag-accepted-ok
src/cleveragents/application/container.py
+59
View File
@@ -66,6 +66,7 @@ from cleveragents.application.services.session_service import (
from cleveragents.application.services.skeleton_compressor import (
SkeletonCompressorService,
)
from cleveragents.application.services.skill_service import SkillService
from cleveragents.application.services.subplan_service import SubplanService
from cleveragents.application.services.trace_service import TraceService
from cleveragents.application.services.uko_indexer import UKOIndexer
@@ -332,6 +333,56 @@ def _build_trace_service(
return TraceService(settings=resolved_settings, repository=repository)
def _build_skill_service(
database_url: str,
) -> SkillService:
"""Build a SkillService with DB persistence.
Follows the same ``_build_*`` pattern used by ``_build_session_service``
and other DB-backed services: create engine, session factory, repository,
then inject into the service constructor.
Falls back to an in-memory ``SkillService()`` when the database is
unavailable (e.g. before ``agents init``). Only expected operational
exceptions (``ImportError``, ``OperationalError``, ``DatabaseError``,
``OSError``) are caught — unexpected errors propagate to the caller.
"""
try:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
except ImportError as exc:
_logger.warning(
"skill_service_db_fallback",
reason="SQLAlchemy not available, falling back to in-memory SkillService",
exc_info=True,
error_type=type(exc).__name__,
)
return SkillService()
from sqlalchemy.exc import DatabaseError, OperationalError
try:
from cleveragents.infrastructure.database.repositories import (
SkillRepository,
)
engine = create_engine(database_url, echo=False)
factory = sessionmaker(bind=engine, expire_on_commit=False)
skill_repo = SkillRepository(session_factory=factory)
return SkillService(
skill_repo=skill_repo,
session_factory=factory,
)
except (OperationalError, DatabaseError, OSError) as exc:
_logger.warning(
"skill_service_db_fallback",
reason="Failed to create DB-backed SkillService, falling back to in-memory",
exc_info=True,
error_type=type(exc).__name__,
)
return SkillService()
def _build_session_service(
database_url: str,
event_bus: ReactiveEventBus | None = None,
@@ -554,6 +605,14 @@ class Container(containers.DeclarativeContainer):
settings=settings,
)
# Skill Service — Singleton so skill state is consistent within the
# same process (Forgejo #887). The CLI ``_get_skill_service()``
# delegates to this provider to avoid divergent instances.
skill_service = providers.Singleton(
_build_skill_service,
database_url=database_url,
)
# Session Service - database-backed session management (Forgejo #554, #570, #680)
session_service = providers.Factory(
_build_session_service,
src/cleveragents/cli/commands/actor.py
+57 -48
View File
@@ -101,64 +101,73 @@ def run(
help="Allow RxPy stream routes in run mode (bypass validation)",
),
] = False,
skill: Annotated[
list[str] | None,
typer.Option(
"--skill",
metavar="NAME",
help="Skill to attach; only augments tool-bearing agents (repeatable)",
),
] = None,
) -> None:
"""Run the reactive network once with actor-first configs."""
app_exec = ReactiveCleverAgentsApp(
config_files=config,
verbose=verbose,
unsafe=unsafe,
temperature_override=temperature,
)
try:
app_exec = ReactiveCleverAgentsApp(
config_files=config,
verbose=verbose,
unsafe=unsafe,
temperature_override=temperature,
skill_names=skill,
)
async def _execute() -> str:
if load_context and context:
ctx_mgr = ContextManager(context, context_dir)
ctx_mgr.import_context(load_context)
if ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
if load_context:
import json as _json
async def _execute() -> str:
if load_context and context:
ctx_mgr = ContextManager(context, context_dir)
ctx_mgr.import_context(load_context)
if ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
if load_context:
import json as _json
with open(load_context, encoding="utf-8") as f:
data = _json.load(f)
if data.get("global_context") and app_exec.config:
app_exec.config.global_context.update(data["global_context"])
with open(load_context, encoding="utf-8") as f:
data = _json.load(f)
if data.get("global_context") and app_exec.config:
app_exec.config.global_context.update(data["global_context"])
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
if context:
ctx_mgr = ContextManager(context, context_dir)
if ctx_mgr.exists() and ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
if context:
ctx_mgr = ContextManager(context, context_dir)
if ctx_mgr.exists() and ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
try:
result = asyncio.run(_execute())
except UnsafeConfigurationError as exc:
typer.echo(f"Error: {exc}", err=True)
src/cleveragents/cli/commands/actor_run.py
+57 -48
View File
@@ -74,64 +74,73 @@ def run(
help="Allow RxPy stream routes in run mode (bypass validation)",
),
] = False,
skill: Annotated[
list[str] | None,
typer.Option(
"--skill",
metavar="NAME",
help="Skill to attach; only augments tool-bearing agents (repeatable)",
),
] = None,
) -> None:
"""Run the reactive network once with actor-first configs."""
app_exec = ReactiveCleverAgentsApp(
config_files=config,
verbose=verbose,
unsafe=unsafe,
temperature_override=temperature,
)
try:
app_exec = ReactiveCleverAgentsApp(
config_files=config,
verbose=verbose,
unsafe=unsafe,
temperature_override=temperature,
skill_names=skill,
)
async def _execute() -> str:
if load_context and context:
ctx_mgr = ContextManager(context, context_dir)
ctx_mgr.import_context(load_context)
if ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
if load_context:
import json as _json
async def _execute() -> str:
if load_context and context:
ctx_mgr = ContextManager(context, context_dir)
ctx_mgr.import_context(load_context)
if ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
if load_context:
import json as _json
with open(load_context, encoding="utf-8") as f:
data = _json.load(f)
if data.get("global_context") and app_exec.config:
app_exec.config.global_context.update(data["global_context"])
with open(load_context, encoding="utf-8") as f:
data = _json.load(f)
if data.get("global_context") and app_exec.config:
app_exec.config.global_context.update(data["global_context"])
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
if context:
ctx_mgr = ContextManager(context, context_dir)
if ctx_mgr.exists() and ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
if context:
ctx_mgr = ContextManager(context, context_dir)
if ctx_mgr.exists() and ctx_mgr.global_context and app_exec.config:
app_exec.config.global_context.update(ctx_mgr.global_context)
ctx_mgr.add_message("user", prompt)
result = await app_exec.run_single_shot(
prompt,
context_manager=ctx_mgr,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
ctx_mgr.add_message("assistant", result)
if app_exec.config:
ctx_mgr.save_global_context(app_exec.config.global_context)
return result
return await app_exec.run_single_shot(
prompt,
context_manager=None,
allow_rxpy_in_run_mode=allow_rxpy_in_run_mode,
)
try:
result = asyncio.run(_execute())
except UnsafeConfigurationError as exc:
typer.echo(f"Error: {exc}", err=True)
src/cleveragents/cli/commands/skill.py
+23 -35
View File
@@ -37,6 +37,7 @@ task C0.skill.cli.
from __future__ import annotations
import logging
from pathlib import Path
from typing import Annotated, Any
@@ -55,6 +56,8 @@ from cleveragents.domain.models.core.skill import (
)
from cleveragents.skills.schema import SkillConfigSchema
logger = logging.getLogger(__name__)
# Create sub-app for skill commands
app = typer.Typer(help="Manage skills (reusable, namespaced tool collections).")
console = Console()
@@ -63,45 +66,21 @@ console = Console()
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# ---------------------------------------------------------------------------
# Module-level service. A module-level singleton keeps state across CLI
# invocations in the same process (e.g. tests). Production usage will use
# the DI container.
# Service accessor. Always delegates to the DI container singleton so
# that the CLI and reactive layers share the same ``SkillService`` instance.
# No module-level cache — the Container's own Singleton provider handles
# caching, and ``reset_container()`` correctly invalidates it.
# ---------------------------------------------------------------------------
_service: SkillService | None = None
def _get_skill_service() -> SkillService:
"""Get or create the module-level SkillService with DB persistence.
"""Return the shared SkillService from the DI container.
Follows the same pattern as ``_get_tool_registry_service`` in
``tool.py``: obtains the database URL from the DI container,
creates an engine/session factory/repository, and injects them
into the ``SkillService`` so that every mutation is persisted.
Delegates to ``get_container().skill_service()`` on every call so
that ``reset_container()`` correctly invalidates the cached instance.
The container provider handles DB-fallback logic.
"""
global _service
if _service is None:
try:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.repositories import (
SkillRepository,
)
container = get_container()
database_url: str = container.database_url()
engine = create_engine(database_url, echo=False)
factory = sessionmaker(bind=engine, expire_on_commit=False)
skill_repo = SkillRepository(session_factory=factory)
_service = SkillService(
skill_repo=skill_repo,
session_factory=factory,
)
except Exception:
# Fallback to pure in-memory if DB is unavailable
_service = SkillService()
return _service
return get_container().skill_service()
def _reset_skill_service(
@@ -114,9 +93,18 @@ def _reset_skill_service(
When ``None``, the next ``_get_skill_service`` call will
create a fresh in-memory service (no DB persistence) to
avoid side-effects during unit testing.
Note:
This patches the container's ``skill_service`` provider so
that ``_get_skill_service()`` (which always delegates to the
container) picks up the replacement. When *service* is
``None`` a plain in-memory ``SkillService()`` is installed.
"""
global _service
_service = service if service is not None else SkillService()
from dependency_injector import providers
svc = service if service is not None else SkillService()
container = get_container()
container.skill_service.override(providers.Object(svc))
# ---------------------------------------------------------------------------
src/cleveragents/reactive/application.py
+125 -250
View File
@@ -5,15 +5,18 @@ from __future__ import annotations
import asyncio
import logging
import re
from collections.abc import Callable
from pathlib import Path
from typing import Any
from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined]
from cleveragents.application.container import get_container
from cleveragents.core.exceptions import CleverAgentsException, UnsafeConfigurationError
from cleveragents.reactive.config_parser import ReactiveConfig, ReactiveConfigParser
from cleveragents.reactive.context_manager import ContextManager
from cleveragents.reactive.route import RouteType
from cleveragents.reactive.graph_executor import GraphExecutor
from cleveragents.reactive.route import RouteConfig, RouteType
from cleveragents.reactive.route_bridge import RouteBridge
from cleveragents.reactive.stream_router import (
ReactiveStreamRouter,
@@ -21,6 +24,10 @@ from cleveragents.reactive.stream_router import (
SimpleToolAgent,
)
# Skill name format: namespace/short-name, ASCII alphanumeric + ._- only.
_SKILL_NAME_RE = re.compile(r"^[\w.-]{1,127}/[\w.-]{1,127}$", re.ASCII)
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]")
logger = logging.getLogger(__name__)
@@ -33,10 +40,15 @@ class ReactiveCleverAgentsApp:
verbose: int = 0,
unsafe: bool = False,
temperature_override: float | None = None,
skill_names: list[str] | None = None,
):
self.verbose = verbose
self.unsafe = unsafe
self.temperature_override = temperature_override
self._skill_names: list[str] = list(
dict.fromkeys(self._sanitize_skill_name(n) for n in (skill_names or []))
)
self._resolved_skill_tools: list[dict[str, Any]] = []
self._configure_logging(verbose)
self.stream_router = ReactiveStreamRouter()
@@ -45,6 +57,9 @@ class ReactiveCleverAgentsApp:
self.route_bridge = None
self.config: ReactiveConfig | None = None
if self._skill_names:
self._resolve_skills()
if config_files:
self.load_configuration(config_files)
if self.temperature_override is not None and self.config:
@@ -56,6 +71,76 @@ class ReactiveCleverAgentsApp:
)
self._enforce_unsafe_flag()
@property
def skill_names(self) -> list[str]:
"""Return the list of skill names attached to this run."""
return list(self._skill_names)
@property
def resolved_skill_tools(self) -> list[dict[str, Any]]:
"""Return the resolved skill tool configurations."""
return list(self._resolved_skill_tools)
@staticmethod
def _sanitize_skill_name(name: str) -> str:
"""Validate format and strip control characters from a skill name.
Raises:
CleverAgentsException: If the name does not match the expected
``namespace/short-name`` pattern or exceeds 256 characters.
"""
sanitized = _CONTROL_CHAR_RE.sub("", name)
if not _SKILL_NAME_RE.match(sanitized):
raise CleverAgentsException(
f"Invalid skill name format: '{sanitized}'. "
"Expected namespace/name (ASCII alphanumeric, ._- only, "
"each segment max 127 chars)."
)
return sanitized
def _resolve_skills(self) -> None:
"""Resolve skill names to tool entries via the DI-provided SkillService.
Obtains the ``SkillService`` singleton from the application
container, eliminating the previous cross-layer import from the
CLI package and the TOCTOU between CLI validation and runtime
resolution.
Raises:
CleverAgentsException: If any skill is not found or invalid.
"""
service = get_container().skill_service()
for name in self._skill_names:
try:
_skill, entries = service.resolve_tools(name)
if not entries:
logger.warning("Skill '%s' resolved to zero tools", name)
for entry in entries:
tool_dict: dict[str, Any] = {
"name": entry.name,
"source_skill": entry.source_skill,
"is_inline": entry.is_inline,
"operation": "identity",
"skill_source": name,
}
if entry.overrides:
tool_dict["overrides"] = dict(entry.overrides)
self._resolved_skill_tools.append(tool_dict)
except KeyError as exc:
raise CleverAgentsException(
f"Skill '{name}' not found in registry"
) from exc
except ValueError as exc:
raise CleverAgentsException(
f"Skill '{name}' resolution failed: {exc}"
) from exc
if self._resolved_skill_tools:
logger.info(
"Resolved %d tools from %d skill(s)",
len(self._resolved_skill_tools),
len(self._skill_names),
)
def _configure_logging(self, verbose: int) -> None:
if verbose == 0:
level = logging.CRITICAL
@@ -93,6 +178,14 @@ class ReactiveCleverAgentsApp:
def _make_agent_instance(name: str, agent_cfg: Any) -> Any:
tools = agent_cfg.config.get("tools", []) if agent_cfg.config else []
if self._resolved_skill_tools and tools:
tools = list(tools) + self._resolved_skill_tools
elif self._resolved_skill_tools and not tools:
logger.debug(
"Skipping skill tool injection for agent '%s' "
"(agent has no base tools)",
name,
)
if tools:
return SimpleToolAgent(tools, unsafe=self.unsafe)
if agent_cfg.type == "llm":
@@ -207,7 +300,7 @@ class ReactiveCleverAgentsApp:
return False
return any(r.type == RouteType.STREAM for r in self.config.routes.values())
def _get_graph_route(self) -> Any:
def _get_graph_route(self) -> RouteConfig | None:
"""Return the first graph route config, or None."""
if not self.config:
return None
@@ -216,20 +309,17 @@ class ReactiveCleverAgentsApp:
return route
return None
# ------------------------------------------------------------------
# Backward-compatible proxies into GraphExecutor.
# Existing tests access these via ``app._xxx()``.
# ------------------------------------------------------------------
def _invoke_agent(self, agent: Any, message: str, context: dict[str, Any]) -> str:
"""Execute an agent and return the string result."""
if isinstance(agent, (SimpleToolAgent | SimpleLLMAgent)):
result = agent.process(message, context=context)
elif hasattr(agent, "process"):
result = agent.process(message)
elif callable(agent):
result = agent(message)
else:
result = message
return str(result) if result else ""
return GraphExecutor._invoke_agent(agent, message, context)
def _parse_graph_topology(
self, route: Any
self, route: RouteConfig
) -> tuple[
str | None,
list[dict[str, Any]],
@@ -237,104 +327,18 @@ class ReactiveCleverAgentsApp:
dict[str, list[dict[str, Any]]],
]:
"""Parse route nodes/edges into router node, rules, actor map, and edge map."""
router_node: str | None = None
router_rules: list[dict[str, Any]] = []
node_actor_map: dict[str, str] = {}
node_edges: dict[str, list[dict[str, Any]]] = {}
for node_name, node_data in route.nodes.items():
node_type = node_data.get("type", "")
if node_type == "message_router":
router_node = node_name
router_rules = node_data.get("rules", [])
elif node_type == "actor":
actor_name = node_data.get("actor", node_name)
node_actor_map[node_name] = actor_name
for edge_data in route.edges:
if isinstance(edge_data, dict):
source = edge_data.get("source", "")
target = edge_data.get("target", "")
condition = edge_data.get("condition")
else:
source = getattr(edge_data, "source", "")
target = getattr(edge_data, "target", "")
condition = getattr(edge_data, "condition", None)
if source and target:
node_edges.setdefault(source, []).append(
{"target": target, "condition": condition}
)
return router_node, router_rules, node_actor_map, node_edges
return GraphExecutor._parse_topology(route)
def _initialize_graph_context(self) -> dict[str, Any]:
"""Ensure global context has required default keys for graph execution."""
context = self.config.global_context if self.config else {}
default_stage_order = [
"intro",
"discovery",
"brainstorming",
"vetting",
"structure",
"section_writing",
"paper_review",
"latex_generation",
]
stage_order = context.get("stage_order")
requires_refresh = not isinstance(stage_order, list) or any(
stage not in (stage_order or [])
for stage in ("section_writing", "paper_review", "latex_generation")
)
if requires_refresh:
context["stage_order"] = list(default_stage_order)
if "writing_stage" not in context or context.get("writing_stage") not in (
context.get("stage_order") or []
):
context["writing_stage"] = "intro"
if "paper_details" not in context:
context["paper_details"] = {
"topic": None,
"length": None,
"audience": None,
"publication": None,
"format": None,
"other": None,
}
return context
executor = GraphExecutor(self.stream_router.agents, self.config)
return executor._initialize_context()
def _match_router_rule(
self, message: str, router_rules: list[dict[str, Any]]
) -> tuple[str | None, str, dict[str, Any] | None]:
"""Match message against router rules.
Returns (target_node, extracted_message, matched_rule).
"""
for rule in router_rules:
match_type = rule.get("match_type", "prefix")
pattern = rule.get("pattern", "")
rule_target = rule.get("target")
extract = rule.get("extract_message", False)
separator = rule.get("separator", ":")
if not rule_target:
continue
matched = False
if match_type == "prefix":
matched = pattern == "" or message.startswith(pattern)
elif match_type == "contains":
matched = bool(pattern and pattern in message)
elif match_type == "suffix":
matched = pattern == "" or message.endswith(pattern)
if matched:
extracted = message
if extract and separator and separator in message:
idx = message.index(separator)
extracted = message[idx + len(separator) :]
return rule_target, extracted, rule
return None, message, None
"""Match message against router rules."""
return GraphExecutor._match_router_rule(message, router_rules)
def _follow_chained_edges(
self,
@@ -344,121 +348,23 @@ class ReactiveCleverAgentsApp:
node_actor_map: dict[str, str],
agents: dict[str, Any],
router_node: str | None,
select_targets_fn: Any,
select_targets_fn: Callable[[str], list[str]],
) -> tuple[str, bool]:
"""Follow chained edges from a node.
Returns (message, should_return_to_caller).
"""
next_node = next_targets[0]
while next_node:
if next_node == "end":
return current_message, True
next_actor = node_actor_map.get(next_node, next_node)
next_agent = agents.get(next_actor)
if next_agent is not None:
result = self._invoke_agent(next_agent, current_message, context)
current_message = result if result else current_message
chained_targets = select_targets_fn(next_node)
if router_node in chained_targets:
return current_message, False
if "end" in chained_targets:
return current_message, True
if not chained_targets:
return current_message, False
next_node = chained_targets[0]
return current_message, False
def _execute_graph_route(self, prompt: str, route: Any) -> str:
"""Execute a graph route by following message_router rules iteratively."""
router_node, router_rules, node_actor_map, node_edges = (
self._parse_graph_topology(route)
"""Follow chained edges from a node."""
return GraphExecutor._follow_chained_edges(
next_targets,
current_message,
context,
node_actor_map,
agents,
router_node,
select_targets_fn,
)
if not router_rules:
return ""
context = self._initialize_graph_context()
agents = self.stream_router.agents
max_iterations = 50
current_message = prompt
def _condition_matches(condition: Any, ctx: dict[str, Any]) -> bool:
if not condition:
return True
if isinstance(condition, dict) and condition.get("type") == "context_value":
key = condition.get("key")
if not key:
return False
return ctx.get(key) == condition.get("value")
return True
def _select_targets(source_node: str) -> list[str]:
edges = node_edges.get(source_node, [])
return [
e["target"]
for e in edges
if _condition_matches(e.get("condition"), context)
]
workflow_hops = 0
for _ in range(max_iterations):
target_node, extracted_message, matched_rule = self._match_router_rule(
current_message, router_rules
)
if target_node == "workflow_controller":
is_default_rule = (
matched_rule
and matched_rule.get("match_type") in {"prefix", "suffix"}
and matched_rule.get("pattern", "") == ""
)
if workflow_hops > 0 and is_default_rule:
return extracted_message
if target_node is None or target_node == "end":
return extracted_message
actor_name = node_actor_map.get(target_node, target_node)
agent = agents.get(actor_name)
if agent is None:
logger.warning(
"Agent '%s' not found for node '%s'", actor_name, target_node
)
return extracted_message
if target_node == "workflow_controller":
workflow_hops += 1
current_message = self._invoke_agent(agent, extracted_message, context)
if not current_message:
return ""
next_targets = _select_targets(target_node)
if router_node in next_targets:
continue
elif "end" in next_targets:
return current_message
elif next_targets:
current_message, should_return = self._follow_chained_edges(
next_targets,
current_message,
context,
node_actor_map,
agents,
router_node,
_select_targets,
)
if should_return:
return current_message
continue
else:
continue
logger.warning("Graph route execution hit max iterations (%d)", max_iterations)
return current_message
def _execute_graph_route(self, prompt: str, route: RouteConfig) -> str:
"""Execute a graph route by following message_router rules iteratively."""
executor = GraphExecutor(self.stream_router.agents, self.config)
return executor.execute(prompt, route)
async def run_single_shot(
self,
@@ -481,21 +387,9 @@ class ReactiveCleverAgentsApp:
# Try graph route execution first
graph_route = self._get_graph_route()
if graph_route is not None:
output = self._execute_graph_route(prompt, graph_route)
for prefix in [
"DISCOVERY_RESPONSE:",
"GOTO_BRAINSTORMING:",
"SET_TOPIC:",
"COMMAND_OUTPUT:",
]:
if output.startswith(prefix):
output = output[len(prefix) :]
# Strip any remaining ROUTE_* prefixes
# (e.g. ROUTE_ASK_TOPIC:, ROUTE_ASK_LENGTH:, etc.)
route_match = re.match(r"^ROUTE_[A-Z_]+:", output)
if route_match:
output = output[route_match.end() :]
return output
executor = GraphExecutor(self.stream_router.agents, self.config)
output = executor.execute(prompt, graph_route)
return GraphExecutor.strip_routing_prefixes(output)
loop = asyncio.get_running_loop()
scheduler = AsyncIOScheduler(loop=loop)
@@ -523,26 +417,7 @@ class ReactiveCleverAgentsApp:
return ""
output = "\n".join(result_container)
routing_prefixes = [
"DISCOVERY_RESPONSE:",
"GOTO_BRAINSTORMING:",
"SET_TOPIC:",
"COMMAND_OUTPUT:",
]
cleaned_lines = []
for line in output.split("\n"):
for prefix in routing_prefixes:
if line.startswith(prefix):
line = line[len(prefix) :]
break
else:
# Strip any remaining ROUTE_* prefixes
route_match = re.match(r"^ROUTE_[A-Z_]+:", line)
if route_match:
line = line[route_match.end() :]
cleaned_lines.append(line)
output = "\n".join(cleaned_lines)
return output
return GraphExecutor.strip_routing_prefixes_multiline(output)
async def run_with_context(
self,
src/cleveragents/reactive/graph_executor.py
+337
View File
@@ -0,0 +1,337 @@
"""Graph route execution engine for reactive configurations.
Extracted from ``ReactiveCleverAgentsApp`` to keep ``application.py``
under the 500-line guideline (CONTRIBUTING.md). The ``GraphExecutor``
class encapsulates iterative message-router graph traversal: topology
parsing, context initialisation, rule matching, and chained-edge
following.
"""
from __future__ import annotations
import logging
import re
from collections.abc import Callable
from typing import Any
from cleveragents.reactive.config_parser import ReactiveConfig
from cleveragents.reactive.route import RouteConfig
from cleveragents.reactive.stream_router import SimpleLLMAgent, SimpleToolAgent
logger = logging.getLogger(__name__)
class GraphExecutor:
"""Execute graph routes by following message_router rules iteratively.
Args:
agents: Registered agent instances (keyed by name).
config: The parsed ``ReactiveConfig``, or ``None``.
"""
def __init__(
self,
agents: dict[str, Any],
config: ReactiveConfig | None = None,
) -> None:
self._agents = agents
self._config = config
# ------------------------------------------------------------------
# Public entry point
# ------------------------------------------------------------------
def execute(self, prompt: str, route: RouteConfig) -> str:
"""Execute a graph route and return the final output string."""
router_node, router_rules, node_actor_map, node_edges = self._parse_topology(
route
)
if not router_rules:
return ""
context = self._initialize_context()
max_iterations = 50
current_message = prompt
def _condition_matches(condition: Any, ctx: dict[str, Any]) -> bool:
if not condition:
return True
if isinstance(condition, dict) and condition.get("type") == "context_value":
key = condition.get("key")
if not key:
return False
return ctx.get(key) == condition.get("value")
return True
def _select_targets(source_node: str) -> list[str]:
edges = node_edges.get(source_node, [])
return [
e["target"]
for e in edges
if _condition_matches(e.get("condition"), context)
]
workflow_hops = 0
for _ in range(max_iterations):
target_node, extracted_message, matched_rule = self._match_router_rule(
current_message, router_rules
)
if target_node == "workflow_controller":
is_default_rule = (
matched_rule
and matched_rule.get("match_type") in {"prefix", "suffix"}
and matched_rule.get("pattern", "") == ""
)
if workflow_hops > 0 and is_default_rule:
return extracted_message
if target_node is None or target_node == "end":
return extracted_message
actor_name = node_actor_map.get(target_node, target_node)
agent = self._agents.get(actor_name)
if agent is None:
logger.warning(
"Agent '%s' not found for node '%s'", actor_name, target_node
)
return extracted_message
if target_node == "workflow_controller":
workflow_hops += 1
current_message = self._invoke_agent(agent, extracted_message, context)
if not current_message:
return ""
next_targets = _select_targets(target_node)
if router_node in next_targets:
continue
elif "end" in next_targets:
return current_message
elif next_targets:
current_message, should_return = self._follow_chained_edges(
next_targets,
current_message,
context,
node_actor_map,
self._agents,
router_node,
_select_targets,
)
if should_return:
return current_message
continue
else:
continue
logger.warning("Graph route execution hit max iterations (%d)", max_iterations)
return current_message
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
@staticmethod
def _invoke_agent(agent: Any, message: str, context: dict[str, Any]) -> str:
"""Execute an agent and return the string result."""
if isinstance(agent, (SimpleToolAgent | SimpleLLMAgent)):
result = agent.process(message, context=context)
elif hasattr(agent, "process"):
result = agent.process(message)
elif callable(agent):
result = agent(message)
else:
result = message
return str(result) if result else ""
@staticmethod
def _parse_topology(
route: RouteConfig,
) -> tuple[
str | None,
list[dict[str, Any]],
dict[str, str],
dict[str, list[dict[str, Any]]],
]:
"""Parse route nodes/edges into router node, rules, actor map, and edge map."""
router_node: str | None = None
router_rules: list[dict[str, Any]] = []
node_actor_map: dict[str, str] = {}
node_edges: dict[str, list[dict[str, Any]]] = {}
for node_name, node_data in route.nodes.items():
node_type = node_data.get("type", "")
if node_type == "message_router":
router_node = node_name
router_rules = node_data.get("rules", [])
elif node_type == "actor":
actor_name = node_data.get("actor", node_name)
node_actor_map[node_name] = actor_name
for edge_data in route.edges:
if isinstance(edge_data, dict):
source = edge_data.get("source", "")
target = edge_data.get("target", "")
condition = edge_data.get("condition")
else:
source = getattr(edge_data, "source", "")
target = getattr(edge_data, "target", "")
condition = getattr(edge_data, "condition", None)
if source and target:
node_edges.setdefault(source, []).append(
{"target": target, "condition": condition}
)
return router_node, router_rules, node_actor_map, node_edges
def _initialize_context(self) -> dict[str, Any]:
"""Ensure global context has required default keys for graph execution."""
context = self._config.global_context if self._config else {}
default_stage_order = [
"intro",
"discovery",
"brainstorming",
"vetting",
"structure",
"section_writing",
"paper_review",
"latex_generation",
]
stage_order = context.get("stage_order")
requires_refresh = not isinstance(stage_order, list) or any(
stage not in (stage_order or [])
for stage in ("section_writing", "paper_review", "latex_generation")
)
if requires_refresh:
context["stage_order"] = list(default_stage_order)
if "writing_stage" not in context or context.get("writing_stage") not in (
context.get("stage_order") or []
):
context["writing_stage"] = "intro"
if "paper_details" not in context:
context["paper_details"] = {
"topic": None,
"length": None,
"audience": None,
"publication": None,
"format": None,
"other": None,
}
return context
@staticmethod
def _match_router_rule(
message: str, router_rules: list[dict[str, Any]]
) -> tuple[str | None, str, dict[str, Any] | None]:
"""Match message against router rules.
Returns (target_node, extracted_message, matched_rule).
"""
for rule in router_rules:
match_type = rule.get("match_type", "prefix")
pattern = rule.get("pattern", "")
rule_target = rule.get("target")
extract = rule.get("extract_message", False)
separator = rule.get("separator", ":")
if not rule_target:
continue
matched = False
if match_type == "prefix":
matched = pattern == "" or message.startswith(pattern)
elif match_type == "contains":
matched = bool(pattern and pattern in message)
elif match_type == "suffix":
matched = pattern == "" or message.endswith(pattern)
if matched:
extracted = message
if extract and separator and separator in message:
idx = message.index(separator)
extracted = message[idx + len(separator) :]
return rule_target, extracted, rule
return None, message, None
@staticmethod
def _follow_chained_edges(
next_targets: list[str],
current_message: str,
context: dict[str, Any],
node_actor_map: dict[str, str],
agents: dict[str, Any],
router_node: str | None,
select_targets_fn: Callable[[str], list[str]],
) -> tuple[str, bool]:
"""Follow chained edges from a node.
Returns (message, should_return_to_caller).
"""
next_node = next_targets[0]
while next_node:
if next_node == "end":
return current_message, True
next_actor = node_actor_map.get(next_node, next_node)
next_agent = agents.get(next_actor)
if next_agent is not None:
result = GraphExecutor._invoke_agent(
next_agent, current_message, context
)
current_message = result if result else current_message
chained_targets = select_targets_fn(next_node)
if router_node in chained_targets:
return current_message, False
if "end" in chained_targets:
return current_message, True
if not chained_targets:
return current_message, False
next_node = chained_targets[0]
return current_message, False
@staticmethod
def strip_routing_prefixes(output: str) -> str:
"""Remove routing prefixes from output string.
Strips known prefixes (``DISCOVERY_RESPONSE:``, ``SET_TOPIC:``,
etc.) and generic ``ROUTE_*:`` patterns.
"""
known_prefixes = [
"DISCOVERY_RESPONSE:",
"GOTO_BRAINSTORMING:",
"SET_TOPIC:",
"COMMAND_OUTPUT:",
]
for prefix in known_prefixes:
if output.startswith(prefix):
output = output[len(prefix) :]
route_match = re.match(r"^ROUTE_[A-Z_]+:", output)
if route_match:
output = output[route_match.end() :]
return output
@staticmethod
def strip_routing_prefixes_multiline(output: str) -> str:
"""Remove routing prefixes from each line of a multi-line output."""
known_prefixes = [
"DISCOVERY_RESPONSE:",
"GOTO_BRAINSTORMING:",
"SET_TOPIC:",
"COMMAND_OUTPUT:",
]
cleaned_lines = []
for line in output.split("\n"):
for prefix in known_prefixes:
if line.startswith(prefix):
line = line[len(prefix) :]
break
else:
route_match = re.match(r"^ROUTE_[A-Z_]+:", line)
if route_match:
line = line[route_match.end() :]
cleaned_lines.append(line)
return "\n".join(cleaned_lines)