idle-draft: per-work-type and per-provider model/effort settings

--effort/--model on child argv; provider entries accept object form
{provider, model, effort}; precedence entry > work-type > profile.
Effort levels validated against the CLI's enumerated set.

Claude-Session: https://claude.ai/code/session_01YQDoWNM7XPPii28khFWoMc
This commit is contained in:
Paul O'Reilly
2026-08-02 22:16:34 +12:00
parent ec25f02f01
commit ca8b538d34
4 changed files with 757 additions and 15 deletions

View File

@@ -62,6 +62,17 @@ ALLOWED_ITEM_KEYS = {
}
ALLOWED_ATTEMPT_WORK_TYPES = {"research", "draft", "review"}
# Effort levels accepted by `claude --effort <level>`. Derived empirically on
# 2026-08-02 by running `claude --effort obviously-bogus-level -p ""` (no API
# call -- an invalid flag value is rejected before any network activity) and
# reading the CLI's own error message, which enumerated the valid set
# verbatim: "Warning: Unknown --effort value 'obviously-bogus-level' —
# ignoring it and using the default effort. Valid values: low, medium, high,
# xhigh, max." Cross-checked against `claude --help`'s `--effort <level>`
# description, which lists the same five values. If a future CLI version
# changes this set, re-run the same probe and update this constant.
CLAUDE_EFFORT_LEVELS = frozenset({"low", "medium", "high", "xhigh", "max"})
DEFAULT_MAX_TURNS = 25
DEFAULT_TASK_TIMEOUT_SECONDS = 1800
DEFAULT_MAX_ATTEMPTS = 2
@@ -224,13 +235,77 @@ def load_config(path: Path) -> dict:
for key in ("profile", "threshold_pct", "five_hour_ceiling", "min_idle"):
if key not in pcfg:
raise ConfigError(f"provider '{pname}' missing key '{key}'")
known_providers = set(data["providers"].keys())
for wtype, wcfg in data["work_types"].items():
if "providers" not in wcfg or not isinstance(wcfg["providers"], list):
raise ConfigError(f"work_type '{wtype}' missing 'providers' list")
for entry in wcfg["providers"]:
# A providers-list entry is either the legacy bare provider-name
# string, or an object carrying a per-provider model/effort
# override: {"provider": "<name>", "model": "...", "effort": "..."}.
# Both forms may appear in the same list (list order still encodes
# preference for select_provider()).
if isinstance(entry, str):
pname_ref = entry
if pname_ref not in known_providers:
raise ConfigError(
f"work_type '{wtype}' providers entry names unknown provider '{pname_ref}'"
)
elif isinstance(entry, dict):
unknown_keys = set(entry.keys()) - {"provider", "model", "effort"}
if unknown_keys:
raise ConfigError(
f"work_type '{wtype}' providers entry has unknown key(s): {sorted(unknown_keys)}"
)
pname_ref = entry.get("provider")
if not isinstance(pname_ref, str) or not pname_ref:
raise ConfigError(
f"work_type '{wtype}' providers entry (object form) missing non-empty 'provider' key"
)
if pname_ref not in known_providers:
raise ConfigError(
f"work_type '{wtype}' providers entry names unknown provider '{pname_ref}'"
)
if "model" in entry:
entry_model = entry["model"]
if not isinstance(entry_model, str) or not entry_model:
raise ConfigError(
f"work_type '{wtype}' providers entry for '{pname_ref}' 'model' must be "
f"a non-empty string"
)
if "effort" in entry:
entry_effort = entry["effort"]
if not isinstance(entry_effort, str) or not entry_effort:
raise ConfigError(
f"work_type '{wtype}' providers entry for '{pname_ref}' 'effort' must be "
f"a non-empty string"
)
if entry_effort not in CLAUDE_EFFORT_LEVELS:
raise ConfigError(
f"work_type '{wtype}' providers entry for '{pname_ref}' 'effort' must be one "
f"of {sorted(CLAUDE_EFFORT_LEVELS)}, got {entry_effort!r}"
)
else:
raise ConfigError(
f"work_type '{wtype}' providers entry must be a string or object, "
f"got {type(entry).__name__}"
)
if "allowed_tools" in wcfg:
at = wcfg["allowed_tools"]
if not isinstance(at, list) or not all(isinstance(x, str) for x in at):
raise ConfigError(f"work_type '{wtype}' 'allowed_tools' must be a list of strings")
if "effort" in wcfg:
effort = wcfg["effort"]
if not isinstance(effort, str) or not effort:
raise ConfigError(f"work_type '{wtype}' 'effort' must be a non-empty string")
if effort not in CLAUDE_EFFORT_LEVELS:
raise ConfigError(
f"work_type '{wtype}' 'effort' must be one of {sorted(CLAUDE_EFFORT_LEVELS)}, got {effort!r}"
)
if "model" in wcfg:
model = wcfg["model"]
if not isinstance(model, str) or not model:
raise ConfigError(f"work_type '{wtype}' 'model' must be a non-empty string")
return data
@@ -656,9 +731,19 @@ def build_ready_queue(repo: Path, config: dict, state: dict, in_flight: set) ->
return candidates
def provider_entry_name(entry) -> str:
"""A providers-list entry is either a plain provider-name string (the
legacy form) or an object {"provider": name, "model": ..., "effort": ...}.
This extracts just the name -- the only field select_provider() itself
needs; the model/effort overrides are resolved later, once a provider has
actually been selected, by resolve_work_type_model/_effort below."""
return entry if isinstance(entry, str) else entry["provider"]
def select_provider(work_type: str, config: dict, gates: dict, credential_ok: dict) -> tuple[str | None, str]:
providers = config["work_types"].get(work_type, {}).get("providers", [])
for pname in providers:
for entry in providers:
pname = provider_entry_name(entry)
gate = gates.get(pname, {})
if gate.get("eligible") and credential_ok.get(pname, (False, ""))[0]:
return pname, "eligible"
@@ -758,11 +843,20 @@ def validate_output(work_type: str, text: str) -> tuple[bool, str]:
def build_claude_argv(
model_id: str | None, max_turns: int, add_dirs: list[str], allowed_tools: list[str] | None = None
model_id: str | None,
max_turns: int,
add_dirs: list[str],
allowed_tools: list[str] | None = None,
effort: str | None = None,
) -> list[str]:
"""`model_id` is the already-resolved effective model (see
resolve_effective_model: work-type "model" beats the profile's MODEL_ID) --
this function emits at most one --model flag, never two."""
argv = ["claude", "-p", "--max-turns", str(max_turns)]
if model_id:
argv += ["--model", model_id]
if effort:
argv += ["--effort", effort]
for d in add_dirs:
argv += ["--add-dir", d]
if allowed_tools:
@@ -821,16 +915,70 @@ def resolve_allowed_tools(config: dict, work_type: str) -> list[str]:
return list(config["work_types"].get(work_type, {}).get("allowed_tools") or [])
def find_provider_entry(config: dict, work_type: str, provider_name: str):
"""The raw providers-list entry (str or dict) for provider_name within
work_type's providers list, or None if not present. Defensive lookup --
callers only ever pass a provider name that select_provider() just chose
from this same list, so a miss shouldn't happen, but resolve_work_type_*
below treat it as "no per-provider override" rather than raising."""
for entry in config["work_types"].get(work_type, {}).get("providers", []):
if provider_entry_name(entry) == provider_name:
return entry
return None
def resolve_work_type_effort(config: dict, work_type: str, provider_name: str | None = None) -> str | None:
"""Effort level for (work_type, provider_name). Precedence:
per-provider entry "effort" > work-type-level "effort" > None (CLI
default). `provider_name` is optional -- omit it to get just the
work-type-level value (e.g. before a provider has been selected). Both
layers are already validated (non-empty string, member of
CLAUDE_EFFORT_LEVELS) at config-load time -- this is a plain lookup."""
wcfg = config["work_types"].get(work_type, {})
if provider_name is not None:
entry = find_provider_entry(config, work_type, provider_name)
per_provider = entry.get("effort") if isinstance(entry, dict) else None
if per_provider:
return per_provider
return wcfg.get("effort")
def resolve_work_type_model(config: dict, work_type: str, provider_name: str | None = None) -> str | None:
"""Model override for (work_type, provider_name). Precedence:
per-provider entry "model" > work-type-level "model" > None (falls back
to the profile's MODEL_ID via resolve_effective_model). `provider_name`
is optional -- omit it to get just the work-type-level value. Both
layers are already type-checked (non-empty string) at config-load time --
this is a plain lookup."""
wcfg = config["work_types"].get(work_type, {})
if provider_name is not None:
entry = find_provider_entry(config, work_type, provider_name)
per_provider = entry.get("model") if isinstance(entry, dict) else None
if per_provider:
return per_provider
return wcfg.get("model")
def resolve_effective_model(profile_model_id: str | None, work_type_model: str | None) -> str | None:
"""Precedence: a work-type "model" override always wins over the
profile's provider.env MODEL_ID. Pure function so the precedence rule is
independently testable without touching a profile dir."""
return work_type_model or profile_model_id
def dispatch_preview(repo: Path, config: dict, candidate: dict, max_turns: int) -> dict:
"""Compute everything a dryrun needs to print, without executing anything."""
dossier = candidate["dossier"]
work_type = candidate["work_type"]
provider = candidate["provider"]
profile_dir = expand(config["providers"][provider]["profile"])
model_id = resolve_model_id(profile_dir)
model_id = resolve_effective_model(
resolve_model_id(profile_dir), resolve_work_type_model(config, work_type, provider)
)
effort = resolve_work_type_effort(config, work_type, provider)
add_dirs = resolve_add_dirs(config, work_type)
allowed_tools = resolve_allowed_tools(config, work_type)
argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools)
argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools, effort)
canonical = canonical_output_path(repo, candidate)
tmp_path = tmp_output_path(canonical)
parsed_env = parse_provider_env(profile_dir / "provider.env")
@@ -993,10 +1141,13 @@ def run_task(repo: Path, config: dict, candidate: dict, max_turns: int, task_tim
prompt_text = render_prompt(work_type, mapping)
profile_dir = expand(config["providers"][provider]["profile"])
model_id = resolve_model_id(profile_dir)
model_id = resolve_effective_model(
resolve_model_id(profile_dir), resolve_work_type_model(config, work_type, provider)
)
effort = resolve_work_type_effort(config, work_type, provider)
add_dirs = resolve_add_dirs(config, work_type)
allowed_tools = resolve_allowed_tools(config, work_type)
argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools)
argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools, effort)
env = build_child_env(profile_dir)
timed_out = False