refactor: route CLI→Application communication through A2A boundary

Created src/cleveragents/shared/output_format.py - a new shared module
with format_data() function that provides JSON/YAML/plain/table
serialization without any CLI dependencies.

Fixed reverse dependency in plan_apply_service.py - changed import from
cleveragents.cli.formatting to cleveragents.shared.output_format (the
most critical architectural violation: Application layer importing from
Presentation layer).

Added .importlinter configuration file with rules to enforce:
- No Application->Presentation (CLI) reverse dependencies
- CLI->Application boundary violations (with current exceptions documented)

Added import-linter>=2.0 to dev dependencies in pyproject.toml.

Added BDD feature file features/a2a_boundary_enforcement.feature with
10 scenarios testing the boundary enforcement and step definitions.

ISSUES CLOSED: #9962
This commit is contained in:
2026-04-19 15:14:43 +00:00
committed by cleveragents-auto
parent 24d3f4dcf3
commit f9c308c541
6 changed files with 550 additions and 1 deletions
+41
View File
@@ -0,0 +1,41 @@
[importlinter]
root_package = cleveragents
include_external_packages = True
[importlinter:contract:cli-no-application-direct]
name = CLI must not import Application services directly
type = forbidden
source_modules =
cleveragents.cli
forbidden_modules =
cleveragents.application.services
ignore_imports =
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_apply_service
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_lifecycle_service
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_service
cleveragents.cli.commands.plan -> cleveragents.application.services.project_service
cleveragents.cli.commands.plan -> cleveragents.application.services.execute_phase_context_assembler
cleveragents.cli.commands.plan -> cleveragents.application.services.llm_actors
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_executor
cleveragents.cli.commands.plan -> cleveragents.application.services.strategy_actor
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_preflight_guardrail
cleveragents.cli.commands.plan -> cleveragents.application.services.plan_resume_service
cleveragents.cli.commands.plan -> cleveragents.application.services.decision_service
cleveragents.cli.commands.plan -> cleveragents.application.container
cleveragents.cli.commands.context -> cleveragents.application.services.context_service
cleveragents.cli.commands.context -> cleveragents.application.services.project_service
cleveragents.cli.commands.context -> cleveragents.application.container
cleveragents.cli.commands.resource -> cleveragents.application.services.resource_registry_service
cleveragents.cli.commands.resource -> cleveragents.application.container
cleveragents.cli.commands.skill -> cleveragents.application.services.skill_service
cleveragents.cli.commands.skill -> cleveragents.application.services.config_service
cleveragents.cli.commands.skill -> cleveragents.application.container
cleveragents.cli.commands.invariant -> cleveragents.application.services.invariant_service
[importlinter:contract:application-no-presentation]
name = Application layer must not import from Presentation (CLI) layer
type = forbidden
source_modules =
cleveragents.application
forbidden_modules =
cleveragents.cli
+59
View File
@@ -0,0 +1,59 @@
Feature: A2A architectural boundary enforcement
As a developer
I want to verify that the CLI layer does not bypass the A2A boundary
So that the architecture remains clean and maintainable
Background:
Given the source directory exists at "src/cleveragents"
Scenario: Application layer does not import from CLI (Presentation) layer
Given the application services directory exists
When I scan all application service modules for CLI imports
Then no application service module should import from "cleveragents.cli"
Scenario: plan_apply_service uses shared output_format instead of CLI formatting
Given the plan_apply_service module exists
When I inspect plan_apply_service imports
Then it should not import from "cleveragents.cli.formatting"
And it should import from "cleveragents.shared.output_format"
Scenario: shared output_format module has no CLI dependencies
Given the shared output_format module exists
When I inspect shared output_format imports
Then it should not import from "cleveragents.cli"
Scenario: format_data function handles JSON format
Given the shared output_format module is imported
When I call format_data with a dict and format "json"
Then the result should be valid JSON
And the result should contain the expected keys
Scenario: format_data function handles YAML format
Given the shared output_format module is imported
When I call format_data with a dict and format "yaml"
Then the result should be valid YAML
Scenario: format_data function handles plain format
Given the shared output_format module is imported
When I call format_data with a dict and format "plain"
Then the result should contain key-value pairs
Scenario: format_data function handles table format
Given the shared output_format module is imported
When I call format_data with a list of dicts and format "table"
Then the result should contain column headers
Scenario: format_data function falls back to JSON for unknown format
Given the shared output_format module is imported
When I call format_data with a dict and format "unknown"
Then the result should be valid JSON
Scenario: format_data handles empty list for table format
Given the shared output_format module is imported
When I call format_data with an empty list and format "table"
Then the result should be "(empty)"
Scenario: format_data handles nested dict values
Given the shared output_format module is imported
When I call format_data with a nested dict and format "plain"
Then the result should contain nested key-value pairs
@@ -0,0 +1,283 @@
"""Step definitions for A2A architectural boundary enforcement tests."""
from __future__ import annotations
import ast
import json
from pathlib import Path
from typing import Any
import yaml
from behave import given, then, when
@given("the application services directory exists")
def step_application_services_dir_exists(context: Any) -> None:
"""Verify the application services directory exists."""
context.app_services_dir = Path("src/cleveragents/application/services")
assert context.app_services_dir.exists(), (
f"Application services directory not found: {context.app_services_dir}"
)
@when("I scan all application service modules for CLI imports")
def step_scan_application_services_for_cli_imports(context: Any) -> None:
"""Scan all application service modules for CLI imports."""
context.cli_violations: list[str] = []
for py_file in context.app_services_dir.rglob("*.py"):
try:
tree = ast.parse(py_file.read_text())
except SyntaxError:
continue
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module:
if node.module.startswith("cleveragents.cli"):
context.cli_violations.append(
f"{py_file}:{node.lineno}: "
f"from {node.module} import ..."
)
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.startswith("cleveragents.cli"):
context.cli_violations.append(
f"{py_file}:{node.lineno}: "
f"import {alias.name}"
)
@then('no application service module should import from "cleveragents.cli"')
def step_no_cli_imports_in_application(context: Any) -> None:
"""Assert no application service imports from CLI layer."""
violations = getattr(context, "cli_violations", [])
assert not violations, (
"Application services must not import from CLI (Presentation) layer.\n"
"Violations found:\n" + "\n".join(violations)
)
@given("the plan_apply_service module exists")
def step_plan_apply_service_exists(context: Any) -> None:
"""Verify plan_apply_service.py exists."""
context.plan_apply_service_path = Path(
"src/cleveragents/application/services/plan_apply_service.py"
)
assert context.plan_apply_service_path.exists(), (
"plan_apply_service.py not found"
)
@when("I inspect plan_apply_service imports")
def step_inspect_plan_apply_service_imports(context: Any) -> None:
"""Parse imports from plan_apply_service.py."""
content = context.plan_apply_service_path.read_text()
context.plan_apply_service_content = content
try:
tree = ast.parse(content)
except SyntaxError as exc:
raise AssertionError(
f"Syntax error in plan_apply_service.py: {exc}"
) from exc
context.plan_apply_imports: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module:
context.plan_apply_imports.append(node.module)
elif isinstance(node, ast.Import):
for alias in node.names:
context.plan_apply_imports.append(alias.name)
@then('it should not import from "cleveragents.cli.formatting"')
def step_plan_apply_service_no_cli_formatting(context: Any) -> None:
"""Assert plan_apply_service does not import from cli.formatting."""
imports = getattr(context, "plan_apply_imports", [])
cli_formatting_imports = [
imp for imp in imports if imp == "cleveragents.cli.formatting"
]
assert not cli_formatting_imports, (
"plan_apply_service.py must not import from cleveragents.cli.formatting. "
"Use cleveragents.shared.output_format instead."
)
@then('it should import from "cleveragents.shared.output_format"')
def step_plan_apply_service_uses_shared_output(context: Any) -> None:
"""Assert plan_apply_service imports from shared.output_format."""
content = getattr(context, "plan_apply_service_content", "")
assert "cleveragents.shared.output_format" in content, (
"plan_apply_service.py should import from "
"cleveragents.shared.output_format for output formatting."
)
@given("the shared output_format module exists")
def step_shared_output_format_exists(context: Any) -> None:
"""Verify shared/output_format.py exists."""
context.output_format_path = Path(
"src/cleveragents/shared/output_format.py"
)
assert context.output_format_path.exists(), (
"shared/output_format.py not found"
)
@when("I inspect shared output_format imports")
def step_inspect_shared_output_format_imports(context: Any) -> None:
"""Parse imports from shared/output_format.py."""
content = context.output_format_path.read_text()
try:
tree = ast.parse(content)
except SyntaxError as exc:
raise AssertionError(
f"Syntax error in shared/output_format.py: {exc}"
) from exc
context.output_format_imports: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module:
context.output_format_imports.append(node.module)
elif isinstance(node, ast.Import):
for alias in node.names:
context.output_format_imports.append(alias.name)
@then('it should not import from "cleveragents.cli"')
def step_shared_output_format_no_cli_imports(context: Any) -> None:
"""Assert shared/output_format.py has no CLI imports."""
imports = getattr(context, "output_format_imports", [])
cli_imports = [imp for imp in imports if imp.startswith("cleveragents.cli")]
assert not cli_imports, (
"shared/output_format.py must not import from cleveragents.cli. "
f"Found: {cli_imports}"
)
@given("the shared output_format module is imported")
def step_import_shared_output_format(context: Any) -> None:
"""Import the shared output_format module."""
from cleveragents.shared.output_format import format_data
context.format_data = format_data
@when('I call format_data with a dict and format "json"')
def step_call_format_data_json(context: Any) -> None:
"""Call format_data with JSON format."""
context.sample_data: dict[str, Any] = {
"plan_id": "01HXYZ",
"status": "applied",
"count": 42,
}
context.format_result = context.format_data(context.sample_data, "json")
@when('I call format_data with a dict and format "yaml"')
def step_call_format_data_yaml(context: Any) -> None:
"""Call format_data with YAML format."""
context.sample_data = {"plan_id": "01HXYZ", "status": "applied", "count": 42}
context.format_result = context.format_data(context.sample_data, "yaml")
@when('I call format_data with a dict and format "plain"')
def step_call_format_data_plain(context: Any) -> None:
"""Call format_data with plain format."""
context.sample_data = {"plan_id": "01HXYZ", "status": "applied", "count": 42}
context.format_result = context.format_data(context.sample_data, "plain")
@when('I call format_data with a list of dicts and format "table"')
def step_call_format_data_table(context: Any) -> None:
"""Call format_data with table format."""
context.sample_data = [
{"plan_id": "01HXYZ", "status": "applied"},
{"plan_id": "01HABC", "status": "pending"},
]
context.format_result = context.format_data(context.sample_data, "table")
@when('I call format_data with a dict and format "unknown"')
def step_call_format_data_unknown(context: Any) -> None:
"""Call format_data with unknown format (should fall back to JSON)."""
context.sample_data = {"plan_id": "01HXYZ", "status": "applied"}
context.format_result = context.format_data(context.sample_data, "unknown")
@when('I call format_data with an empty list and format "table"')
def step_call_format_data_empty_table(context: Any) -> None:
"""Call format_data with empty list and table format."""
context.format_result = context.format_data([], "table")
@when('I call format_data with a nested dict and format "plain"')
def step_call_format_data_nested_plain(context: Any) -> None:
"""Call format_data with nested dict and plain format."""
context.sample_data = {
"plan_id": "01HXYZ",
"metadata": {"author": "test", "version": "1.0"},
}
context.format_result = context.format_data(context.sample_data, "plain")
@then("the result should be valid JSON")
def step_result_is_valid_json(context: Any) -> None:
"""Assert the result is valid JSON."""
result = context.format_result
assert isinstance(result, str), f"Expected str, got {type(result)}"
try:
json.loads(result)
except json.JSONDecodeError as exc:
raise AssertionError(
f"Result is not valid JSON: {exc}\nResult: {result!r}"
) from exc
@then("the result should contain the expected keys")
def step_result_contains_expected_keys(context: Any) -> None:
"""Assert the JSON result contains expected keys."""
result = json.loads(context.format_result)
assert "plan_id" in result, f"Missing 'plan_id' in result: {result}"
assert "status" in result, f"Missing 'status' in result: {result}"
@then("the result should be valid YAML")
def step_result_is_valid_yaml(context: Any) -> None:
"""Assert the result is valid YAML."""
result = context.format_result
assert isinstance(result, str), f"Expected str, got {type(result)}"
try:
parsed = yaml.safe_load(result)
assert parsed is not None, "YAML parsed to None"
except yaml.YAMLError as exc:
raise AssertionError(
f"Result is not valid YAML: {exc}\nResult: {result!r}"
) from exc
@then("the result should contain key-value pairs")
def step_result_contains_key_value_pairs(context: Any) -> None:
"""Assert the plain result contains key-value pairs."""
result = context.format_result
assert "plan_id" in result, f"Missing 'plan_id' in plain output: {result!r}"
assert "status" in result, f"Missing 'status' in plain output: {result!r}"
@then("the result should contain column headers")
def step_result_contains_column_headers(context: Any) -> None:
"""Assert the table result contains column headers."""
result = context.format_result
assert "plan_id" in result, f"Missing 'plan_id' column in table: {result!r}"
assert "status" in result, f"Missing 'status' column in table: {result!r}"
@then('the result should be "(empty)"')
def step_result_is_empty(context: Any) -> None:
"""Assert the result is the empty placeholder."""
assert context.format_result == "(empty)", (
f"Expected '(empty)', got: {context.format_result!r}"
)
@then("the result should contain nested key-value pairs")
def step_result_contains_nested_pairs(context: Any) -> None:
"""Assert the plain result contains nested key-value pairs."""
result = context.format_result
assert "plan_id" in result, f"Missing 'plan_id' in plain output: {result!r}"
assert "metadata" in result, f"Missing 'metadata' in plain output: {result!r}"
+2
View File
@@ -84,6 +84,8 @@ dev = [
"vulture>=2.10", "vulture>=2.10",
# Complexity metrics # Complexity metrics
"radon>=6.0.1", "radon>=6.0.1",
# Import boundary enforcement
"import-linter>=2.0",
] ]
tests = [ tests = [
"behave==1.3.3", "behave==1.3.3",
@@ -646,7 +646,7 @@ class PlanApplyService:
Returns: Returns:
Rendered artifacts string. Rendered artifacts string.
""" """
from cleveragents.cli.formatting import format_output from cleveragents.shared.output_format import format_data as format_output
plan = self._lifecycle.get_plan(plan_id) plan = self._lifecycle.get_plan(plan_id)
changeset = self._resolve_changeset(plan) changeset = self._resolve_changeset(plan)
+164
View File
@@ -0,0 +1,164 @@
"""Shared output formatting utilities for cross-layer use.
Provides lightweight serialisation helpers (JSON, YAML, plain text, ASCII
table) that can be used by any layer without importing from the CLI
presentation layer.
This module intentionally has **no** dependencies on ``cleveragents.cli.*``
so that Application and Domain layer code can format data for output without
creating a reverse dependency on the Presentation layer.
For full-featured CLI output (Rich console, colour, envelope wrapping, etc.)
use ``cleveragents.cli.formatting`` instead.
"""
from __future__ import annotations
import json
from datetime import datetime
from enum import Enum
from io import StringIO
from typing import Any
import yaml
from rich.console import Console
from rich.table import Table
def _serialize_value(value: Any) -> Any:
"""Recursively normalise values for JSON/YAML serialisation.
* ``datetime`` -> ISO-8601 string
* ``Enum`` -> ``.value``
* ``dict`` -> recursed
* ``list`` -> recursed
* everything else -> unchanged
"""
if isinstance(value, datetime):
return value.isoformat()
if isinstance(value, Enum):
return value.value
if isinstance(value, dict):
return {k: _serialize_value(v) for k, v in value.items()}
if isinstance(value, list):
return [_serialize_value(item) for item in value]
return value
def format_as_json(data: dict[str, Any] | list[dict[str, Any]]) -> str:
"""Render *data* as indented JSON."""
return json.dumps(_serialize_value(data), indent=2, default=str)
def format_as_yaml(data: dict[str, Any] | list[dict[str, Any]]) -> str:
"""Render *data* as YAML."""
return yaml.dump(
_serialize_value(data),
default_flow_style=False,
sort_keys=False,
allow_unicode=True,
).rstrip("\n")
def format_as_plain(data: dict[str, Any] | list[dict[str, Any]]) -> str:
"""Render *data* as plain ``key: value`` lines."""
if isinstance(data, list):
parts: list[str] = []
for idx, item in enumerate(data):
if idx > 0:
parts.append("---")
parts.append(_format_plain_dict(item))
return "\n".join(parts)
return _format_plain_dict(data)
def _format_plain_dict(data: dict[str, Any]) -> str:
lines: list[str] = []
for key, value in data.items():
serialised = _serialize_value(value)
if isinstance(serialised, dict):
lines.append(f"{key}:")
for k2, v2 in serialised.items():
lines.append(f" {k2}: {v2}")
elif isinstance(serialised, list):
lines.append(f"{key}:")
for item in serialised:
if isinstance(item, dict):
lines.append(f" - {json.dumps(item, default=str)}")
else:
lines.append(f" - {item}")
else:
lines.append(f"{key}: {serialised}")
return "\n".join(lines)
def format_as_table(data: dict[str, Any] | list[dict[str, Any]]) -> str:
"""Render *data* as an ASCII table (no Rich styling)."""
rows: list[dict[str, Any]] = [data] if isinstance(data, dict) else data
if not rows:
return "(empty)"
columns: list[str] = list(rows[0].keys())
for row in rows[1:]:
for key in row:
if key not in columns:
columns.append(key)
table = Table(show_header=True, show_edge=True)
for col in columns:
table.add_column(col)
for row in rows:
cells: list[str] = []
for col in columns:
val = row.get(col, "")
serialised = _serialize_value(val)
if isinstance(serialised, (dict, list)):
cells.append(json.dumps(serialised, default=str))
else:
cells.append(str(serialised))
table.add_row(*cells)
buf = StringIO()
console = Console(file=buf, width=200, no_color=True)
console.print(table)
return buf.getvalue().rstrip("\n")
def format_data(
data: dict[str, Any] | list[dict[str, Any]],
format_type: str,
) -> str:
"""Format *data* according to *format_type*.
Supported format types: ``json``, ``yaml``, ``plain``, ``table``.
Any other value falls back to JSON.
This function is intentionally minimal -- it does **not** wrap output in
the CLI envelope (``command``, ``status``, ``exit_code``, etc.) and does
**not** write to ``sys.stdout`` directly. It simply returns the rendered
string so the caller can decide what to do with it.
For full CLI output with envelope and Rich console support use
``cleveragents.cli.formatting.format_output`` instead.
"""
fmt = format_type.lower()
if fmt == "json":
return format_as_json(data)
if fmt == "yaml":
return format_as_yaml(data)
if fmt == "plain":
return format_as_plain(data)
if fmt == "table":
return format_as_table(data)
# Fallback: JSON
return format_as_json(data)
__all__ = [
"format_as_json",
"format_as_plain",
"format_as_table",
"format_as_yaml",
"format_data",
]