346 lines
12 KiB
Python
346 lines
12 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Settings and configuration resolution for direct Aider worker behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field, replace
|
|
import os
|
|
import shlex
|
|
from pathlib import Path
|
|
from typing import Mapping, Sequence
|
|
|
|
from band_aider.env_aliases import (
|
|
OPENAI_PROVIDER_ALIAS_FAMILIES,
|
|
discovery_roots,
|
|
flatten_api_base_aliases,
|
|
flatten_api_key_aliases,
|
|
flatten_model_aliases,
|
|
nearest_env_file,
|
|
read_env_alias,
|
|
resolve_provider_alias_family,
|
|
)
|
|
from band_aider.judgement.repo_exec import RepoExec, current_repo_exec
|
|
|
|
AIDER_CMD_ALIASES = ("BAND_AIDER_AIDER_CMD",)
|
|
WRAPPER_MODE_ALIASES = ("BAND_AIDER_WRAPPER_MODE",)
|
|
MODEL_ALIASES = flatten_model_aliases(OPENAI_PROVIDER_ALIAS_FAMILIES)
|
|
API_BASE_ALIASES = flatten_api_base_aliases(OPENAI_PROVIDER_ALIAS_FAMILIES)
|
|
API_KEY_ALIASES = flatten_api_key_aliases(OPENAI_PROVIDER_ALIAS_FAMILIES)
|
|
EDIT_FORMAT_ALIASES = ("BAND_AIDER_EDIT_FORMAT",)
|
|
EXTRA_ARGS_ALIASES = ("BAND_AIDER_EXTRA_ARGS",)
|
|
# TODO: Deliberate with a human whether to remove this arbitrary escalation-config cap.
|
|
MAX_AIDER_ESCALATION_CONFIGS = 10
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AiderConfig:
|
|
"""Configuration for direct Aider or an existing wrapper script."""
|
|
|
|
aider_cmd: str
|
|
wrapper_mode: bool
|
|
model: str
|
|
api_base: str | None
|
|
api_key: str | None
|
|
edit_format: str = "diff"
|
|
yes: bool = False
|
|
extra_args: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AiderWorkerSettings:
|
|
"""AiderWorkerSettings stores filesystem and Aider config settings for AiderWorker."""
|
|
|
|
repo_root: Path
|
|
repo_exec: RepoExec = field(default_factory=current_repo_exec, repr=False, compare=False)
|
|
configs: tuple[AiderConfig, ...] = ()
|
|
# NOTE: AI Do NOT change this. We ignore all hidden files, they're none of our business.
|
|
ignored_artifact_globs: tuple[str, ...] = field(
|
|
default=(
|
|
".*",
|
|
".**/*",
|
|
)
|
|
)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Resolve repo-root-relative settings and merge config variants into one chain."""
|
|
|
|
object.__setattr__(self, "repo_root", Path(str(self.repo_root)))
|
|
object.__setattr__(
|
|
self,
|
|
"ignored_artifact_globs",
|
|
self._dedupe(
|
|
(
|
|
*self.ignored_artifact_globs,
|
|
*self._gitignore_globs(),
|
|
)
|
|
),
|
|
)
|
|
self._resolved_configs()
|
|
|
|
@property
|
|
def resolved_configs(self) -> tuple[AiderConfig, ...]:
|
|
"""Return the normalized config chain."""
|
|
|
|
return self.configs
|
|
|
|
def with_configs(
|
|
self,
|
|
*,
|
|
configs: Sequence[AiderConfig] | None = None,
|
|
) -> "AiderWorkerSettings":
|
|
"""Return one settings instance with additional config values merged in order."""
|
|
|
|
merged = self._merge_configs(
|
|
self.configs,
|
|
tuple(configs or ()),
|
|
)
|
|
return replace(self, configs=merged)
|
|
|
|
def _resolved_configs(self) -> tuple[AiderConfig, ...]:
|
|
"""Normalize the constructor config fields into one config tuple."""
|
|
|
|
merged = self._merge_configs(self.configs)
|
|
object.__setattr__(self, "configs", merged)
|
|
return merged
|
|
|
|
@staticmethod
|
|
def _merge_configs(*groups: object) -> tuple[AiderConfig, ...]:
|
|
"""Return the merged config chain from singular and plural config inputs."""
|
|
|
|
merged: list[AiderConfig] = []
|
|
for group in groups:
|
|
if group is None:
|
|
continue
|
|
if isinstance(group, AiderConfig):
|
|
merged.append(group)
|
|
continue
|
|
for item in group:
|
|
if item is None:
|
|
continue
|
|
merged.append(item)
|
|
return tuple(merged)
|
|
|
|
def _gitignore_globs(self) -> tuple[str, ...]:
|
|
"""Return artifact glob patterns loaded from the repo .gitignore."""
|
|
|
|
gitignore = self.repo_root / ".gitignore"
|
|
if not self.repo_exec.exists(gitignore):
|
|
return ()
|
|
|
|
patterns: list[str] = []
|
|
for line in self.repo_exec.read_text(gitignore, encoding="utf-8").splitlines():
|
|
pattern = self._normalize_gitignore_line(line)
|
|
if pattern:
|
|
patterns.extend(self._expand_gitignore_pattern(pattern))
|
|
|
|
return tuple(patterns)
|
|
|
|
@staticmethod
|
|
def _normalize_gitignore_line(line: str) -> str:
|
|
"""Return one usable .gitignore pattern or an empty string."""
|
|
|
|
pattern = line.strip()
|
|
if not pattern:
|
|
return ""
|
|
if pattern.startswith("#"):
|
|
return ""
|
|
if pattern.startswith("!"):
|
|
return ""
|
|
return pattern
|
|
|
|
@staticmethod
|
|
def _expand_gitignore_pattern(pattern: str) -> tuple[str, ...]:
|
|
"""Return glob variants for one .gitignore pattern."""
|
|
|
|
anchored = pattern.startswith("/")
|
|
directory_only = pattern.endswith("/")
|
|
clean = pattern.strip("/")
|
|
|
|
if not clean:
|
|
return ()
|
|
|
|
if directory_only:
|
|
if anchored:
|
|
return (f"{clean}/**",)
|
|
return (
|
|
f"{clean}/**",
|
|
f"**/{clean}/**",
|
|
)
|
|
|
|
if "/" in clean or anchored:
|
|
return (
|
|
clean,
|
|
f"{clean}/**",
|
|
)
|
|
|
|
return (
|
|
clean,
|
|
f"**/{clean}",
|
|
f"{clean}/**",
|
|
f"**/{clean}/**",
|
|
)
|
|
|
|
@staticmethod
|
|
def _dedupe(patterns: tuple[str, ...]) -> tuple[str, ...]:
|
|
"""Return patterns in first-seen order without duplicates."""
|
|
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for pattern in patterns:
|
|
if pattern in seen:
|
|
continue
|
|
seen.add(pattern)
|
|
result.append(pattern)
|
|
return tuple(result)
|
|
|
|
class AiderConfigFactory:
|
|
"""Factory methods for constructing AiderConfig instances from args and env.
|
|
|
|
Deprecated for OpenAI model-client configuration use.
|
|
"""
|
|
|
|
@classmethod
|
|
def _discovery_roots(cls, start: Path, home: Path) -> tuple[Path, ...]:
|
|
"""Return the bounded ancestor search path from start up to home, inclusive."""
|
|
|
|
return discovery_roots(start, home)
|
|
|
|
@classmethod
|
|
def _nearest_numeric_env(cls, index: int, roots: Sequence[Path]) -> Path | None:
|
|
"""Return the nearest ancestor numeric env file for one escalation tier."""
|
|
|
|
return nearest_env_file(f"band-aider.{index}.env", roots)
|
|
|
|
@classmethod
|
|
def _nearest_default_env(cls, roots: Sequence[Path]) -> Path | None:
|
|
"""Return the nearest ancestor default env file."""
|
|
|
|
return nearest_env_file("band-aider.env", roots)
|
|
|
|
@classmethod
|
|
def read_env_alias(cls, env: Mapping[str, str], aliases: Sequence[str]) -> str | None:
|
|
"""Return the first nonblank value found across an env alias group."""
|
|
|
|
return read_env_alias(env, aliases)
|
|
|
|
@classmethod
|
|
def read_env_alias_string(cls, aliases: Sequence[str]) -> str:
|
|
"""Return a slash-joined display string for one env alias group."""
|
|
|
|
return "/".join(aliases)
|
|
|
|
@classmethod
|
|
def read_env_file(cls, path: Path) -> dict[str, str]:
|
|
"""Parse a simple dotenv file without mutating process environment."""
|
|
|
|
if not path.exists():
|
|
return {}
|
|
|
|
values: dict[str, str] = {}
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
|
continue
|
|
key, value = stripped.split("=", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
value = value[1:-1]
|
|
values[key] = value
|
|
return values
|
|
|
|
@classmethod
|
|
def from_env(
|
|
cls,
|
|
args: object,
|
|
env: Mapping[str, str],
|
|
*,
|
|
priority_env: Mapping[str, str] | None = None,
|
|
) -> AiderConfig:
|
|
"""Resolve one Aider configuration from CLI args and an injected environment mapping."""
|
|
|
|
wrapper = Path.home() / "workspace/aider-iona/scripts/aider-iona"
|
|
aider_cmd = cls._first(args, "aider_cmd", env, AIDER_CMD_ALIASES) or "aider"
|
|
wrapper_mode = cls._bool_arg(args, "aider_wrapper_mode")
|
|
if wrapper_mode is None:
|
|
env_wrapper = cls.read_env_alias(env, WRAPPER_MODE_ALIASES)
|
|
wrapper_mode = _as_bool(env_wrapper) if env_wrapper is not None else aider_cmd == str(wrapper)
|
|
resolved = resolve_provider_alias_family(env, OPENAI_PROVIDER_ALIAS_FAMILIES, priority_env=priority_env)
|
|
explicit_model = getattr(args, "aider_model", None) or None
|
|
model = (
|
|
explicit_model
|
|
or (resolved.model if resolved is not None else None)
|
|
or "default"
|
|
)
|
|
return AiderConfig(
|
|
aider_cmd=aider_cmd,
|
|
wrapper_mode=bool(wrapper_mode),
|
|
model=model,
|
|
api_base=(
|
|
getattr(args, "aider_api_base", None)
|
|
or (resolved.api_base if resolved is not None else None)
|
|
or "https://iona.local.arpeggio.one/api"
|
|
),
|
|
api_key=getattr(args, "aider_api_key", None) or (resolved.api_key if resolved is not None else None),
|
|
edit_format=cls._first(args, "aider_edit_format", env, EDIT_FORMAT_ALIASES) or "diff",
|
|
extra_args=cls._extra_args(args, env),
|
|
)
|
|
|
|
@classmethod
|
|
def default(cls, args: object) -> AiderConfig:
|
|
"""Resolve one Aider configuration from CLI args and process environment."""
|
|
|
|
roots = cls._discovery_roots(Path.cwd(), Path.home())
|
|
env_path = cls._nearest_default_env(roots)
|
|
file_env = cls.read_env_file(env_path) if env_path is not None else {}
|
|
merged_env = dict(file_env)
|
|
merged_env.update(os.environ)
|
|
return cls.from_env(args, merged_env, priority_env=os.environ)
|
|
|
|
@classmethod
|
|
def sequence(cls, args: object, discovery_root: Path | None = None) -> tuple[AiderConfig, ...]:
|
|
"""Return the default Aider config plus any numeric escalation configs from a discovery root."""
|
|
|
|
configs = [cls.default(args)]
|
|
root = (discovery_root or Path.cwd()).resolve()
|
|
roots = cls._discovery_roots(root, Path.home())
|
|
default_env_path = cls._nearest_default_env(roots)
|
|
base_file_env = cls.read_env_file(default_env_path) if default_env_path is not None else {}
|
|
for index in range(1, MAX_AIDER_ESCALATION_CONFIGS + 1):
|
|
env_path = cls._nearest_numeric_env(index, roots)
|
|
if env_path is None:
|
|
continue
|
|
env_overlay = cls.read_env_file(env_path)
|
|
merged_env = dict(base_file_env)
|
|
merged_env.update(env_overlay)
|
|
configs.append(cls.from_env(args, merged_env, priority_env=env_overlay))
|
|
return tuple(configs)
|
|
|
|
@classmethod
|
|
def _first(cls, args: object, attr: str, env: Mapping[str, str], aliases: Sequence[str]) -> str | None:
|
|
value = getattr(args, attr, None)
|
|
if value:
|
|
return value
|
|
return cls.read_env_alias(env, aliases)
|
|
|
|
@classmethod
|
|
def _extra_args(cls, args: object, env: Mapping[str, str]) -> list[str]:
|
|
values = list(getattr(args, "aider_extra_arg", []) or [])
|
|
env_values = shlex.split(cls.read_env_alias(env, EXTRA_ARGS_ALIASES) or "")
|
|
return env_values + values
|
|
|
|
@classmethod
|
|
def _bool_arg(cls, args: object, attr: str) -> bool | None:
|
|
value = getattr(args, attr, None)
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return value
|
|
return _as_bool(str(value))
|
|
|
|
|
|
def _as_bool(value: str | None) -> bool:
|
|
return str(value).lower() in {"1", "true", "yes", "on"}
|