feat: per-role tool lists and system prompt overlays

Each role in model_registry.json can now carry two optional keys:
  system_append — injected into the system prompt at position 7 (after
                  memory, closest to the turn) for the active chat_role
  tools         — explicit tool allow-list; intersected with the user's
                  access-level filter so it can only restrict, never elevate

No changes needed for existing users — missing keys fall back to current
behavior. Add keys to a role to give it a specialty focus:

  "coder": {
    "primary": "claude_cli",
    "system_append": "You are in code-specialist mode...",
    "tools": ["web_search", "file_read", "shell_exec", "scratch_write"]
  }

Changes:
- model_registry.py: get_role_config() returns system_append + tools
- context_loader.py: role_append param appended as "--- Role Context ---"
- tools/__init__.py: get_tools_for_role/get_openai_tools_for_role accept
  optional tool_list and intersect with access-level filter
- orchestrator_engine.py: tool_list threaded through run/resume/checkpoint
- openai_orchestrator.py: tool_list threaded through run/resume/checkpoint;
  _build_client now calls get_openai_tools_for_role instead of returning
  unfiltered OPENAI_TOOL_SCHEMAS
- routers/orchestrator.py: pulls role_cfg for chat_role, passes both
  role_append and tool_list to context loader and engine

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

cortex/context_loader.py

@@ -17,6 +17,7 @@ def load_context(
     include_long: bool = True,
     include_mid: bool = True,
     include_short: bool = True,
+    role_append: str = "",
 ) -> str:
     """
     Build the system-prompt context block for a given tier and memory toggles.
@@ -28,6 +29,9 @@ def load_context(
     Tier 2  — + USER full + PROTOCOLS + memory      (~5,000 tokens)
     Tier 3  — + last 2 raw session logs             (~15,000 tokens)
     Tier 4  — + last 7 raw session logs             (~50,000 tokens)
+
+    role_append — optional text injected last (closest to the turn),
+                  sourced from the active role's system_append config.
     """
     inara_dir = persona_path()
     parts = []
@@ -107,4 +111,8 @@ def load_context(
             for sf in session_files:
                 parts.append(f"--- Session: {sf.name} ---\n{sf.read_text()}")
 
+    # ── 7. Role-specific instructions (always last — closest to the turn) ──
+    if role_append and role_append.strip():
+        parts.append(f"--- Role Context ---\n{role_append.strip()}")
+
     return "\n\n".join(parts)

cortex/model_registry.py

@@ -415,6 +415,22 @@ def get_best_local_model(username: str, role: str = "chat") -> dict | None:
     return None
 
 
+def get_role_config(username: str, role: str) -> dict:
+    """
+    Return supplemental config for a role: system_append and tools.
+
+    Both keys are optional in the registry — missing means "use defaults":
+      system_append: str  — appended to the system prompt for this role
+      tools: list[str] | None — explicit tool allow-list (None = no restriction)
+    """
+    registry = _load(username)
+    role_cfg = registry.get("roles", {}).get(role, {})
+    return {
+        "system_append": role_cfg.get("system_append", ""),
+        "tools":         role_cfg.get("tools") or None,
+    }
+
+
 def get_model_for_slot(username: str, role: str, slot: str) -> dict | None:
     """
     Resolve a single named priority slot from a role without walking the fallback chain.

cortex/openai_orchestrator.py

@@ -45,6 +45,7 @@ async def run(
     model_cfg: dict | None = None,
     respond_with_final: bool = True,
     user_role: str = "user",
+    tool_list: list[str] | None = None,
     confirm_allow: set[str] | None = None,
     confirm_deny: set[str] | None = None,
 ) -> OrchestratorResult:
@@ -71,7 +72,7 @@ async def run(
     _confirm_deny = frozenset(confirm_deny or ())
     effective_confirm = (CONFIRM_REQUIRED - set(_confirm_allow)) | set(_confirm_deny)
 
-    client, model_name, active_tools = _build_client(model_cfg)
+    client, model_name, active_tools = _build_client(model_cfg, user_role, tool_list)
 
     sys_content = (system_prompt or "") + _TOOL_INSTRUCTION
     messages: list[dict] = [{"role": "system", "content": sys_content}]
@@ -95,6 +96,7 @@ async def run(
         model_cfg=model_cfg,
         respond_with_final=respond_with_final,
         user_role=user_role,
+        tool_list=tool_list,
         confirm_allow=_confirm_allow,
         confirm_deny=_confirm_deny,
         starting_round=0,
@@ -121,7 +123,7 @@ async def run(
 
 async def resume(checkpoint: OrchestrateCheckpoint, confirmed: bool) -> OrchestratorResult:
     """Continue an OpenAI orchestrator job that was paused at a confirmation gate."""
-    client, model_name, active_tools = _build_client(checkpoint.model_cfg)
+    client, model_name, active_tools = _build_client(checkpoint.model_cfg, checkpoint.user_role, checkpoint.tool_list)
 
     effective_confirm = (CONFIRM_REQUIRED - set(checkpoint.confirm_allow)) | set(checkpoint.confirm_deny)
 
@@ -138,8 +140,7 @@ async def resume(checkpoint: OrchestrateCheckpoint, confirmed: bool) -> Orchestr
 
     for pt in checkpoint.pending_tools:
         if confirmed:
-            _, callables = get_tools_for_role(checkpoint.user_role)
+            result_str = await _execute_tool_dict(pt["name"], pt["args"], checkpoint.user_role, checkpoint.tool_list)
-            result_str = await _execute_tool_dict(pt["name"], pt["args"], checkpoint.user_role)
             logger.info("Confirmed tool %s → %d chars", pt["name"], len(result_str))
         else:
             result_str = "Action denied by user."
@@ -162,6 +163,7 @@ async def resume(checkpoint: OrchestrateCheckpoint, confirmed: bool) -> Orchestr
         model_cfg=checkpoint.model_cfg,
         respond_with_final=checkpoint.respond_with_final,
         user_role=checkpoint.user_role,
+        tool_list=checkpoint.tool_list,
         confirm_allow=checkpoint.confirm_allow,
         confirm_deny=checkpoint.confirm_deny,
         starting_round=checkpoint.rounds_used,
@@ -200,6 +202,7 @@ async def _run_from_messages(
     confirm_allow: frozenset,
     confirm_deny: frozenset,
     starting_round: int = 0,
+    tool_list: list[str] | None = None,
 ) -> tuple[str, OrchestrateCheckpoint | None]:
     """
     Run the OpenAI ReAct loop from the current messages state.
@@ -253,7 +256,7 @@ async def _run_from_messages(
                     pending_tools.append({"name": name, "args": args_parsed, "tool_call_id": tc.id})
                     logger.info("Tool %s blocked — confirmation required", name)
                 else:
-                    result_str = await _execute_tool(name, tc.function.arguments, user_role)
+                    result_str = await _execute_tool(name, tc.function.arguments, user_role, tool_list)
                     logger.info("Tool %s → %d chars", name, len(result_str))
                     executed_results.append({"name": name, "args": args_parsed, "result": result_str, "tool_call_id": tc.id})
                     tool_call_log.append({"tool": name, "args": args_parsed, "result": result_str})
@@ -286,6 +289,7 @@ async def _run_from_messages(
                     model_cfg=model_cfg,
                     respond_with_final=respond_with_final,
                     user_role=user_role,
+                    tool_list=tool_list,
                     confirm_allow=confirm_allow,
                     confirm_deny=confirm_deny,
                     rounds_used=round_num + 2,
@@ -311,7 +315,11 @@ async def _run_from_messages(
     return final_response, None
 
 
-def _build_client(model_cfg: dict | None) -> tuple:
+def _build_client(
+    model_cfg: dict | None,
+    user_role: str = "user",
+    tool_list: list[str] | None = None,
+) -> tuple:
     """Build AsyncOpenAI client and return (client, model_name, active_tools)."""
     if not model_cfg:
         raise RuntimeError("model_cfg is required for the OpenAI orchestrator")
@@ -327,12 +335,18 @@ def _build_client(model_cfg: dict | None) -> tuple:
     if host_type == "openwebui":
         base_url = base_url + "/api"
     client = AsyncOpenAI(base_url=base_url, api_key=api_key)
-    return client, model_name, OPENAI_TOOL_SCHEMAS
+    active_tools = get_openai_tools_for_role(user_role, tool_list)
+    return client, model_name, active_tools
 
 
-async def _execute_tool(name: str, arguments_json: str, user_role: str = "user") -> str:
+async def _execute_tool(
+    name: str,
+    arguments_json: str,
+    user_role: str = "user",
+    tool_list: list[str] | None = None,
+) -> str:
     """Parse tool arguments and execute with role-filtered callables."""
-    _, callables = get_tools_for_role(user_role)
+    _, callables = get_tools_for_role(user_role, tool_list)
     try:
         args = json.loads(arguments_json)
     except json.JSONDecodeError:
@@ -344,9 +358,14 @@ async def _execute_tool(name: str, arguments_json: str, user_role: str = "user")
         return f"Tool error: {e}"
 
 
-async def _execute_tool_dict(name: str, args: dict, user_role: str = "user") -> str:
+async def _execute_tool_dict(
+    name: str,
+    args: dict,
+    user_role: str = "user",
+    tool_list: list[str] | None = None,
+) -> str:
     """Execute a tool from a pre-parsed args dict."""
-    _, callables = get_tools_for_role(user_role)
+    _, callables = get_tools_for_role(user_role, tool_list)
     try:
         return await call_tool(name, args, callables)
     except Exception as e:

cortex/orchestrator_engine.py

@@ -65,6 +65,7 @@ class OrchestrateCheckpoint:
     respond_with_final: bool = True
     # Common
     user_role: str = "user"
+    tool_list: list[str] | None = None
     confirm_allow: frozenset = field(default_factory=frozenset)
     confirm_deny: frozenset = field(default_factory=frozenset)
     rounds_used: int = 0
@@ -88,6 +89,7 @@ async def run(
     model_name: str | None = None,
     response_role: str = "chat",
     user_role: str = "user",
+    tool_list: list[str] | None = None,
     confirm_allow: set[str] | None = None,
     confirm_deny: set[str] | None = None,
 ) -> OrchestratorResult:
@@ -101,6 +103,8 @@ async def run(
         respond_with_claude: If False, return Gemini's summary as the response (useful for
                              background/cron tasks where a polished reply isn't needed)
         gemini_api_key:     Per-user Gemini API key (falls back to GEMINI_API_KEY in .env)
+        tool_list:          Optional explicit tool allow-list from role config; intersected
+                            with user_role access-level filter (cannot elevate privileges)
         confirm_allow:      Tools to bypass the confirmation gate for this user
         confirm_deny:       Tools to always block for this user
 
@@ -124,7 +128,7 @@ async def run(
     contents: list[types.Content] = [
         types.Content(role="user", parts=[types.Part(text=task_with_context)])
     ]
-    tool_declarations, tool_callables = get_tools_for_role(user_role)
+    tool_declarations, tool_callables = get_tools_for_role(user_role, tool_list)
     tool_call_log: list[dict] = []
 
     gemini_summary, checkpoint = await _run_from_contents(
@@ -141,6 +145,7 @@ async def run(
         respond_with_claude=respond_with_claude,
         response_role=response_role,
         user_role=user_role,
+        tool_list=tool_list,
         confirm_allow=_confirm_allow,
         confirm_deny=_confirm_deny,
         starting_round=0,
@@ -171,7 +176,7 @@ async def resume(checkpoint: OrchestrateCheckpoint, confirmed: bool) -> Orchestr
     """Continue a job that was paused at a confirmation gate."""
     api_key = checkpoint.gemini_api_key or settings.gemini_api_key
     client = genai.Client(api_key=api_key)
-    tool_declarations, tool_callables = get_tools_for_role(checkpoint.user_role)
+    tool_declarations, tool_callables = get_tools_for_role(checkpoint.user_role, checkpoint.tool_list)
 
     effective_confirm = (CONFIRM_REQUIRED - set(checkpoint.confirm_allow)) | set(checkpoint.confirm_deny)
 
@@ -215,6 +220,7 @@ async def resume(checkpoint: OrchestrateCheckpoint, confirmed: bool) -> Orchestr
         respond_with_claude=checkpoint.respond_with_claude,
         response_role=checkpoint.response_role,
         user_role=checkpoint.user_role,
+        tool_list=checkpoint.tool_list,
         confirm_allow=checkpoint.confirm_allow,
         confirm_deny=checkpoint.confirm_deny,
         starting_round=checkpoint.rounds_used,
@@ -259,6 +265,7 @@ async def _run_from_contents(
     confirm_deny: frozenset,
     starting_round: int = 0,
     gemini_api_key: str | None = None,
+    tool_list: list[str] | None = None,
 ) -> tuple[str, OrchestrateCheckpoint | None]:
     """
     Run the ReAct loop from the current contents state.
@@ -364,6 +371,7 @@ async def _run_from_contents(
                 respond_with_claude=respond_with_claude,
                 response_role=response_role,
                 user_role=user_role,
+                tool_list=tool_list,
                 confirm_allow=confirm_allow,
                 confirm_deny=confirm_deny,
                 rounds_used=round_num + 2,

cortex/routers/orchestrator.py

@@ -196,11 +196,13 @@ async def _run_job(job_id: str, req: OrchestrateRequest, user: str) -> None:
         from session_store import load as load_session, save as save_session, generate_session_id
 
         tier = req.tier or settings.default_tier
+        role_cfg = model_registry.get_role_config(user, req.chat_role)
         system_prompt = load_context(
             tier,
             include_long=req.include_long,
             include_mid=req.include_mid,
             include_short=req.include_short,
+            role_append=role_cfg.get("system_append", ""),
         )
 
         session_id = req.session_id or generate_session_id()
@@ -209,6 +211,7 @@ async def _run_job(job_id: str, req: OrchestrateRequest, user: str) -> None:
 
         orch_model = model_registry.get_model_for_role(user, "orchestrator")
         user_role = get_user_role(user)
+        tool_list = role_cfg.get("tools")
 
         policy = get_tool_policy(user)
         confirm_allow = set(policy.get("allow", []))
@@ -222,6 +225,7 @@ async def _run_job(job_id: str, req: OrchestrateRequest, user: str) -> None:
                 model_cfg=orch_model,
                 respond_with_final=req.respond_with_claude,
                 user_role=user_role,
+                tool_list=tool_list,
                 confirm_allow=confirm_allow,
                 confirm_deny=confirm_deny,
             )
@@ -239,6 +243,7 @@ async def _run_job(job_id: str, req: OrchestrateRequest, user: str) -> None:
                 model_name=orch_model.get("model_name") if orch_model else None,
                 response_role=req.chat_role,
                 user_role=user_role,
+                tool_list=tool_list,
                 confirm_allow=confirm_allow,
                 confirm_deny=confirm_deny,
             )

cortex/tools/__init__.py

@@ -1100,19 +1100,38 @@ OPENAI_TOOL_SCHEMAS: list[dict] = _build_openai_tools()
 # Role-filtered tool access
 # ---------------------------------------------------------------------------
 
-def get_tools_for_role(role: str) -> tuple[list, dict]:
+def get_tools_for_role(
+    role: str,
+    tool_list: list[str] | None = None,
+) -> tuple[list, dict]:
     """Return (gemini_tool_declarations, callables_dict) filtered to tools the role can use.
 
+    role      — user access level ("user" | "admin"); gates admin-only tools
+    tool_list — optional explicit allow-list from role config (e.g. coder role);
+                intersected with the access-level filter so it can only restrict,
+                never elevate privileges
+
     Usage in orchestrator:
-        tool_declarations, tool_callables = get_tools_for_role(user_role)
+        tool_declarations, tool_callables = get_tools_for_role(user_role, tool_list)
     """
     allowed = {name for name in _CALLABLES if _role_allowed(name, role)}
+    if tool_list is not None:
+        allowed &= set(tool_list)
     decls = [d for d in _ALL_DECLARATIONS if d.name in allowed]
     callables = {k: v for k, v in _CALLABLES.items() if k in allowed}
     return [types.Tool(function_declarations=decls)], callables
 
 
-def get_openai_tools_for_role(role: str) -> list[dict]:
+def get_openai_tools_for_role(
-    """Return OpenAI tool schemas filtered to tools the role can use."""
+    role: str,
+    tool_list: list[str] | None = None,
+) -> list[dict]:
+    """Return OpenAI tool schemas filtered to tools the role can use.
+
+    role      — user access level ("user" | "admin")
+    tool_list — optional explicit allow-list from role config
+    """
     allowed = {name for name in _CALLABLES if _role_allowed(name, role)}
+    if tool_list is not None:
+        allowed &= set(tool_list)
     return [t for t in OPENAI_TOOL_SCHEMAS if t["function"]["name"] in allowed]