Coverage for src/keel/agents.py: 100%

1"""Agent dispatch + attribution — the pure resolution logic. 

2 

3The backbone dispatches agentic steps (implement / review / extensions) to a 

4configured agent: the **host agent** by default, a per-run **delegate** override, 

5or a per-role agent from ``knobs.implementer_agents``. Attribution records the 

6*effective* implementer as labels (``agent:<vendor>`` + a versionless 

7``model:<base>``), reusing the ship #2036 stripping algorithm. 

8 

9All functions here are pure and deterministic — no subprocess, no network. 

10""" 

11 

12from __future__ import annotations 

13 

14from .config import ProjectConfig 

15 

16#: Default host agent when nothing else is resolved. 

17HOST_DEFAULT = "claude" 

18 

19 

20def split_delegate(value: str) -> tuple[str, str | None]: 

21 """Split ``ollama:qwen2.5`` -> ``("ollama", "qwen2.5")``; ``codex`` -> ``("codex", None)``.""" 

22 vendor, sep, model = value.partition(":") 

23 return vendor, (model if (sep and model) else None) 

24 

25 

26def resolve_agent( 

27 config: ProjectConfig, 

28 *, 

29 role: str | None = None, 

30 delegate: str | None = None, 

31 host_agent: str = HOST_DEFAULT, 

32) -> str: 

33 """Resolve which agent runs a step. 

34 

35 Precedence: explicit ``delegate`` > per-role ``implementer_agents`` mapping > 

36 ``host_agent`` default. 

37 """ 

38 if delegate: 

39 return delegate 

40 if role and role in config.knobs.implementer_agents: 

41 return config.knobs.implementer_agents[role] 

42 return host_agent 

43 

44 

45def model_base(model: str) -> str: 

46 """Strip a model id to a coarse, versionless base label (ship #2036 algorithm). 

47 

48 Examples: ``qwen2.5:7b`` -> ``qwen``, ``gemma2`` -> ``gemma``, 

49 ``llama3.1`` -> ``llama``, ``gpt-5.5`` -> ``gpt-5``, ``gpt-4o`` -> ``gpt-4o``. 

50 """ 

51 m = model.strip().lower() 

52 if not m: 

53 return "" 

54 m = m.split(":", 1)[0] # (1) drop any ollama :tag 

55 if "-" in m: 

56 # (3) hyphenated family: keep <word>-<major>, drop the .minor 

57 head, _, tail = m.partition("-") 

58 major = tail.split(".", 1)[0] 

59 return f"{head}-{major}" 

60 # (2) non-hyphenated family: drop the trailing numeric run (digits + dots) 

61 i = len(m) 

62 while i > 0 and (m[i - 1].isdigit() or m[i - 1] == "."): 

63 i -= 1 

64 return m[:i] 

65 

66 

67def agent_label(vendor: str) -> str: 

68 """The persistent ``agent:<vendor>`` label.""" 

69 return f"agent:{vendor}" 

70 

71 

72def model_label(model: str) -> str | None: 

73 """The versionless ``model:<base>`` label, or ``None`` when no base is known.""" 

74 base = model_base(model) 

75 return f"model:{base}" if base else None 

76 

77 

78def attribution(vendor: str, model: str | None = None) -> dict[str, str | None]: 

79 """Resolve the effective attribution for an implementer/reviewer. 

80 

81 Returns ``{"agent_label", "model_label", "system"}`` where ``system`` is the 

82 full ``vendor`` or ``vendor:model`` string for the closure comment. 

83 """ 

84 system = f"{vendor}:{model}" if model else vendor 

85 return { 

86 "agent_label": agent_label(vendor), 

87 "model_label": model_label(model) if model else None, 

88 "system": system, 

89 }