mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
feat(grok-cli): render config.toml + map per-role grok CLI flags
First piece of the Grok CLI provider that replaces the opencode runtime: a pure, unit-tested module the agent entrypoint runs to translate the mounted mcp-config.json into ~/.grok/config.toml ([mcp_servers]) and compute the per-role 'grok -p' flags — subagent/shell/edit tool removal, raw-git-mutation and rm-rf denies, reasoning effort — mirroring ClaudeCodeProvider's per-role permissions with native grok flags instead of an opencode permission block + JS guard plugins. Uses tomli_w. The rendered config + env injection are validated live against grok-build (the model called the server through it).
This commit is contained in:
@@ -0,0 +1,198 @@
|
||||
"""Render a Grok CLI agent's runtime config + per-role flags at container start.
|
||||
|
||||
The ``roboco-agent-grok-cli`` image's entrypoint runs ``python -m
|
||||
roboco.llm.providers.grok_cli_config`` to turn the mounted Claude Code
|
||||
``mcp-config.json`` into ``~/.grok/config.toml`` (``[mcp_servers]``) and to
|
||||
compute the per-role ``grok -p`` flags (permissions / reasoning effort / turn
|
||||
cap), written one token per line to an args file the entrypoint splices into the
|
||||
command. Keeping the translation in importable Python (not a shell heredoc)
|
||||
makes it unit-testable.
|
||||
|
||||
Parity with ``ClaudeCodeProvider``'s per-role permissions, expressed as native
|
||||
grok flags instead of an opencode permission block + a JS guard plugin:
|
||||
|
||||
* **subagents** — every role gets ``--disallowed-tools Agent``: no RoboCo agent
|
||||
spawns the CLI's own subagents (work is driven through the gateway verbs).
|
||||
* **editing** — roles that don't write code (``role_config.allows_write`` is
|
||||
False) have the edit tool removed (``--disallowed-tools search_replace``).
|
||||
* **shell** — roles that never run a shell (review / board) have bash removed
|
||||
(``--disallowed-tools run_terminal_cmd``).
|
||||
* **git mutation** — bash-capable roles keep a shell, but raw git mutation is
|
||||
denied (``--deny "Bash(git push*)"`` ...): agents commit / push through the
|
||||
gateway verbs, never raw git (the opencode secret-scrub git-ops rule, ported).
|
||||
* **destructive** — ``--deny "Bash(rm -rf*)"`` for every bash-capable role.
|
||||
* **reasoning** — ``--effort`` by role (``low`` for coordination / docs / board;
|
||||
full for the code-quality roles). A global ``ROBOCO_GROK_REASONING_EFFORT``
|
||||
override wins.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
|
||||
from roboco.agents_config import get_agent_role
|
||||
from roboco.services.gateway.role_config import get_role_config
|
||||
|
||||
# grok reads its global config from ``$HOME/.grok/config.toml`` (the agent's HOME
|
||||
# is ``/home/agent``; the host ``auth.json`` is mounted alongside it).
|
||||
GROK_CONFIG_PATH = Path.home() / ".grok" / "config.toml"
|
||||
# The entrypoint reads the computed flags (one token per line) from this file.
|
||||
GROK_ARGS_PATH = Path(os.environ.get("ROBOCO_GROK_ARGS_FILE", "/tmp/roboco-grok-args"))
|
||||
|
||||
# Hard ceiling on agentic turns (loop guard; replaces the opencode budget-feed
|
||||
# loop cap). Operator-tunable.
|
||||
_DEFAULT_MAX_TURNS = 200
|
||||
|
||||
# Roles that request reduced reasoning (grok bills reasoning at the output rate,
|
||||
# so it dominates cost). Code-quality roles keep full reasoning; coordination /
|
||||
# docs / board roles ask for ``low``.
|
||||
_MINIMAL_REASONING_ROLES = frozenset(
|
||||
{
|
||||
"cell_pm",
|
||||
"main_pm",
|
||||
"documenter",
|
||||
"product_owner",
|
||||
"head_marketing",
|
||||
"auditor",
|
||||
"prompter",
|
||||
"secretary",
|
||||
}
|
||||
)
|
||||
_FULL_REASONING_OVERRIDES = frozenset({"default", "full", "none", ""})
|
||||
|
||||
# Roles that legitimately run a shell. Review / board roles never do.
|
||||
_BASH_ROLES = frozenset({"developer", "documenter", "cell_pm", "main_pm"})
|
||||
|
||||
# Grok CLI tool IDs (from the CLI's --tools/--disallowed-tools reference).
|
||||
_TOOL_SHELL = "run_terminal_cmd"
|
||||
_TOOL_EDIT = "search_replace"
|
||||
_TOOL_SUBAGENT = "Agent"
|
||||
|
||||
# Raw git mutation is gateway-mediated (the commit / open_pr verbs); agents never
|
||||
# push or commit via raw bash. Denied for every bash-capable role.
|
||||
_GIT_MUTATE_DENY = (
|
||||
"Bash(git push*)",
|
||||
"Bash(git fetch*)",
|
||||
"Bash(git pull*)",
|
||||
"Bash(git clone*)",
|
||||
"Bash(git commit*)",
|
||||
"Bash(git remote*)",
|
||||
"Bash(git reset*)",
|
||||
)
|
||||
_DESTRUCTIVE_DENY = ("Bash(rm -rf*)",)
|
||||
|
||||
|
||||
def render_config_toml(mcp_config: dict[str, Any]) -> str:
|
||||
"""Translate Claude Code ``mcpServers`` into grok's ``[mcp_servers]`` TOML.
|
||||
|
||||
``{"command": "uv", "args": [...], "env": {...}}`` becomes a
|
||||
``[mcp_servers.<name>]`` table with the same fields. Returns an empty string
|
||||
when there are no servers (a config with no MCP block is valid).
|
||||
"""
|
||||
servers: dict[str, dict[str, Any]] = {}
|
||||
for name, spec in (mcp_config.get("mcpServers") or {}).items():
|
||||
block: dict[str, Any] = {
|
||||
"command": str(spec.get("command", "")),
|
||||
"args": [str(a) for a in (spec.get("args") or [])],
|
||||
}
|
||||
env = spec.get("env") or {}
|
||||
if env:
|
||||
block["env"] = {str(k): str(v) for k, v in env.items()}
|
||||
servers[str(name)] = block
|
||||
return tomli_w.dumps({"mcp_servers": servers}) if servers else ""
|
||||
|
||||
|
||||
def _allows_write(role: str) -> bool:
|
||||
"""True if the role writes code (``role_config.allows_write``)."""
|
||||
try:
|
||||
return bool(get_role_config(role).allows_write)
|
||||
except KeyError:
|
||||
return False
|
||||
|
||||
|
||||
def _disallowed_tools(role: str) -> str:
|
||||
"""Comma-separated ``--disallowed-tools`` value for a role."""
|
||||
tools = [_TOOL_SUBAGENT]
|
||||
if role not in _BASH_ROLES:
|
||||
tools.append(_TOOL_SHELL)
|
||||
if not _allows_write(role):
|
||||
tools.append(_TOOL_EDIT)
|
||||
return ",".join(tools)
|
||||
|
||||
|
||||
def _deny_rules(role: str) -> list[str]:
|
||||
"""``--deny`` permission rules for a role (only bash-capable roles need any)."""
|
||||
if role not in _BASH_ROLES:
|
||||
return [] # bash removed entirely → nothing left to gate
|
||||
return [*_DESTRUCTIVE_DENY, *_GIT_MUTATE_DENY]
|
||||
|
||||
|
||||
def _effort_for(role: str) -> str | None:
|
||||
"""Resolve ``--effort`` for a role; ``None`` keeps grok's default reasoning.
|
||||
|
||||
A global ``ROBOCO_GROK_REASONING_EFFORT`` override wins over the per-role
|
||||
default (``default`` / ``full`` / empty disables the reduction).
|
||||
"""
|
||||
override = os.environ.get("ROBOCO_GROK_REASONING_EFFORT", "").strip()
|
||||
if override:
|
||||
return (
|
||||
None if override.lower() in _FULL_REASONING_OVERRIDES else override.lower()
|
||||
)
|
||||
return "low" if role in _MINIMAL_REASONING_ROLES else None
|
||||
|
||||
|
||||
def grok_cli_args(agent_id: str, *, max_turns: int = _DEFAULT_MAX_TURNS) -> list[str]:
|
||||
"""The per-role ``grok -p`` flag tokens for an agent (excludes ``-p``/model/cwd).
|
||||
|
||||
Order: tool removal, turn cap, deny rules, then effort. Each token is a
|
||||
separate list element so the entrypoint can splice them without shell quoting.
|
||||
"""
|
||||
role = get_agent_role(agent_id) or ""
|
||||
args: list[str] = ["--disallowed-tools", _disallowed_tools(role)]
|
||||
args += ["--max-turns", str(max_turns)]
|
||||
for rule in _deny_rules(role):
|
||||
args += ["--deny", rule]
|
||||
effort = _effort_for(role)
|
||||
if effort:
|
||||
args += ["--effort", effort]
|
||||
return args
|
||||
|
||||
|
||||
def _load_mcp_config(path: str) -> dict[str, Any]:
|
||||
"""Load the mounted mcp-config.json, tolerating a missing / invalid file."""
|
||||
try:
|
||||
with Path(path).open(encoding="utf-8") as fh:
|
||||
loaded = json.load(fh)
|
||||
return loaded if isinstance(loaded, dict) else {}
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Entrypoint: write ``~/.grok/config.toml`` + the per-role args file."""
|
||||
agent_id = os.environ.get("ROBOCO_AGENT_ID", "")
|
||||
mcp_path = os.environ.get("ROBOCO_MCP_CONFIG", "/app/mcp-config.json")
|
||||
try:
|
||||
max_turns = int(
|
||||
os.environ.get("ROBOCO_GROK_MAX_TURNS", str(_DEFAULT_MAX_TURNS))
|
||||
)
|
||||
except ValueError:
|
||||
max_turns = _DEFAULT_MAX_TURNS
|
||||
|
||||
GROK_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
GROK_CONFIG_PATH.write_text(
|
||||
render_config_toml(_load_mcp_config(mcp_path)), encoding="utf-8"
|
||||
)
|
||||
GROK_ARGS_PATH.write_text(
|
||||
"\n".join(grok_cli_args(agent_id, max_turns=max_turns)) + "\n", encoding="utf-8"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,94 @@
|
||||
"""grok_cli_config — mcp-config → config.toml + per-role grok CLI flags."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tomllib
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from roboco.llm.providers import grok_cli_config as gc
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest
|
||||
|
||||
_SAMPLE_MCP = {
|
||||
"mcpServers": {
|
||||
"roboco-flow": {
|
||||
"command": "uv",
|
||||
"args": ["run", "--no-sync", "python", "-m", "roboco.mcp.flow_server"],
|
||||
"env": {"ROBOCO_AGENT_ID": "be-dev-1", "ROBOCO_AGENT_TOKEN": "tok-123"},
|
||||
},
|
||||
"roboco-do": {"command": "uv", "args": ["run", "x"]},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _disallowed(args: list[str]) -> str:
|
||||
return args[args.index("--disallowed-tools") + 1]
|
||||
|
||||
|
||||
def test_render_config_toml_is_valid_toml_and_injects_env() -> None:
|
||||
parsed = tomllib.loads(gc.render_config_toml(_SAMPLE_MCP))
|
||||
flow = parsed["mcp_servers"]["roboco-flow"]
|
||||
assert flow["command"] == "uv"
|
||||
assert flow["args"][:2] == ["run", "--no-sync"]
|
||||
# The gateway env (token) reaches the server block — the keystone.
|
||||
assert flow["env"]["ROBOCO_AGENT_TOKEN"] == "tok-123"
|
||||
# A server with no env omits the key entirely.
|
||||
assert "env" not in parsed["mcp_servers"]["roboco-do"]
|
||||
|
||||
|
||||
def test_render_config_toml_empty_when_no_servers() -> None:
|
||||
assert gc.render_config_toml({}) == ""
|
||||
assert gc.render_config_toml({"mcpServers": {}}) == ""
|
||||
|
||||
|
||||
def test_developer_flags(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("ROBOCO_GROK_REASONING_EFFORT", raising=False)
|
||||
args = gc.grok_cli_args("be-dev-1")
|
||||
dis = _disallowed(args)
|
||||
# Developer writes code + runs a shell → only subagents removed.
|
||||
assert "Agent" in dis
|
||||
assert "run_terminal_cmd" not in dis
|
||||
assert "search_replace" not in dis
|
||||
# Raw git mutation + destructive ops denied even with a shell.
|
||||
assert "--deny" in args
|
||||
assert "Bash(git push*)" in args
|
||||
assert "Bash(rm -rf*)" in args
|
||||
# Code-quality role keeps full reasoning.
|
||||
assert "--effort" not in args
|
||||
|
||||
|
||||
def test_pr_reviewer_flags(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("ROBOCO_GROK_REASONING_EFFORT", raising=False)
|
||||
args = gc.grok_cli_args("pr-reviewer-1")
|
||||
dis = _disallowed(args)
|
||||
assert "run_terminal_cmd" in dis # read-only reviewer: no shell
|
||||
assert "search_replace" in dis # no editing
|
||||
assert "--deny" not in args # bash removed → no command rules left
|
||||
assert "--effort" not in args # code-quality reviewer keeps full reasoning
|
||||
|
||||
|
||||
def test_main_pm_keeps_shell_but_denies_git(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("ROBOCO_GROK_REASONING_EFFORT", raising=False)
|
||||
args = gc.grok_cli_args("main-pm")
|
||||
dis = _disallowed(args)
|
||||
assert "run_terminal_cmd" not in dis # PM keeps a shell
|
||||
assert "search_replace" in dis # but does not write code
|
||||
assert "Bash(git push*)" in args # raw git mutation denied
|
||||
assert args[args.index("--effort") + 1] == "low"
|
||||
|
||||
|
||||
def test_effort_override(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("ROBOCO_GROK_REASONING_EFFORT", "high")
|
||||
assert (
|
||||
gc.grok_cli_args("be-dev-1")[gc.grok_cli_args("be-dev-1").index("--effort") + 1]
|
||||
== "high"
|
||||
)
|
||||
# "full" disables the per-role reduction entirely.
|
||||
monkeypatch.setenv("ROBOCO_GROK_REASONING_EFFORT", "full")
|
||||
assert "--effort" not in gc.grok_cli_args("main-pm")
|
||||
|
||||
|
||||
def test_max_turns_is_emitted() -> None:
|
||||
args = gc.grok_cli_args("be-dev-1", max_turns=7)
|
||||
assert args[args.index("--max-turns") + 1] == "7"
|
||||
Reference in New Issue
Block a user