diff --git a/.importlinter b/.importlinter new file mode 100644 index 000000000..107b08d3e --- /dev/null +++ b/.importlinter @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md index c78105dfa..f1f295c80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and ## [Unreleased] - **feat(resources): resource type extension interface** (#9998): New `cleveragents.resources` package providing the stable public API third-party developers use to add custom resource types without modifying core code. Includes `ResourceType` ABC with five abstract lifecycle methods (`provision`, `deprovision`, `status`, `validate_config`, `to_dict`), a `ResourceConfig` Pydantic model (`name`, `resource_type`, `properties`), a `ResourceStatus` StrEnum (`PENDING`, `ACTIVE`, `FAILED`, `DEPROVISIONED`), and registry functions `register_resource_type` / `get_resource_type` / `list_resource_types`. Custom types are registered under namespaced names (e.g. `myorg/database`); registration raises `TypeError` for non-`ResourceType` subclasses and `ValueError` for duplicate names. 25 BDD scenarios in `features/resource_type_extension_interface.feature` cover enum values, config instantiation, ABC enforcement, all lifecycle method return types, and registry CRUD + error paths. +- **refactor(a2a): route CLI→Application communication through A2A boundary** (Refs #9962, #4253): Introduced `cleveragents.shared.output_format` as a layer-neutral serialiser (`format_data` supporting `json`/`yaml`/`plain`/`table`) with no dependency on `cleveragents.cli.*`, eliminating a reverse dependency from `PlanApplyService.artifacts()` on the CLI presentation layer. The shared formatter returns raw payloads with no CLI envelope wrapping (`{"data": ..., "command": ..., "status": ...}`); callers that previously parsed `parsed["data"]` from `apply_service.artifacts(fmt="json")` output now read fields at the top level. Updated `features/steps/plan_diff_artifacts_steps.py` (`step_artifacts_json_validation`, `step_artifacts_json_apply_summary`) to drop the stale envelope unwrap that caused `KeyError: 'data'` under the new boundary. Removed stale `@tdd_expected_fail` tag from `WF02 Mocked Generation Produces Test Artifacts Only` in `robot/wf02_test_generation_integration.robot` — the scenario now passes naturally through the A2A facade dispatch path (`_cleveragents/plan/artifacts`) introduced by this refactor. - **fix(test): move advanced context strategy test doubles to features/mocks** (#7574): Extracted `FakeEmbeddings`, `RelevanceScoringStrategy`, `AdaptiveContextSelector`, `ContextFusionStrategy`, and `_pack_budget` from `features/steps/advanced_context_strategies_steps.py` into a new `features/mocks/advanced_context_strategies_mocks.py` file per CONTRIBUTING.md mock-placement rules. Updated the Robot Framework helper `robot/helper_advanced_context_strategies.py` to import directly from `features.mocks` rather than manipulating `sys.path` to reach the Behave steps file. Added `None` guard in `step_assemble_context_query` before calling `selected.assemble()`, and added explicit `ValueError` for unknown strategy types in both `step_load_yaml_strategy` and `load_strategy_from_yaml_impl`. - **fix(a2a): regression tests for stale cleveragents.acp removal** (#5566): Added two Behave BDD scenarios verifying that `cleveragents.acp` is not importable (raises `ImportError`) and that `src/cleveragents/acp/` does not exist in the source tree. These guard against regression of the `__pycache__`-based import that allowed the removed ACP module to still be loaded from bytecode after the v3.6.0 rename to `a2a`. - **Virtual Resource Type Base Class** (#8610): Implemented `VirtualResource` base class with two example concrete implementations (`MetricResource`, `APIEndpointResource`) for abstract/computed resources that are derived rather than mapped to physical files. Virtual resources are computed on demand via a `compute_fn` callable. Includes Behave BDD scenarios in `features/resource_virtual_types.feature` exercising construction, computation, name validation, kwargs passthrough, exception handling, string representation, and subclassing. Resource names are validated against `^[a-zA-Z][a-zA-Z0-9_-]*$` (must start with a letter; alphanumeric, hyphens, and underscores otherwise). diff --git a/features/a2a_boundary_enforcement.feature b/features/a2a_boundary_enforcement.feature new file mode 100644 index 000000000..6616c2599 --- /dev/null +++ b/features/a2a_boundary_enforcement.feature @@ -0,0 +1,91 @@ +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 + + Scenario: format_data handles list of dicts in plain format + Given the shared output_format module is imported + When I call format_data with a list of dicts and format "plain" + Then the result should contain multiple entries separated by dashes + + Scenario: format_data handles dict with list values in plain format + Given the shared output_format module is imported + When I call format_data with a dict containing list values and format "plain" + Then the result should contain list items with dash prefix + + Scenario: format_data handles datetime values in JSON format + Given the shared output_format module is imported + When I call format_data with a dict containing a datetime value and format "json" + Then the result should be valid JSON + And the datetime value should be serialized as ISO string + + Scenario: format_data handles enum values in JSON format + Given the shared output_format module is imported + When I call format_data with a dict containing an enum value and format "json" + Then the result should be valid JSON + And the enum value should be serialized as its value + + Scenario: format_data handles single dict in table format + Given the shared output_format module is imported + When I call format_data with a single dict and format "table" + Then the result should contain column headers + + Scenario: format_data handles list with dict values in table format + Given the shared output_format module is imported + When I call format_data with a list containing dict values and format "table" + Then the result should contain serialized dict values diff --git a/features/steps/a2a_boundary_enforcement_steps.py b/features/steps/a2a_boundary_enforcement_steps.py new file mode 100644 index 000000000..01f35d8cd --- /dev/null +++ b/features/steps/a2a_boundary_enforcement_steps.py @@ -0,0 +1,385 @@ +"""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}: 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}: 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}" + + +@when('I call format_data with a list of dicts and format "plain"') +def step_call_format_data_list_plain(context: Any) -> None: + """Call format_data with list of dicts and plain format.""" + context.sample_data = [ + {"plan_id": "01HXYZ", "status": "applied"}, + {"plan_id": "01HABC", "status": "pending"}, + ] + context.format_result = context.format_data(context.sample_data, "plain") + + +@then("the result should contain multiple entries separated by dashes") +def step_result_contains_multiple_entries(context: Any) -> None: + """Assert the plain result contains multiple entries separated by ---.""" + result = context.format_result + assert "plan_id" in result, f"Missing 'plan_id' in plain output: {result!r}" + assert "---" in result, f"Missing separator '---' in plain output: {result!r}" + + +@when('I call format_data with a dict containing list values and format "plain"') +def step_call_format_data_dict_with_list_plain(context: Any) -> None: + """Call format_data with dict containing list values and plain format.""" + context.sample_data = { + "plan_id": "01HXYZ", + "tags": ["tag1", "tag2", "tag3"], + } + context.format_result = context.format_data(context.sample_data, "plain") + + +@then("the result should contain list items with dash prefix") +def step_result_contains_list_items(context: Any) -> None: + """Assert the plain result contains list items with dash prefix.""" + result = context.format_result + assert "plan_id" in result, f"Missing 'plan_id' in plain output: {result!r}" + assert "tags" in result, f"Missing 'tags' in plain output: {result!r}" + assert "- tag1" in result, f"Missing '- tag1' in plain output: {result!r}" + + +@when('I call format_data with a dict containing a datetime value and format "json"') +def step_call_format_data_datetime_json(context: Any) -> None: + """Call format_data with dict containing datetime value and JSON format.""" + from datetime import UTC, datetime + + context.sample_datetime = datetime(2026, 4, 24, 12, 0, 0, tzinfo=UTC) + context.sample_data = { + "plan_id": "01HXYZ", + "created_at": context.sample_datetime, + } + context.format_result = context.format_data(context.sample_data, "json") + + +@then("the datetime value should be serialized as ISO string") +def step_datetime_serialized_as_iso(context: Any) -> None: + """Assert the datetime value is serialized as ISO string.""" + result = json.loads(context.format_result) + assert "created_at" in result, f"Missing 'created_at' in result: {result}" + assert isinstance(result["created_at"], str), ( + f"Expected str for datetime, got {type(result['created_at'])}" + ) + assert "2026-04-24" in result["created_at"], ( + f"Expected ISO date in 'created_at': {result['created_at']}" + ) + + +@when('I call format_data with a dict containing an enum value and format "json"') +def step_call_format_data_enum_json(context: Any) -> None: + """Call format_data with dict containing enum value and JSON format.""" + from enum import Enum + + class StatusEnum(Enum): + APPLIED = "applied" + PENDING = "pending" + + context.sample_data = { + "plan_id": "01HXYZ", + "status": StatusEnum.APPLIED, + } + context.format_result = context.format_data(context.sample_data, "json") + + +@then("the enum value should be serialized as its value") +def step_enum_serialized_as_value(context: Any) -> None: + """Assert the enum value is serialized as its .value.""" + result = json.loads(context.format_result) + assert "status" in result, f"Missing 'status' in result: {result}" + assert result["status"] == "applied", ( + f"Expected 'applied' for enum value, got {result['status']!r}" + ) + + +@when('I call format_data with a single dict and format "table"') +def step_call_format_data_single_dict_table(context: Any) -> None: + """Call format_data with single dict and table format.""" + context.sample_data = {"plan_id": "01HXYZ", "status": "applied"} + context.format_result = context.format_data(context.sample_data, "table") + + +@when('I call format_data with a list containing dict values and format "table"') +def step_call_format_data_list_with_dict_values_table(context: Any) -> None: + """Call format_data with list containing dict values and table format.""" + context.sample_data = [ + {"plan_id": "01HXYZ", "metadata": {"key": "value"}}, + {"plan_id": "01HABC", "metadata": {"key": "other"}}, + ] + context.format_result = context.format_data(context.sample_data, "table") + + +@then("the result should contain serialized dict values") +def step_result_contains_serialized_dict_values(context: Any) -> None: + """Assert the table result contains serialized dict values.""" + result = context.format_result + assert "plan_id" in result, f"Missing 'plan_id' in table output: {result!r}" + assert "metadata" in result, f"Missing 'metadata' in table output: {result!r}" diff --git a/features/steps/plan_diff_artifacts_steps.py b/features/steps/plan_diff_artifacts_steps.py index 049ce5a46..e34bf739b 100644 --- a/features/steps/plan_diff_artifacts_steps.py +++ b/features/steps/plan_diff_artifacts_steps.py @@ -434,8 +434,7 @@ def step_artifacts_has_files(context: Context) -> None: @then("the artifacts JSON should contain validation summary") def step_artifacts_json_validation(context: Context) -> None: - parsed = json.loads(context.artifacts_output) - data = parsed["data"] + data = json.loads(context.artifacts_output) assert "validation_summary" in data assert data["validation_summary"]["total"] == 3 @@ -565,8 +564,7 @@ def step_plan_with_changeset_id_but_no_store_entry(context: Context) -> None: @then("the artifacts JSON should contain apply summary") def step_artifacts_json_apply_summary(context: Context) -> None: - parsed = json.loads(context.artifacts_output) - data = parsed["data"] + data = json.loads(context.artifacts_output) assert "apply_summary" in data assert data["apply_summary"]["files_changed"] == "7" assert data["apply_summary"]["validations_run"] == "4" diff --git a/pyproject.toml b/pyproject.toml index 09960045b..1479fdebc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -84,6 +84,8 @@ dev = [ "vulture>=2.10", # Complexity metrics "radon>=6.0.1", + # Import boundary enforcement + "import-linter>=2.0", ] tests = [ "behave==1.3.3", diff --git a/robot/wf02_test_generation_integration.robot b/robot/wf02_test_generation_integration.robot index eb9bea4d9..4f2eaca08 100644 --- a/robot/wf02_test_generation_integration.robot +++ b/robot/wf02_test_generation_integration.robot @@ -28,7 +28,7 @@ WF02 Mocked Generation Produces Test Artifacts Only [Documentation] Verify mocked LLM responses generate deterministic test ... files, assert user-facing plan artifacts command path, ... and artifacts list only test paths (no prod edits). - [Tags] tdd_issue tdd_issue_4296 tdd_expected_fail mocking artifacts invariants + [Tags] tdd_issue tdd_issue_4296 mocking artifacts invariants ${result}= Run Process ${PYTHON} ${HELPER} mocked-generation-artifacts cwd=${WORKSPACE} timeout=120s on_timeout=kill Log ${result.stdout} Log ${result.stderr} diff --git a/src/cleveragents/application/services/plan_apply_service.py b/src/cleveragents/application/services/plan_apply_service.py index bf7bee8a6..78a4e9387 100644 --- a/src/cleveragents/application/services/plan_apply_service.py +++ b/src/cleveragents/application/services/plan_apply_service.py @@ -646,7 +646,7 @@ class PlanApplyService: Returns: 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) changeset = self._resolve_changeset(plan) diff --git a/src/cleveragents/shared/output_format.py b/src/cleveragents/shared/output_format.py new file mode 100644 index 000000000..8e684275c --- /dev/null +++ b/src/cleveragents/shared/output_format.py @@ -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", +]