Compare commits

...

1 Commits

Author SHA1 Message Date
freemo 53cc7a6786 fix(cli): add missing Origin panel, Overridden field, Winner indicator, and JSON envelope in config get output
Implements all spec-required output for `agents config get`:

Rich output:
- Rename panel title from 'Configuration Value' to 'Config' per spec
- Add 'Overridden' field to the Config panel
- Add 'Origin' panel showing File, Line, and Default fields
- Add 'Resolution Chain' panel (always shown, not just with --verbose)
- Add 'Winner' indicator to the Resolution Chain panel
- Use spec-required type strings (string/boolean/integer) instead of
  Python type names (str/bool/int)
- Add '✓ OK Config read' confirmation message

JSON output:
- Wrap result in standard envelope (command, status, exit_code, data,
  timing, messages)
- Add 'overridden' field to data
- Add 'origin' nested object (file, line, default)
- Add 'winner' nested object (source, level)
- Use human-readable source names (CLI flag, Env var, Config file,
  Default) instead of internal enum values
- Add 'started' timestamp to timing

BDD:
- Add features/config_get_spec_output.feature with 20 scenarios
  covering all Rich panels and JSON envelope fields
- Update existing tests to match new spec-compliant output

ISSUES CLOSED: #3423
2026-05-07 21:55:31 +00:00
12 changed files with 846 additions and 198 deletions
+11
View File
@@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
### Added
- **CLI — `config get` spec-compliant output** (#3423): The `agents config get` command now
renders the full specification-required output including three Rich panels (*Config*, *Origin*,
*Resolution Chain*) with Overridden indicator, Winner line, and Origin details (File, Line,
Default). JSON output is wrapped in a standard envelope with `command`, `status`, `exit_code`,
`data` (key, value, source short-name, overridden, type spec-string, origin, resolution_chain,
winner), `timing` (started ISO-8601 + duration_ms), and messages. Added module-level helpers in
`_config_display.py` and `_config_helpers.py`, plus 18 new BDD scenarios in
`config_get_spec_output.feature`.
### Fixed
- **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed
+2 -2
View File
@@ -78,14 +78,14 @@ Feature: Config CLI commands
# --- RESOLUTION CHAIN ---
Scenario: Resolution chain shows default source
When I run config get key "core.log.level" formatted as "json"
Then the resolution chain should include "default" source
Then the resolution chain should include "Default" source
# --- SET/GET ROUNDTRIP ---
Scenario: Set and get roundtrip
Given I have set config "core.log.file-enabled" to "true"
When I run config get key "core.log.file-enabled" formatted as "json"
Then the config get should succeed
And the get value source should be "global"
And the get value source should be "config"
# --- FORMAT ---
Scenario: Config list with JSON format
@@ -197,8 +197,8 @@ Feature: Config CLI safety-net coverage
Scenario: safety-net config get rich format displays panel and chain
Given a safety-net isolated temp config directory
When the safety-net CLI gets key "core.log.level" with format "rich" and verbose
Then the safety-net get rich output should contain "Configuration Value"
And the safety-net get rich output should contain "Resolution chain"
Then the safety-net get rich output should contain "Config"
And the safety-net get rich output should contain "Resolution Chain"
@tdd_issue @tdd_issue_4238 @tdd_expected_fail
Scenario: safety-net config get yaml format produces valid YAML
@@ -285,4 +285,4 @@ Feature: Config CLI safety-net coverage
Given a safety-net isolated temp config directory
And a safety-net patched console for capturing rich output
When the safety-net CLI gets key "core.log.level" with format "rich" and verbose
Then the safety-net captured console output should contain "active"
Then the safety-net captured console output should contain "Winner"
+120
View File
@@ -0,0 +1,120 @@
Feature: agents config get spec-compliant output
As a developer
I want `agents config get` to render all spec-required panels and JSON fields
So that the output matches the specification exactly
Background:
Given a clean config get spec test environment
# --- Rich output: three panels ---
Scenario: Rich output shows Config panel with correct title
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain panel "Config"
Scenario: Rich output shows Overridden field in Config panel
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain "Overridden"
Scenario: Rich output shows Origin panel
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain panel "Origin"
Scenario: Rich output shows Resolution Chain panel
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain panel "Resolution Chain"
Scenario: Rich output shows Winner in Resolution Chain panel
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain "Winner"
Scenario: Rich output uses spec type string not Python type name
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain "string"
And the spec config get rich output should not contain "str"
Scenario: Rich output shows OK confirmation message
When I run spec config get "core.log.level"
Then the spec config get should succeed
And the spec config get rich output should contain "OK"
# --- JSON output: standard envelope ---
Scenario: JSON output is wrapped in standard envelope
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON envelope should have field "command" equal to "config get"
And the spec config get JSON envelope should have field "status" equal to "ok"
And the spec config get JSON envelope should have field "exit_code" equal to 0
And the spec config get JSON envelope should have field "data"
And the spec config get JSON envelope should have field "timing"
And the spec config get JSON envelope should have field "messages"
Scenario: JSON data contains all required fields
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data should have field "key"
And the spec config get JSON data should have field "value"
And the spec config get JSON data should have field "source"
And the spec config get JSON data should have field "overridden"
And the spec config get JSON data should have field "type"
And the spec config get JSON data should have field "origin"
And the spec config get JSON data should have field "resolution_chain"
And the spec config get JSON data should have field "winner"
Scenario: JSON data type field uses spec type string
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data field "type" should equal "string"
Scenario: JSON data overridden field is false for default value
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data field "overridden" should be false
Scenario: JSON data overridden field is true when value is set
Given I have set spec config "core.log.level" to "DEBUG"
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data field "overridden" should be true
Scenario: JSON data origin contains default field
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON origin should have field "default"
Scenario: JSON data resolution chain uses human-readable source names
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON resolution chain should use human-readable source names
Scenario: JSON data winner contains source and level
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON winner should have field "source"
And the spec config get JSON winner should have field "level"
Scenario: JSON timing contains started and duration_ms
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON timing should have field "started"
And the spec config get JSON timing should have field "duration_ms"
# --- Source name mapping ---
Scenario: JSON source uses short name "config" when value from global config
Given I have set spec config "core.log.level" to "WARNING"
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data field "source" should equal "config"
Scenario: JSON source uses short name "default" when value from default
When I run spec config get "core.log.level" with format "json"
Then the spec config get should succeed
And the spec config get JSON data field "source" should equal "default"
@@ -41,12 +41,16 @@ from cleveragents.application.services.config_service import (
ResolvedValue,
)
from cleveragents.cli.commands import config as config_mod
from cleveragents.cli.commands.config import (
from cleveragents.cli.commands._config_helpers import (
_env_var_for_key,
)
from cleveragents.cli.commands.config import (
_resolution_chain,
_resolve_source,
)
from cleveragents.cli.commands.config import app as config_app
from cleveragents.cli.commands.config import (
app as config_app,
)
_runner = CliRunner()
@@ -262,6 +266,7 @@ def step_cfg_boost_get_verbose_path(context: Context, key: str, fmt: str) -> Non
chain_with_path = [
{"source": "cli_flag", "value": None},
{"source": "env_var", "value": None},
{"source": "local", "value": None},
{"source": "project", "value": None},
{"source": "global", "value": Path("/tmp/fake/path")},
{"source": "default", "value": Path("/default/path")},
@@ -284,7 +289,7 @@ def step_cfg_boost_get_verbose_path(context: Context, key: str, fmt: str) -> Non
@then("the cfg-boost JSON resolution chain entries should have string values")
def step_cfg_boost_json_chain_strings(context: Context) -> None:
parsed = json.loads(context.cfg_boost_result.output)
parsed = json.loads(context.cfg_boost_result.output)
data = _unwrap_envelope(parsed)
chain = data.get("resolution_chain", [])
assert len(chain) > 0, "Expected non-empty resolution chain"
@@ -26,19 +26,23 @@ from rich.console import Console
from typer.testing import CliRunner
from cleveragents.cli.commands import config as config_mod
from cleveragents.cli.commands.config import (
from cleveragents.cli.commands._config_helpers import (
_env_var_for_key,
_is_secret_key,
_mask_value,
_normalize_key,
_validate_key,
)
from cleveragents.cli.commands.config import (
_read_config_file,
_resolution_chain,
_resolve_source,
_settings_fields,
_validate_key,
_write_config_file,
)
from cleveragents.cli.commands.config import app as config_app
from cleveragents.cli.commands.config import (
app as config_app,
)
_runner = CliRunner()
@@ -529,16 +533,17 @@ def step_sn_get_valid_yaml(context: Context) -> None:
@then('the safety-net get YAML should contain key "{key}"')
def step_sn_get_yaml_key(context: Context, key: str) -> None:
assert key in context._sn_get_yaml, (
f"Key '{key}' not in YAML: {list(context._sn_get_yaml.keys())}"
)
# YAML output is now wrapped in a standard envelope; fields are in data
yaml_data = context._sn_get_yaml
inner = yaml_data.get("data", yaml_data)
assert key in inner, f"Key '{key}' not in YAML data: {list(inner.keys())}"
@then("the safety-net get output should be valid JSON with type field")
def step_sn_get_json_type(context: Context) -> None:
result = context._sn_result
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
parsed = json.loads(result.output.strip())
parsed = json.loads(result.output.strip())
data = _unwrap_envelope(parsed)
assert "type" in data, f"'type' not in JSON keys: {list(data.keys())}"
assert "key" in data, f"'key' not in JSON keys: {list(data.keys())}"
+5 -3
View File
@@ -143,9 +143,11 @@ def step_config_get_contains_key(context: Context, key: str) -> None:
@then("the config get output should show the resolution chain")
def step_config_get_resolution_chain(context: Context) -> None:
output = context.result.output
assert "Resolution chain" in output or "resolution_chain" in output, (
f"Expected resolution chain in output: {output}"
)
assert (
"Resolution Chain" in output
or "Resolution chain" in output
or "resolution_chain" in output
), f"Expected resolution chain in output: {output}"
@then("the config get should fail with unknown key error")
@@ -33,10 +33,12 @@ from cleveragents.application.services.config_service import (
ResolvedValue,
)
from cleveragents.cli.commands import config as config_mod
from cleveragents.cli.commands.config import (
_resolve_source,
from cleveragents.cli.commands._config_helpers import (
_settings_defaults,
_validate_key,
)
from cleveragents.cli.commands.config import (
_resolve_source,
_write_config_file,
)
from cleveragents.cli.commands.config import (
@@ -336,9 +338,11 @@ def step_get_verbose_succeed(context: Context) -> None:
def step_get_verbose_chain(context: Context) -> None:
result = context._cfg_branch_result
output = result.output
assert "Resolution chain" in output or "resolution_chain" in output, (
f"Expected resolution chain in output: {output}"
)
assert (
"Resolution Chain" in output
or "Resolution chain" in output
or "resolution_chain" in output
), f"Expected resolution chain in output: {output}"
# ===================================================================
@@ -362,6 +366,7 @@ def step_mock_resolve_path(context: Context, key: str) -> None:
"value": None,
"env_name": f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}",
},
{"source": ConfigLevel.LOCAL.value, "value": None},
{"source": ConfigLevel.PROJECT.value, "value": None},
{
"source": ConfigLevel.GLOBAL.value,
@@ -404,8 +409,10 @@ def step_get_result_json(context: Context) -> None:
@then("the config cli branch get JSON value should be a string not a Path")
def step_get_json_value_string(context: Context) -> None:
data = context._cfg_branch_get_json
# The top-level "value" should be a string (serialised from Path)
envelope = context._cfg_branch_get_json
# JSON output is now wrapped in a standard envelope; value is in data
data = envelope.get("data", envelope)
# The "value" should be a string (serialised from Path)
assert isinstance(data["value"], str), (
f"Expected str, got {type(data['value'])}: {data['value']}"
)
@@ -0,0 +1,296 @@
"""Step definitions for config_get_spec_output.feature.
Tests that `agents config get` output matches the specification:
- Rich output: three panels (Config, Origin, Resolution Chain)
- JSON output: standard envelope with all required fields
- Spec-required type strings and human-readable source names
"""
from __future__ import annotations
import json
import os
import re
import shutil
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import patch
from behave import given, then, when
from behave.runner import Context
from typer.testing import CliRunner
from cleveragents.cli.commands import config as config_mod
from cleveragents.cli.commands.config import app as config_app
_runner = CliRunner()
# Human-readable source names per spec
_SPEC_SOURCE_NAMES = {
"CLI flag",
"Env var",
"Local file",
"Project file",
"Config file",
"Default",
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _setup_spec_temp(context: Context) -> None:
"""Redirect config dir/path to a fresh temp directory."""
if not hasattr(context, "_spec_tmpdir"):
context._spec_tmpdir = tempfile.mkdtemp(prefix="cfg_spec_")
tmp = Path(context._spec_tmpdir)
context._spec_dir_patch = patch.object(config_mod, "_CONFIG_DIR", tmp)
context._spec_path_patch = patch.object(
config_mod, "_CONFIG_PATH", tmp / "config.toml"
)
context._spec_dir_patch.start()
context._spec_path_patch.start()
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers: list[Any] = []
context._cleanup_handlers.append(lambda: _teardown_spec_temp(context))
def _teardown_spec_temp(context: Context) -> None:
for attr in ("_spec_dir_patch", "_spec_path_patch"):
patcher = getattr(context, attr, None)
if patcher is not None and getattr(patcher, "_active", True):
patcher.stop()
tmpdir = getattr(context, "_spec_tmpdir", None)
if tmpdir and os.path.isdir(tmpdir):
shutil.rmtree(tmpdir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a clean config get spec test environment")
def step_spec_clean_env(context: Context) -> None:
_setup_spec_temp(context)
# ---------------------------------------------------------------------------
# When
# ---------------------------------------------------------------------------
@when('I run spec config get "{key}"')
def step_spec_run_get(context: Context, key: str) -> None:
context.spec_result = _runner.invoke(config_app, ["get", key])
@when('I run spec config get "{key}" with format "{fmt}"')
def step_spec_run_get_fmt(context: Context, key: str, fmt: str) -> None:
context.spec_result = _runner.invoke(config_app, ["get", key, "--format", fmt])
@given('I have set spec config "{key}" to "{value}"')
def step_spec_preset_config(context: Context, key: str, value: str) -> None:
result = _runner.invoke(config_app, ["set", key, value])
assert result.exit_code == 0, f"Pre-set failed: {result.output}"
# ---------------------------------------------------------------------------
# Then — basic
# ---------------------------------------------------------------------------
@then("the spec config get should succeed")
def step_spec_get_succeed(context: Context) -> None:
assert context.spec_result.exit_code == 0, (
f"Expected exit 0, got {context.spec_result.exit_code}: "
f"{context.spec_result.output}"
)
@then('the spec config get rich output should contain panel "{panel_title}"')
def step_spec_rich_contains_panel(context: Context, panel_title: str) -> None:
output = context.spec_result.output
assert panel_title in output, (
f"Expected panel '{panel_title}' in Rich output: {output}"
)
@then('the spec config get rich output should contain "{text}"')
def step_spec_rich_contains(context: Context, text: str) -> None:
output = context.spec_result.output
assert text in output, f"Expected '{text}' in Rich output: {output}"
@then('the spec config get rich output should not contain "{text}"')
def step_spec_rich_not_contains(context: Context, text: str) -> None:
output = context.spec_result.output
# Check that the text doesn't appear as a standalone type name
# (e.g. "str" should not appear but "string" is fine)
# We check for exact word boundary to avoid false positives
# Look for the text as a standalone word (not part of "string", "integer", etc.)
pattern = rf"\b{re.escape(text)}\b"
# Exclude lines that contain "string", "integer", "boolean" etc.
lines = output.split("\n")
for line in lines:
if re.search(pattern, line):
# Check if this is a false positive (e.g. "str" inside "string")
# Only flag if the match is not part of a longer type word
if text == "str" and "string" in line:
continue
if text == "bool" and "boolean" in line:
continue
if text == "int" and "integer" in line:
continue
raise AssertionError(
f"Found '{text}' as standalone word in Rich output line: {line!r}"
)
# ---------------------------------------------------------------------------
# Then — JSON envelope
# ---------------------------------------------------------------------------
def _get_spec_json(context: Context) -> dict[str, Any]:
"""Parse and cache the JSON output."""
if not hasattr(context, "_spec_json"):
context._spec_json = json.loads(context.spec_result.output.strip())
return context._spec_json
@then(
'the spec config get JSON envelope should have field "{field}" equal to "{value}"'
)
def step_spec_json_envelope_field_str(context: Context, field: str, value: str) -> None:
envelope = _get_spec_json(context)
assert field in envelope, (
f"Field '{field}' not in envelope: {list(envelope.keys())}"
)
assert str(envelope[field]) == value, (
f"Expected envelope['{field}'] == '{value}', got {envelope[field]!r}"
)
@then(
'the spec config get JSON envelope should have field "{field}" equal to {value:d}'
)
def step_spec_json_envelope_field_int(context: Context, field: str, value: int) -> None:
envelope = _get_spec_json(context)
assert field in envelope, (
f"Field '{field}' not in envelope: {list(envelope.keys())}"
)
assert envelope[field] == value, (
f"Expected envelope['{field}'] == {value}, got {envelope[field]!r}"
)
@then('the spec config get JSON envelope should have field "{field}"')
def step_spec_json_envelope_has_field(context: Context, field: str) -> None:
envelope = _get_spec_json(context)
assert field in envelope, (
f"Field '{field}' not in envelope: {list(envelope.keys())}"
)
# ---------------------------------------------------------------------------
# Then — JSON data
# ---------------------------------------------------------------------------
def _get_spec_json_data(context: Context) -> dict[str, Any]:
"""Return the data object from the JSON envelope."""
envelope = _get_spec_json(context)
return envelope["data"]
@then('the spec config get JSON data should have field "{field}"')
def step_spec_json_data_has_field(context: Context, field: str) -> None:
data = _get_spec_json_data(context)
assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
@then('the spec config get JSON data field "{field}" should equal "{value}"')
def step_spec_json_data_field_str(context: Context, field: str, value: str) -> None:
data = _get_spec_json_data(context)
assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
assert data[field] == value, (
f"Expected data['{field}'] == '{value}', got {data[field]!r}"
)
@then('the spec config get JSON data field "{field}" should be false')
def step_spec_json_data_field_false(context: Context, field: str) -> None:
data = _get_spec_json_data(context)
assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
assert data[field] is False, (
f"Expected data['{field}'] == False, got {data[field]!r}"
)
@then('the spec config get JSON data field "{field}" should be true')
def step_spec_json_data_field_true(context: Context, field: str) -> None:
data = _get_spec_json_data(context)
assert field in data, f"Field '{field}' not in JSON data: {list(data.keys())}"
assert data[field] is True, f"Expected data['{field}'] == True, got {data[field]!r}"
# ---------------------------------------------------------------------------
# Then — JSON origin
# ---------------------------------------------------------------------------
@then('the spec config get JSON origin should have field "{field}"')
def step_spec_json_origin_has_field(context: Context, field: str) -> None:
data = _get_spec_json_data(context)
origin = data.get("origin", {})
assert field in origin, f"Field '{field}' not in JSON origin: {list(origin.keys())}"
# ---------------------------------------------------------------------------
# Then — JSON resolution chain
# ---------------------------------------------------------------------------
@then(
"the spec config get JSON resolution chain should use human-readable source names"
)
def step_spec_json_chain_human_readable(context: Context) -> None:
data = _get_spec_json_data(context)
chain = data.get("resolution_chain", [])
assert len(chain) > 0, "Expected non-empty resolution chain"
for entry in chain:
src = entry.get("source", "")
assert src in _SPEC_SOURCE_NAMES, (
f"Source '{src}' is not a spec-required human-readable name. "
f"Expected one of: {_SPEC_SOURCE_NAMES}"
)
# ---------------------------------------------------------------------------
# Then — JSON winner
# ---------------------------------------------------------------------------
@then('the spec config get JSON winner should have field "{field}"')
def step_spec_json_winner_has_field(context: Context, field: str) -> None:
data = _get_spec_json_data(context)
winner = data.get("winner", {})
assert field in winner, f"Field '{field}' not in JSON winner: {list(winner.keys())}"
# ---------------------------------------------------------------------------
# Then — JSON timing
# ---------------------------------------------------------------------------
@then('the spec config get JSON timing should have field "{field}"')
def step_spec_json_timing_has_field(context: Context, field: str) -> None:
envelope = _get_spec_json(context)
timing = envelope.get("timing", {})
assert field in timing, f"Field '{field}' not in JSON timing: {list(timing.keys())}"
@@ -0,0 +1,171 @@
"""Display helpers for the ``agents config get`` command.
Provides spec-required formatting utilities for converting internal
``ConfigService`` data structures to the output format defined in
``docs/specification.md`` (section "agents config get").
Separated from ``config.py`` to keep that module under the 500-line limit.
"""
from __future__ import annotations
from typing import Any
from cleveragents.application.services.config_service import ConfigLevel
# ---------------------------------------------------------------------------
# Source name mappings
# ---------------------------------------------------------------------------
# Mapping from internal ConfigLevel values to spec-required human-readable
# names used in the resolution chain and winner objects.
_SOURCE_DISPLAY_NAMES: dict[str, str] = {
ConfigLevel.CLI_FLAG.value: "CLI flag",
ConfigLevel.ENV_VAR.value: "Env var",
ConfigLevel.LOCAL.value: "Local file",
ConfigLevel.PROJECT.value: "Project file",
ConfigLevel.GLOBAL.value: "Config file",
ConfigLevel.DEFAULT.value: "Default",
}
# Short source names used in the top-level ``data.source`` field and the
# Rich ``Source:`` line. These match the spec examples exactly.
_SOURCE_SHORT_NAMES: dict[str, str] = {
ConfigLevel.CLI_FLAG.value: "cli",
ConfigLevel.ENV_VAR.value: "env",
ConfigLevel.LOCAL.value: "local",
ConfigLevel.PROJECT.value: "project",
ConfigLevel.GLOBAL.value: "config",
ConfigLevel.DEFAULT.value: "default",
}
# Mapping from Python type names to spec-required type strings.
_PYTHON_TYPE_TO_SPEC: dict[str, str] = {
"str": "string",
"bool": "boolean",
"int": "integer",
"float": "number",
"NoneType": "null",
"None": "null",
}
# ---------------------------------------------------------------------------
# Public helpers
# ---------------------------------------------------------------------------
def spec_type_name(value: Any) -> str:
"""Return the spec-required type string for *value*.
Maps Python type names to the spec-required strings:
``str`` ``string``, ``bool`` ``boolean``, ``int`` ``integer``,
``float`` ``number``, ``None`` ``null``.
"""
if value is None:
return "null"
python_name = type(value).__name__
return _PYTHON_TYPE_TO_SPEC.get(python_name, python_name)
def spec_source_short(internal_source: str) -> str:
"""Return the short source name for the ``data.source`` field.
Used in the top-level ``data.source`` JSON field and the Rich
``Source:`` line. Matches the spec examples (e.g. ``"config"``,
``"default"``, ``"env"``).
"""
return _SOURCE_SHORT_NAMES.get(internal_source, internal_source)
def spec_source_display(internal_source: str) -> str:
"""Return the human-readable source name for resolution chain entries.
Used inside ``resolution_chain[].source`` and ``winner.source``.
Returns names like ``"CLI flag"``, ``"Config file"``, ``"Default"``.
"""
return _SOURCE_DISPLAY_NAMES.get(internal_source, internal_source)
def build_spec_resolution_chain(
chain: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Convert internal chain entries to spec-required format.
The spec requires:
``[{"level": 1, "source": "CLI flag", "value": null}, ...]``
Human-readable source names are used inside the chain per spec.
"""
result: list[dict[str, Any]] = []
for idx, entry in enumerate(chain, start=1):
result.append(
{
"level": idx,
"source": spec_source_display(entry.get("source", "")),
"value": entry.get("value"),
}
)
return result
def find_winner_in_chain(
chain: list[dict[str, Any]],
winning_source: str,
) -> dict[str, Any]:
"""Return the winner dict ``{"source": ..., "level": ...}`` for the chain.
*winning_source* is the internal ``ConfigLevel.value`` string.
The spec shows ``winner.source`` in lowercase (e.g. ``"config file"``).
"""
# Use lowercase human-readable name per spec example
display_name = spec_source_display(winning_source).lower()
for idx, entry in enumerate(chain, start=1):
if entry.get("source") == winning_source and entry.get("value") is not None:
return {"source": display_name, "level": idx}
# Fallback: last entry (default)
return {"source": display_name, "level": len(chain)}
def build_origin(
chain: list[dict[str, Any]],
winning_source: str,
default_value: Any,
) -> dict[str, Any]:
"""Build the ``origin`` object for the JSON envelope.
The origin shows where the winning value came from (file path, line,
and the registered default). Line numbers are not tracked by the
service, so ``line`` is always ``null``.
TODO: Track line numbers in ConfigService to populate ``line`` field.
"""
file_path: str | None = None
line: int | None = None # ConfigService does not track line numbers
# Find the winning chain entry to extract path info
for entry in chain:
if entry.get("source") == winning_source and entry.get("value") is not None:
raw_path = entry.get("path")
if raw_path:
file_path = str(raw_path)
break
# If no file path found, use a descriptive label based on source
if file_path is None:
if winning_source == ConfigLevel.ENV_VAR.value:
for entry in chain:
if entry.get("source") == winning_source:
env_name = entry.get("env_name")
if env_name:
file_path = f"${env_name}"
break
elif winning_source == ConfigLevel.CLI_FLAG.value:
file_path = "(CLI flag)"
elif winning_source == ConfigLevel.DEFAULT.value:
file_path = "(built-in default)"
return {
"file": file_path,
"line": line,
"default": default_value,
}
@@ -0,0 +1,85 @@
"""Internal helpers for the ``agents config`` command group.
Provides pure key-normalisation, validation, and secret-masking utilities
used by ``config.py``. Extracted to keep ``config.py`` under the
500-line limit. Functions here have no side effects and do not call
``ConfigService``.
"""
from __future__ import annotations
import re
from typing import Any
import typer
from cleveragents.application.services.config_service import _REGISTRY
# ---------------------------------------------------------------------------
# Secret helpers
# ---------------------------------------------------------------------------
_SECRET_PATTERNS: re.Pattern[str] = re.compile(
r"(api[_\-]?key|token|secret|password)", re.IGNORECASE
)
def _is_secret_key(key: str) -> bool:
"""Return ``True`` when *key* looks like it stores a secret."""
return _SECRET_PATTERNS.search(key) is not None
def _mask_value(value: str) -> str:
"""Replace a value with ``****``."""
return "****"
# ---------------------------------------------------------------------------
# Key helpers
# ---------------------------------------------------------------------------
def _normalize_key(key: str) -> str:
"""Pass-through normalisation — keys are matched as-is against the registry.
Backward compatibility: underscores are converted to dots so that
``log_level`` resolves to ``core.log.level`` if there is a single
match. If the key already exists in the registry, it is returned
unchanged.
"""
stripped = key.strip()
if stripped in _REGISTRY:
return stripped
# Try underscore-to-dot conversion
dotted = stripped.replace("_", ".")
if dotted in _REGISTRY:
return dotted
return stripped
def _validate_key(key: str) -> str:
"""Validate and normalise *key*, raising :class:`typer.BadParameter` on failure."""
normalized = _normalize_key(key)
if not normalized:
raise typer.BadParameter("Key must not be empty.")
if normalized not in _REGISTRY:
valid_sample = ", ".join(sorted(_REGISTRY)[:8])
raise typer.BadParameter(
f"Unknown configuration key: '{key}' "
f"(normalized: '{normalized}'). "
f"Valid keys include: {valid_sample} ..."
)
return normalized
def _env_var_for_key(key: str) -> str:
"""Return the environment-variable name for a registered key."""
entry = _REGISTRY.get(key)
if entry is not None:
return entry.env_var
return f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}"
def _settings_defaults() -> dict[str, Any]:
"""Return all registered config keys mapped to their default values."""
return {key: entry.default for key, entry in _REGISTRY.items()}
+119 -173
View File
@@ -1,22 +1,9 @@
"""Configuration management commands for CleverAgents CLI.
The ``agents config`` command group manages runtime configuration values
persisted in a TOML file at ``~/.cleveragents/config.toml``.
## Commands
| Command | Description |
|---------------------------------|-----------------------------------------|
| ``agents config set <K> <V>`` | Set a configuration value |
| ``agents config get <K>`` | Get a configuration value |
| ``agents config list [PATTERN]``| List configuration values |
## Key Format
Keys use the hierarchical dot-path names defined in the specification
(e.g. ``core.log.level``, ``plan.budget.per-plan``). Resolution is
delegated to :class:`ConfigService` which implements the five-level
precedence chain.
Manages runtime configuration values persisted in
``~/.cleveragents/config.toml``. Commands: ``set``, ``get``, ``list``.
Keys use hierarchical dot-path names (e.g. ``core.log.level``).
Resolution is delegated to :class:`ConfigService`.
Based on implementation_plan.md task A8.cli.
"""
@@ -25,6 +12,8 @@ from __future__ import annotations
import fnmatch
import re
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Annotated, Any
@@ -38,31 +27,30 @@ from cleveragents.application.services.config_service import (
ConfigScope,
ConfigService,
)
from cleveragents.cli.commands._config_display import (
build_origin,
build_spec_resolution_chain,
find_winner_in_chain,
spec_source_short,
spec_type_name,
)
from cleveragents.cli.commands._config_helpers import (
_is_secret_key,
_mask_value,
_settings_defaults,
_validate_key,
)
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.cli.renderers import _get_console
app = typer.Typer(help="Manage configuration settings for CleverAgents.")
console = _get_console()
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_CONFIG_DIR = Path.home() / ".cleveragents"
_CONFIG_PATH = _CONFIG_DIR / "config.toml"
_SECRET_PATTERNS: re.Pattern[str] = re.compile(
r"(api[_\-]?key|token|secret|password)", re.IGNORECASE
)
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _get_service() -> ConfigService:
"""Return a ``ConfigService`` wired to the standard config paths."""
container = get_container()
@@ -73,20 +61,6 @@ def _get_service() -> ConfigService:
)
def _is_secret_key(key: str) -> bool:
"""Return ``True`` when *key* looks like it stores a secret."""
return _SECRET_PATTERNS.search(key) is not None
def _mask_value(value: str) -> str:
"""Replace a value with ``****``."""
return "****"
# Legacy helpers retained for backward compatibility with tests that
# patch or call them directly. They now delegate to ConfigService.
def _settings_fields() -> dict[str, Any]:
"""Return all registered config keys mapped to their resolved values."""
svc = _get_service()
@@ -97,52 +71,6 @@ def _settings_fields() -> dict[str, Any]:
return results
def _settings_defaults() -> dict[str, Any]:
"""Return all registered config keys mapped to their default values."""
return {key: entry.default for key, entry in _REGISTRY.items()}
def _normalize_key(key: str) -> str:
"""Pass-through normalisation — keys are matched as-is against the registry.
Backward compatibility: underscores are converted to dots so that
``log_level`` resolves to ``core.log.level`` if there is a single
match. If the key already exists in the registry, it is returned
unchanged.
"""
stripped = key.strip()
if stripped in _REGISTRY:
return stripped
# Try underscore-to-dot conversion
dotted = stripped.replace("_", ".")
if dotted in _REGISTRY:
return dotted
return stripped
def _validate_key(key: str) -> str:
"""Validate and normalise *key*, raising :class:`typer.BadParameter` on failure."""
normalized = _normalize_key(key)
if not normalized:
raise typer.BadParameter("Key must not be empty.")
if normalized not in _REGISTRY:
valid_sample = ", ".join(sorted(_REGISTRY)[:8])
raise typer.BadParameter(
f"Unknown configuration key: '{key}' "
f"(normalized: '{normalized}'). "
f"Valid keys include: {valid_sample} ..."
)
return normalized
def _env_var_for_key(key: str) -> str:
"""Return the environment-variable name for a registered key."""
entry = _REGISTRY.get(key)
if entry is not None:
return entry.env_var
return f"CLEVERAGENTS_{key.upper().replace('.', '_').replace('-', '_')}"
def _resolve_source(key: str) -> str:
"""Determine which level supplied the winning value for *key*."""
svc = _get_service()
@@ -175,11 +103,6 @@ def _write_config_file(data: dict[str, Any]) -> None:
svc.write_config(data)
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
@app.command("set")
def config_set(
key: Annotated[str, typer.Argument(help="Configuration key (dot-path)")],
@@ -200,16 +123,12 @@ def config_set(
) -> None:
"""Set a configuration value.
Writes the value to the config file for the specified scope.
Scopes: ``global`` (~/.cleveragents/config.toml),
``project`` (<project-root>/config.toml),
``local`` (<project-root>/config.local.toml).
Scopes: ``global``, ``project``, ``local``.
Examples::
agents config set core.log.level DEBUG
agents config set plan.concurrency 8 --scope local
agents config set core.automation-profile manual --project local/prod
"""
normalized = _validate_key(key)
svc = _get_service()
@@ -217,8 +136,6 @@ def config_set(
# Coerce to the registered type
coerced: Any = svc.validate_type(normalized, value)
# Resolve the target scope
config_scope: ConfigScope | None = None
if scope is not None:
try:
@@ -227,12 +144,8 @@ def config_set(
raise typer.BadParameter(
f"Invalid scope: '{scope}'. Use 'global', 'project', or 'local'."
) from exc
# Read previous value
config_data = svc.read_config()
if project is not None:
# Legacy project-scoped set (--project flag)
if not entry.project_scopable:
raise typer.BadParameter(f"Key '{normalized}' is not project-scopable.")
project_section = config_data.get("project", {})
@@ -248,7 +161,6 @@ def config_set(
svc.write_config(config_data)
scope_display = f"project:{project}"
elif config_scope is not None:
# New three-scope set (--scope flag)
if config_scope == ConfigScope.GLOBAL:
previous = config_data.get(normalized)
elif config_scope == ConfigScope.PROJECT:
@@ -292,79 +204,128 @@ def config_set(
@app.command("get")
def config_get(
key: Annotated[str, typer.Argument(help="Configuration key (dot-path)")],
verbose: Annotated[
bool,
typer.Option("--verbose", "-v", help="Show full resolution chain"),
] = False,
project: Annotated[
str | None,
typer.Option("--project", "-p", help="Project name for scoped resolution"),
] = None,
verbose: Annotated[
bool,
typer.Option(
"--verbose",
"-v",
hidden=True,
help="[DEPRECATED] Verbose output (no longer needed — chain always shown)",
),
] = False,
fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
) -> None:
"""Get a configuration value with its resolution chain.
Shows the effective value and which source it was resolved from
(CLI flag, environment variable, project config, global file, or
default).
Use ``--verbose`` to display the full five-level resolution chain.
Shows the Config, Origin, and Resolution Chain panels per spec.
The full chain and Winner indicator are always displayed.
Examples::
agents config get core.log.level
agents config get plan.budget.per-plan --verbose
agents config get core.automation-profile --project local/prod
"""
started_at = datetime.now(tz=UTC)
t0 = time.monotonic()
normalized = _validate_key(key)
svc = _get_service()
resolved = svc.resolve(
normalized,
project_name=project,
verbose=verbose,
)
result: dict[str, Any] = {
"key": normalized,
"value": resolved.value,
"source": resolved.source.value,
"type": type(resolved.value).__name__ if resolved.value is not None else "None",
}
if verbose:
result["resolution_chain"] = resolved.chain
entry = _REGISTRY[normalized]
# Always verbose — spec requires the full chain in all output formats.
resolved = svc.resolve(normalized, project_name=project, verbose=True)
elapsed_ms = round((time.monotonic() - t0) * 1000)
overridden = resolved.value != entry.default
spec_chain = build_spec_resolution_chain(resolved.chain)
winner = find_winner_in_chain(resolved.chain, resolved.source.value)
origin = build_origin(resolved.chain, resolved.source.value, entry.default)
spec_type = spec_type_name(resolved.value)
short_source = spec_source_short(resolved.source.value)
display_value: Any = resolved.value
if hasattr(display_value, "__fspath__"):
display_value = str(display_value)
if fmt != OutputFormat.RICH.value:
# Serialise Path objects to strings for JSON/YAML
if hasattr(result["value"], "__fspath__"):
result["value"] = str(result["value"])
for entry in result.get("resolution_chain", []):
if hasattr(entry.get("value"), "__fspath__"):
entry["value"] = str(entry["value"])
typer.echo(format_output(result, fmt))
data: dict[str, Any] = {
"key": normalized,
"value": display_value,
"source": short_source,
"overridden": overridden,
"type": spec_type,
"origin": origin,
"resolution_chain": spec_chain,
"winner": winner,
}
# Serialise Path objects in chain values
for chain_entry in data["resolution_chain"]:
if hasattr(chain_entry.get("value"), "__fspath__"):
chain_entry["value"] = str(chain_entry["value"])
envelope: dict[str, Any] = {
"command": "config get",
"status": "ok",
"exit_code": 0,
"data": data,
"timing": {
"started": started_at.isoformat(),
"duration_ms": elapsed_ms,
},
"messages": ["Config read"],
}
typer.echo(format_output(envelope, fmt))
return
# Rich output: three panels per spec
overridden_display = "yes" if overridden else "no"
console.print(
Panel(
f"[bold]Key:[/bold] {normalized}\n"
f"[bold]Value:[/bold] {resolved.value}\n"
f"[bold]Source:[/bold] {resolved.source.value}\n"
"[bold]Type:[/bold] "
+ (type(resolved.value).__name__ if resolved.value is not None else "None"),
title="Configuration Value",
f"[bold]Value:[/bold] {display_value}\n"
f"[bold]Source:[/bold] {short_source}\n"
f"[bold]Overridden:[/bold] {overridden_display}\n"
f"[bold]Type:[/bold] {spec_type}",
title="Config",
expand=False,
)
)
origin_file = origin.get("file") or "(unknown)"
origin_line = origin.get("line")
origin_default = origin.get("default")
origin_line_display = str(origin_line) if origin_line is not None else "(unknown)"
origin_default_display = (
str(origin_default) if origin_default is not None else "(none)"
)
console.print(
Panel(
f"[bold]File:[/bold] {origin_file}\n"
f"[bold]Line:[/bold] {origin_line_display}\n"
f"[bold]Default:[/bold] {origin_default_display}",
title="Origin",
expand=False,
)
)
chain_lines: list[str] = []
for chain_entry in spec_chain:
val = chain_entry.get("value")
val_display = str(val) if val is not None else "(not set)"
chain_lines.append(
f"{chain_entry['level']}. {chain_entry['source']}: {val_display}"
)
winner_src = winner.get("source", "")
winner_level = winner.get("level", "")
chain_lines.append(f"[bold]Winner:[/bold] {winner_src} (level {winner_level})")
console.print(
Panel(
"\n".join(chain_lines),
title="Resolution Chain",
expand=False,
)
)
if verbose and resolved.chain:
console.print("\n[bold]Resolution chain[/bold] (highest → lowest priority):")
for chain_entry in resolved.chain:
src = chain_entry["source"]
val = chain_entry.get("value")
is_winner = src == resolved.source.value and val is not None
marker = " [green]◀ active[/green]" if is_winner else ""
display_val = str(val) if val is not None else "(not set)"
console.print(f" {src:14s}{display_val}{marker}")
console.print("[green]✓ OK[/green] Config read")
@app.command("list")
@@ -389,17 +350,13 @@ def config_list(
) -> None:
"""List all configuration values.
Optionally filter by key glob pattern (e.g. ``plan.*``, ``context.pipeline.*``)
and/or value regex. Secret values (API keys, tokens, passwords) are
masked by default; use ``--show-secrets`` to reveal them.
Filter by key glob (e.g. ``plan.*``) and/or value regex.
Secrets are masked by default; use ``--show-secrets`` to reveal.
Examples::
agents config list
agents config list "plan.*"
agents config list "context.pipeline.*"
agents config list --filter-values "DEBUG"
agents config list --show-secrets
agents config list --filter-values "DEBUG" --show-secrets
"""
# Compile regex patterns (fail fast on invalid regex)
key_re: re.Pattern[str] | None = None
@@ -420,8 +377,6 @@ def config_list(
raise typer.Exit(code=1) from exc
svc = _get_service()
# When --project is given, show only project-scoped overrides
if project is not None:
proj_overrides = svc.get_project_overrides(project)
settings_list: list[dict[str, Any]] = []
@@ -429,19 +384,14 @@ def config_list(
raw_value = proj_overrides[key]
str_value = str(raw_value) if raw_value is not None else ""
# Key filter
if (
key_re is not None
and not key_re.search(key)
and (pattern is None or not fnmatch.fnmatch(key, pattern))
):
continue
# Value filter
if val_re is not None and not val_re.search(str_value):
continue
# Mask secrets unless --show-secrets
display_value: Any = raw_value
if _is_secret_key(key) and not show_secrets and raw_value is not None:
display_value = _mask_value(str(raw_value))
@@ -509,12 +459,9 @@ def config_list(
if val_re is not None and not val_re.search(str_value):
continue
# Modified flag: value differs from default
default_val = defaults.get(key)
modified = raw_value != default_val
# Mask secrets unless --show-secrets
display_value: Any = raw_value
display_value = raw_value
if _is_secret_key(key) and not show_secrets and raw_value is not None:
display_value = _mask_value(str(raw_value))
@@ -532,7 +479,6 @@ def config_list(
return
if fmt != OutputFormat.RICH.value:
# Serialise Path objects for JSON/YAML
for entry in settings_list_all:
if hasattr(entry["value"], "__fspath__"):
entry["value"] = str(entry["value"])