Implemented the next missing parity slice around agent configurations.

This commit is contained in:
Abdelrahman Abdallah
2026-04-17 05:15:17 +02:00
parent 02674ba594
commit 872f4f89b5
11 changed files with 770 additions and 6 deletions
+14
View File
@@ -13,11 +13,18 @@ from .agent_context import (
from .agent_manager import AgentManager
from .agent_registry import (
AgentLoadError,
AgentMutationResult,
AgentRegistrySnapshot,
create_agent_definition,
delete_agent_definition,
find_agent_definition,
load_agent_registry,
normalize_mutable_source,
render_agent_detail,
render_agent_mutation,
render_agents_report,
scaffold_agent_definition,
update_agent_definition,
)
from .agent_runtime import LocalCodingAgent
from .agent_session import AgentMessage, AgentSessionState
@@ -51,6 +58,7 @@ __all__ = [
'AgentContextSnapshot',
'AgentManager',
'AgentLoadError',
'AgentMutationResult',
'AgentPermissions',
'AgentRegistrySnapshot',
'AgentRunResult',
@@ -119,8 +127,10 @@ __all__ = [
'clear_context_caches',
'clear_token_counter_cache',
'count_tokens',
'create_agent_definition',
'calculate_token_budget',
'default_tool_registry',
'delete_agent_definition',
'describe_token_counter',
'estimate_chat_overhead',
'execute_tool',
@@ -130,9 +140,13 @@ __all__ = [
'get_user_context',
'load_agent_registry',
'load_session',
'normalize_mutable_source',
'render_agent_detail',
'render_agent_mutation',
'render_agents_report',
'run_parity_audit',
'scaffold_agent_definition',
'save_session',
'set_system_prompt_injection',
'update_agent_definition',
]
+292
View File
@@ -16,6 +16,13 @@ _SOURCE_ORDER = {
'userSettings': 1,
'projectSettings': 2,
}
_MUTABLE_SOURCE_ALIASES = {
'project': 'projectSettings',
'projectSettings': 'projectSettings',
'user': 'userSettings',
'userSettings': 'userSettings',
}
_UNSET = object()
@dataclass(frozen=True)
@@ -33,6 +40,15 @@ class AgentRegistrySnapshot:
failed_files: tuple[AgentLoadError, ...]
@dataclass(frozen=True)
class AgentMutationResult:
action: str
agent_type: str
source: str
file_path: str
overwritten: bool = False
def load_agent_registry(cwd: Path) -> AgentRegistrySnapshot:
builtin_agents = tuple(get_builtin_agents())
loaded_agents: list[AgentDefinition] = list(builtin_agents)
@@ -258,6 +274,274 @@ def render_agent_detail(snapshot: AgentRegistrySnapshot, agent_type: str) -> str
return '\n'.join(lines)
def normalize_mutable_source(source: str | None, *, allow_auto: bool = False) -> str:
if source is None:
return 'auto' if allow_auto else 'projectSettings'
normalized = source.strip()
if allow_auto and normalized in {'', 'auto'}:
return 'auto'
resolved = _MUTABLE_SOURCE_ALIASES.get(normalized)
if resolved is None:
choices = ', '.join(sorted(_MUTABLE_SOURCE_ALIASES))
if allow_auto:
choices = 'auto, ' + choices
raise ValueError(f'Unsupported agent source: {source}. Expected one of: {choices}')
return resolved
def format_agent_markdown(
*,
agent_type: str,
description: str,
system_prompt: str,
tools: tuple[str, ...] | None = None,
model: str | None = None,
color: str | None = None,
permission_mode: str | None = None,
max_turns: int | None = None,
initial_prompt: str | None = None,
background: bool = False,
one_shot: bool = False,
omit_claude_md: bool = False,
) -> str:
lines = [
'---',
f'name: {agent_type}',
f'description: "{_escape_frontmatter_text(description)}"',
]
if tools is not None:
if tools:
lines.append(f'tools: {", ".join(tools)}')
else:
lines.append('tools: []')
if model:
lines.append(f'model: {model}')
if color:
lines.append(f'color: {color}')
if permission_mode:
lines.append(f'permissionMode: {permission_mode}')
if max_turns is not None:
lines.append(f'maxTurns: {max_turns}')
if initial_prompt:
lines.append(f'initialPrompt: "{_escape_frontmatter_text(initial_prompt)}"')
if background:
lines.append('background: true')
if one_shot:
lines.append('oneShot: true')
if omit_claude_md:
lines.append('omitClaudeMd: true')
lines.extend(['---', '', system_prompt.strip(), ''])
return '\n'.join(lines)
def create_agent_definition(
cwd: Path,
*,
agent_type: str,
description: str,
system_prompt: str,
source: str = 'projectSettings',
overwrite: bool = False,
tools: tuple[str, ...] | None = None,
model: str | None = None,
color: str | None = None,
permission_mode: str | None = None,
max_turns: int | None = None,
initial_prompt: str | None = None,
background: bool = False,
one_shot: bool = False,
omit_claude_md: bool = False,
) -> AgentMutationResult:
resolved_source = normalize_mutable_source(source)
file_path = get_agent_file_path(cwd, resolved_source, agent_type)
file_path.parent.mkdir(parents=True, exist_ok=True)
existed_before = file_path.exists()
if existed_before and not overwrite:
raise ValueError(f'Agent file already exists: {file_path}')
file_path.write_text(
format_agent_markdown(
agent_type=agent_type,
description=description,
system_prompt=system_prompt,
tools=tools,
model=model,
color=color,
permission_mode=permission_mode,
max_turns=max_turns,
initial_prompt=initial_prompt,
background=background,
one_shot=one_shot,
omit_claude_md=omit_claude_md,
),
encoding='utf-8',
)
return AgentMutationResult(
action='created',
agent_type=agent_type,
source=resolved_source,
file_path=str(file_path),
overwritten=existed_before and overwrite,
)
def update_agent_definition(
cwd: Path,
*,
agent_type: str,
source: str = 'auto',
description: str | object = _UNSET,
system_prompt: str | object = _UNSET,
tools: tuple[str, ...] | None | object = _UNSET,
model: str | None | object = _UNSET,
color: str | None | object = _UNSET,
permission_mode: str | None | object = _UNSET,
max_turns: int | None | object = _UNSET,
initial_prompt: str | None | object = _UNSET,
background: bool | object = _UNSET,
one_shot: bool | object = _UNSET,
omit_claude_md: bool | object = _UNSET,
) -> AgentMutationResult:
resolved_source = normalize_mutable_source(source, allow_auto=True)
snapshot = load_agent_registry(cwd)
target = find_mutable_agent(snapshot, agent_type, source=resolved_source)
if target is None:
raise ValueError(f'No editable agent definition found for: {agent_type}')
file_path = get_agent_file_path(
cwd,
target.source,
target.agent_type,
filename=target.filename,
)
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(
format_agent_markdown(
agent_type=target.agent_type,
description=target.when_to_use if description is _UNSET else str(description),
system_prompt=target.system_prompt if system_prompt is _UNSET else str(system_prompt),
tools=target.tools if tools is _UNSET else tools,
model=target.model if model is _UNSET else model,
color=target.color if color is _UNSET else color,
permission_mode=(
target.permission_mode if permission_mode is _UNSET else permission_mode
),
max_turns=target.max_turns if max_turns is _UNSET else max_turns,
initial_prompt=target.initial_prompt if initial_prompt is _UNSET else initial_prompt,
background=target.background if background is _UNSET else bool(background),
one_shot=target.one_shot if one_shot is _UNSET else bool(one_shot),
omit_claude_md=(
target.omit_claude_md if omit_claude_md is _UNSET else bool(omit_claude_md)
),
),
encoding='utf-8',
)
return AgentMutationResult(
action='updated',
agent_type=target.agent_type,
source=target.source,
file_path=str(file_path),
)
def delete_agent_definition(
cwd: Path,
*,
agent_type: str,
source: str = 'auto',
) -> AgentMutationResult:
resolved_source = normalize_mutable_source(source, allow_auto=True)
snapshot = load_agent_registry(cwd)
target = find_mutable_agent(snapshot, agent_type, source=resolved_source)
if target is None:
raise ValueError(f'No editable agent definition found for: {agent_type}')
file_path = get_agent_file_path(
cwd,
target.source,
target.agent_type,
filename=target.filename,
)
if not file_path.exists():
raise ValueError(f'Agent file does not exist: {file_path}')
file_path.unlink()
return AgentMutationResult(
action='deleted',
agent_type=target.agent_type,
source=target.source,
file_path=str(file_path),
)
def scaffold_agent_definition(
cwd: Path,
*,
agent_type: str,
source: str = 'projectSettings',
overwrite: bool = False,
description: str | None = None,
system_prompt: str | None = None,
) -> AgentMutationResult:
resolved_description = description or f'Use this agent when the task calls for {agent_type}.'
resolved_prompt = system_prompt or (
f'You are the {agent_type} agent.\n'
'Read the task carefully, use the available tools deliberately, and return a concise result.'
)
return create_agent_definition(
cwd,
agent_type=agent_type,
description=resolved_description,
system_prompt=resolved_prompt,
source=source,
overwrite=overwrite,
)
def find_mutable_agent(
snapshot: AgentRegistrySnapshot,
agent_type: str,
*,
source: str = 'auto',
) -> AgentDefinition | None:
if source == 'auto':
candidates = [
agent
for agent in snapshot.all_agents
if agent.agent_type == agent_type and agent.source in _MUTABLE_SOURCE_ALIASES.values()
]
if not candidates:
return None
return max(candidates, key=lambda agent: _source_rank(agent.source))
for agent in snapshot.all_agents:
if agent.agent_type == agent_type and agent.source == source:
return agent
return None
def get_agent_file_path(
cwd: Path,
source: str,
agent_type: str,
*,
filename: str | None = None,
) -> Path:
resolved_source = normalize_mutable_source(source)
directories = dict(iter_agent_directories(cwd))
directory = directories[resolved_source]
return directory / f'{filename or agent_type}.md'
def render_agent_mutation(result: AgentMutationResult) -> str:
return '\n'.join(
[
'# Agent',
'',
f'action={result.action}',
f'agent_type={result.agent_type}',
f'source={result.source}',
f'file_path={result.file_path}',
f'overwritten={result.overwritten}',
]
)
def _split_frontmatter(text: str) -> tuple[dict[str, Any], str]:
normalized = text.replace('\r\n', '\n')
match = _FRONTMATTER_RE.match(normalized)
@@ -303,6 +587,14 @@ def _parse_frontmatter_value(value: str) -> Any:
return value
def _escape_frontmatter_text(value: str) -> str:
return (
value.replace('\\', '\\\\')
.replace('"', '\\"')
.replace('\n', '\\\\n')
)
def _parse_tool_list(value: Any) -> tuple[str, ...] | None:
if value is None or value == '':
return None
+58
View File
@@ -15,10 +15,15 @@ from .agent_context_usage import collect_context_usage, estimate_tokens, format_
from .compact import compact_conversation
from .ask_user_runtime import AskUserRuntime
from .agent_registry import (
delete_agent_definition,
find_agent_definition,
normalize_mutable_source,
load_agent_registry,
render_agent_mutation,
render_agent_detail,
render_agents_report,
scaffold_agent_definition,
update_agent_definition,
)
from .config_runtime import ConfigRuntime
from .hook_policy import HookPolicyRuntime
@@ -3418,6 +3423,59 @@ class LocalCodingAgent:
snapshot = self.load_agent_registry()
return render_agent_detail(snapshot, agent_type)
def render_agent_create_report(
self,
agent_type: str,
*,
source: str = 'projectSettings',
description: str | None = None,
system_prompt: str | None = None,
overwrite: bool = False,
) -> str:
result = scaffold_agent_definition(
self.runtime_config.cwd,
agent_type=agent_type,
source=normalize_mutable_source(source),
overwrite=overwrite,
description=description,
system_prompt=system_prompt,
)
return render_agent_mutation(result)
def render_agent_update_report(
self,
agent_type: str,
*,
source: str = 'auto',
description: str | None = None,
system_prompt: str | None = None,
) -> str:
update_kwargs: dict[str, object] = {}
if description is not None:
update_kwargs['description'] = description
if system_prompt is not None:
update_kwargs['system_prompt'] = system_prompt
result = update_agent_definition(
self.runtime_config.cwd,
agent_type=agent_type,
source=normalize_mutable_source(source, allow_auto=True),
**update_kwargs,
)
return render_agent_mutation(result)
def render_agent_delete_report(
self,
agent_type: str,
*,
source: str = 'auto',
) -> str:
result = delete_agent_definition(
self.runtime_config.cwd,
agent_type=agent_type,
source=normalize_mutable_source(source, allow_auto=True),
)
return render_agent_mutation(result)
def render_memory_report(self) -> str:
prompt_context = self.build_prompt_context()
claude_md = prompt_context.user_context.get('claudeMd')
+103 -1
View File
@@ -809,14 +809,116 @@ def _handle_agents(agent: 'LocalCodingAgent', args: str, input_text: str) -> Sla
return _local_result(input_text, agent.render_agents_report())
if trimmed == 'all':
return _local_result(input_text, agent.render_agents_report(show_all=True))
if trimmed.startswith('create '):
try:
source, agent_type, description, system_prompt = _parse_agent_mutation_payload(
trimmed[7:].strip(),
default_source='projectSettings',
mode='create',
)
except ValueError as exc:
return _local_result(input_text, str(exc))
return _local_result(
input_text,
agent.render_agent_create_report(
agent_type,
source=source,
description=description,
system_prompt=system_prompt,
),
)
if trimmed.startswith('update '):
try:
source, agent_type, description, system_prompt = _parse_agent_mutation_payload(
trimmed[7:].strip(),
default_source='auto',
mode='update',
)
except ValueError as exc:
return _local_result(input_text, str(exc))
if description is None and system_prompt is None:
return _local_result(
input_text,
'Usage: /agents update [user|project] <agent-type> [description] [:: system prompt]',
)
return _local_result(
input_text,
agent.render_agent_update_report(
agent_type,
source=source,
description=description,
system_prompt=system_prompt,
),
)
if trimmed.startswith('delete '):
try:
source, agent_type = _parse_agent_target(trimmed[7:].strip(), default_source='auto')
except ValueError as exc:
return _local_result(input_text, str(exc))
return _local_result(
input_text,
agent.render_agent_delete_report(agent_type, source=source),
)
if trimmed.startswith('show '):
agent_type = trimmed[5:].strip()
if not agent_type:
return _local_result(input_text, 'Usage: /agents [all|active|show <agent-type>]')
return _local_result(input_text, _agents_usage())
return _local_result(input_text, agent.render_agent_detail_report(agent_type))
return _local_result(input_text, agent.render_agent_detail_report(trimmed))
def _agents_usage() -> str:
return (
'Usage: /agents [all|active|show <agent-type>|'
'create [user|project] <agent-type> [description] [:: system prompt]|'
'update [user|project] <agent-type> [description] [:: system prompt]|'
'delete [user|project] <agent-type>]'
)
def _parse_agent_target(payload: str, *, default_source: str) -> tuple[str, str]:
tokens = payload.split()
if not tokens:
raise ValueError(_agents_usage())
source = default_source
if tokens[0] in {'user', 'project', 'userSettings', 'projectSettings', 'auto'}:
source = tokens.pop(0)
if not tokens:
raise ValueError(_agents_usage())
return source, tokens[0]
def _parse_agent_mutation_payload(
payload: str,
*,
default_source: str,
mode: str,
) -> tuple[str, str, str | None, str | None]:
parts = [part.strip() for part in payload.split('::')]
source, agent_type = _parse_agent_target(parts[0], default_source=default_source)
head_tokens = parts[0].split()
if head_tokens and head_tokens[0] in {'user', 'project', 'userSettings', 'projectSettings', 'auto'}:
head_tokens = head_tokens[1:]
trailing_description = ' '.join(head_tokens[1:]).strip() if len(head_tokens) > 1 else ''
description = trailing_description or None
system_prompt = None
if mode == 'create':
if len(parts) >= 2 and parts[1]:
description = parts[1]
if len(parts) >= 3 and parts[2]:
system_prompt = parts[2]
else:
if len(parts) >= 2 and parts[1]:
if trailing_description:
system_prompt = parts[1]
else:
system_prompt = parts[1]
if len(parts) >= 3:
description = parts[1] or description
system_prompt = parts[2] or system_prompt
return source, agent_type, description, system_prompt
def _handle_memory(agent: 'LocalCodingAgent', _args: str, input_text: str) -> SlashCommandResult:
return _local_result(input_text, agent.render_memory_report())
+131
View File
@@ -11,6 +11,13 @@ from typing import Callable
from .background_runtime import BackgroundSessionRuntime, build_background_worker_command
from .account_runtime import AccountRuntime
from .ask_user_runtime import AskUserRuntime
from .agent_registry import (
create_agent_definition,
delete_agent_definition,
normalize_mutable_source,
render_agent_mutation,
update_agent_definition,
)
from .agent_runtime import LocalCodingAgent
from .agent_types import (
AgentPermissions,
@@ -163,6 +170,47 @@ def _load_output_schema_config(args: argparse.Namespace) -> OutputSchemaConfig |
)
def _add_agent_management_args(parser: argparse.ArgumentParser) -> None:
parser.add_argument('--cwd', default='.')
parser.add_argument(
'--source',
default='project',
choices=('project', 'user', 'auto', 'projectSettings', 'userSettings'),
)
parser.add_argument('--description')
parser.add_argument('--prompt')
parser.add_argument('--prompt-file')
parser.add_argument('--tools')
parser.add_argument('--model')
parser.add_argument('--color')
parser.add_argument('--permission-mode')
parser.add_argument('--max-turns', type=int)
parser.add_argument('--initial-prompt')
parser.add_argument('--background', action='store_true')
parser.add_argument('--one-shot', action='store_true')
parser.add_argument('--omit-claude-md', action='store_true')
parser.add_argument('--overwrite', action='store_true')
def _resolve_agent_prompt_text(args: argparse.Namespace) -> str | None:
prompt_text = getattr(args, 'prompt', None)
prompt_file = getattr(args, 'prompt_file', None)
if prompt_text and prompt_file:
raise ValueError('Specify only one of --prompt or --prompt-file')
if prompt_file:
return Path(prompt_file).read_text(encoding='utf-8')
return prompt_text
def _parse_tools_flag(raw_tools: str | None) -> tuple[str, ...] | None:
if raw_tools is None:
return None
cleaned = raw_tools.strip()
if not cleaned or cleaned == '*':
return None
return tuple(item.strip() for item in cleaned.split(',') if item.strip())
def _build_agent(args: argparse.Namespace) -> LocalCodingAgent:
return LocalCodingAgent(
model_config=_build_model_config(args),
@@ -879,6 +927,23 @@ def build_parser() -> argparse.ArgumentParser:
agents_parser.add_argument('agent_type', nargs='?')
agents_parser.add_argument('--all', action='store_true')
_add_agent_common_args(agents_parser, include_backend=False)
agents_create_parser = subparsers.add_parser('agents-create', help='create a local agent definition markdown file')
agents_create_parser.add_argument('agent_type')
_add_agent_management_args(agents_create_parser)
agents_update_parser = subparsers.add_parser('agents-update', help='update an existing local agent definition markdown file')
agents_update_parser.add_argument('agent_type')
_add_agent_management_args(agents_update_parser)
agents_delete_parser = subparsers.add_parser('agents-delete', help='delete an existing local agent definition markdown file')
agents_delete_parser.add_argument('agent_type')
agents_delete_parser.add_argument('--cwd', default='.')
agents_delete_parser.add_argument(
'--source',
default='auto',
choices=('project', 'user', 'auto', 'projectSettings', 'userSettings'),
)
return parser
@@ -1519,6 +1584,72 @@ def main(argv: list[str] | None = None) -> int:
else:
print(agent.render_agents_report(show_all=bool(args.all)))
return 0
if args.command == 'agents-create':
prompt_text = _resolve_agent_prompt_text(args)
result = create_agent_definition(
Path(args.cwd).resolve(),
agent_type=args.agent_type,
description=args.description or f'Use this agent when the task calls for {args.agent_type}.',
system_prompt=prompt_text
or (
f'You are the {args.agent_type} agent.\n'
'Read the task carefully, use the available tools deliberately, and return a concise result.'
),
source=normalize_mutable_source(args.source),
overwrite=bool(args.overwrite),
tools=_parse_tools_flag(args.tools),
model=args.model,
color=args.color,
permission_mode=args.permission_mode,
max_turns=args.max_turns,
initial_prompt=args.initial_prompt,
background=bool(args.background),
one_shot=bool(args.one_shot),
omit_claude_md=bool(args.omit_claude_md),
)
print(render_agent_mutation(result))
return 0
if args.command == 'agents-update':
prompt_text = _resolve_agent_prompt_text(args)
update_kwargs: dict[str, object] = {}
if args.description is not None:
update_kwargs['description'] = args.description
if prompt_text is not None:
update_kwargs['system_prompt'] = prompt_text
if args.tools is not None:
update_kwargs['tools'] = _parse_tools_flag(args.tools)
if args.model is not None:
update_kwargs['model'] = args.model
if args.color is not None:
update_kwargs['color'] = args.color
if args.permission_mode is not None:
update_kwargs['permission_mode'] = args.permission_mode
if args.max_turns is not None:
update_kwargs['max_turns'] = args.max_turns
if args.initial_prompt is not None:
update_kwargs['initial_prompt'] = args.initial_prompt
if args.background:
update_kwargs['background'] = True
if args.one_shot:
update_kwargs['one_shot'] = True
if args.omit_claude_md:
update_kwargs['omit_claude_md'] = True
result = update_agent_definition(
Path(args.cwd).resolve(),
agent_type=args.agent_type,
source=normalize_mutable_source(args.source, allow_auto=True),
**update_kwargs,
)
print(render_agent_mutation(result))
return 0
if args.command == 'agents-delete':
result = delete_agent_definition(
Path(args.cwd).resolve(),
agent_type=args.agent_type,
source=normalize_mutable_source(args.source, allow_auto=True),
)
print(render_agent_mutation(result))
return 0
parser.error(f'unknown command: {args.command}')
return 2