ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Unsloth start: add local subagents for Claude Code, Codex, OpenCode and Pi (#7326)

Browse source 
Bring the local-subagent support onto main. The original change (#7316) merged
into the stacked pr/daniel-unsloth-start-audit branch rather than main, and #7313
reached main via squash, so these files never landed on main.

Adds --as-subagent for claude, codex, opencode and pi: the parent agent keeps its
own cloud model while a locally served GGUF is registered as a delegated subagent,
using ephemeral per-session config that never touches the user's real agent config.
This commit is contained in:
Daniel Han 2026-07-22 04:34:58 -07:00 • committed by GitHub
commit 968e6230a0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 2108 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

7
README.md
View file

@ -86,6 +86,13 @@ Replace `claude` with any supported agent:
| OpenCode | `unsloth start opencode` |
| Pi Coding Agent | `unsloth start pi` |
Claude Code, Codex, OpenCode and Pi can keep their current model and use Unsloth as a local
subagent:
```bash
unsloth start claude --as-subagent --model unsloth/model-GGUF:quant
```
## 📥 Install
Unsloth can be used in two ways: through **[Unsloth Studio](https://unsloth.ai/docs/new/studio/)**, the web UI, or through **Unsloth Core**, the code-based version. Each has different requirements.

2
pyproject.toml
View file

@ -42,7 +42,7 @@ version = {attr = "unsloth.models._utils.__version__"}
include-package-data = true
[tool.setuptools.package-data]
unsloth_cli = ["codex_fallback_prompt.md"]
unsloth_cli = ["codex_fallback_prompt.md", "pi_subagent.ts"]
studio = [
"*.sh",
"*.ps1",

366
unsloth_cli/claude_subagent_mcp.py Normal file
View file

@ -0,0 +1,366 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Small stdio MCP bridge from cloud Claude Code to a local Claude Code child."""
from __future__ import annotations
import json
import os
import signal
import shutil
import subprocess
import sys
import threading
import time
from typing import Any, Callable
from unsloth_cli.commands.start import (
_CLAUDE_ENV_UNSET,
_SUBAGENT_DESCRIPTION,
_SUBAGENT_INSTRUCTIONS,
_claude_flags,
_claude_local_env,
_wsl_shim_env,
)
_MAX_RESULT_CHARACTERS = 100_000
_CANCEL_POLL_SECONDS = 0.1
_CANCEL_GRACE_SECONDS = 2.0
def _required_env(name: str) -> str:
value = os.environ.get(name, "").strip()
if not value:
raise RuntimeError(f"Missing {name}.")
return value
def _bounded(text: str) -> str:
if len(text) <= _MAX_RESULT_CHARACTERS:
return text
return text[:_MAX_RESULT_CHARACTERS] + "\n\n[Local agent output truncated]"
def _result_text(stdout: str) -> str:
lines = [line for line in stdout.splitlines() if line.strip()]
candidates = [stdout.strip(), *reversed(lines)]
for candidate in candidates:
try:
payload = json.loads(candidate)
except ValueError:
continue
if not isinstance(payload, dict):
continue
result = payload.get("result")
if payload.get("is_error"):
raise RuntimeError(str(result or "The local Claude agent failed."))
if isinstance(result, str) and result.strip():
return _bounded(result.strip())
raise RuntimeError("The local Claude agent returned no readable result.")
def _stop_child(process: subprocess.Popen) -> None:
"""Stop the Claude child and any tool processes it started."""
if process.poll() is not None:
if os.name != "nt":
# Leader exited, but its tool processes may still be running.
try:
os.killpg(process.pid, signal.SIGTERM)
except OSError:
return
time.sleep(_CANCEL_GRACE_SECONDS)
try:
os.killpg(process.pid, signal.SIGKILL)
except OSError:
pass
return
if os.name == "nt":
try:
completed = subprocess.run(
["taskkill", "/PID", str(process.pid), "/T", "/F"],
capture_output = True,
timeout = 15,
check = False,
)
except Exception:
completed = None
# A failed taskkill must not leave the child running through the grace wait.
if (completed is None or completed.returncode != 0) and process.poll() is None:
process.terminate()
else:
try:
os.killpg(process.pid, signal.SIGTERM)
except OSError:
process.terminate()
try:
process.wait(timeout = _CANCEL_GRACE_SECONDS)
except subprocess.TimeoutExpired:
if os.name == "nt":
process.kill()
else:
try:
os.killpg(process.pid, signal.SIGKILL)
except OSError:
process.kill()
process.wait()
else:
if os.name != "nt":
# Leader is gone; kill any surviving group members.
try:
os.killpg(process.pid, signal.SIGKILL)
except OSError:
pass
def run_local_agent(task: str, cancel_event: threading.Event | None = None) -> str:
base = _required_env("UNSLOTH_CLAUDE_SUBAGENT_BASE_URL")
key = _required_env("UNSLOTH_CLAUDE_SUBAGENT_API_KEY")
model = _required_env("UNSLOTH_CLAUDE_SUBAGENT_MODEL")
window = int(os.environ.get("UNSLOTH_CLAUDE_SUBAGENT_CONTEXT_WINDOW", "0") or 0)
entry = {"id": model, "context_length": window}
local_env = _claude_local_env(base, key, entry)
child_env = dict(os.environ)
executable = shutil.which("claude")
if executable is None:
raise RuntimeError("`claude` is not installed or is not on PATH.")
cancel_event = cancel_event or threading.Event()
if cancel_event.is_set():
raise RuntimeError("The local Claude agent was cancelled.")
command = [
"claude",
"--model",
model,
*_claude_flags(model),
"--permission-mode",
(
"bypassPermissions"
if os.environ.get("UNSLOTH_CLAUDE_SUBAGENT_BYPASS_PERMISSIONS") == "1"
else "acceptEdits"
),
"--print",
"--output-format",
"json",
"--no-session-persistence",
"--append-system-prompt",
_SUBAGENT_INSTRUCTIONS,
f"Task: {task}",
]
bridged, wsl_names = _wsl_shim_env(command, local_env, _CLAUDE_ENV_UNSET)
if wsl_names:
from unsloth_cli.commands.start import _merge_wslenv
bridged = {**bridged, "PWD": os.getcwd()}
child_env["WSLENV"] = _merge_wslenv(child_env.get("WSLENV", ""), wsl_names)
for name in _CLAUDE_ENV_UNSET:
child_env[name] = ""
else:
for name in _CLAUDE_ENV_UNSET:
child_env.pop(name, None)
child_env.update(bridged)
popen_kwargs: dict[str, Any] = {
"cwd": os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd(),
"env": child_env,
"stdin": subprocess.DEVNULL,
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"text": True,
}
if os.name == "nt":
popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
else:
popen_kwargs["start_new_session"] = True
process = subprocess.Popen(
[executable, *command[1:]],
**popen_kwargs,
)
try:
while True:
try:
stdout, stderr = process.communicate(timeout = _CANCEL_POLL_SECONDS)
break
except subprocess.TimeoutExpired:
if cancel_event.is_set():
_stop_child(process)
raise RuntimeError("The local Claude agent was cancelled.")
except BaseException:
if process.poll() is None:
_stop_child(process)
raise
if process.returncode != 0:
detail = stderr.strip() or stdout.strip()
raise RuntimeError(
_bounded(detail) or f"Local Claude exited with code {process.returncode}."
)
return _result_text(stdout)
def _response(request: dict, run_agent: Callable[[str], str] = run_local_agent) -> dict | None:
request_id = request.get("id")
method = request.get("method")
if request_id is None:
return None
if method == "initialize":
protocol = (request.get("params") or {}).get("protocolVersion") or "2025-06-18"
result = {
"protocolVersion": protocol,
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "unsloth-local-agent", "version": "1.0.0"},
}
elif method == "ping":
result = {}
elif method == "tools/list":
result = {
"tools": [
{
"name": "unsloth_agent",
"title": "Unsloth local agent",
"description": _SUBAGENT_DESCRIPTION,
"inputSchema": {
"type": "object",
"properties": {
"task": {
"type": "string",
"description": "The complete task for the local Unsloth agent.",
}
},
"required": ["task"],
"additionalProperties": False,
},
"annotations": {
"readOnlyHint": False,
"destructiveHint": True,
"idempotentHint": False,
"openWorldHint": True,
},
"_meta": {"anthropic/maxResultSizeChars": _MAX_RESULT_CHARACTERS},
}
]
}
elif method == "tools/call":
params = request.get("params") or {}
arguments = params.get("arguments") or {}
task = arguments.get("task") if params.get("name") == "unsloth_agent" else None
if not isinstance(task, str) or not task.strip():
result = {
"content": [{"type": "text", "text": "A non-empty task is required."}],
"isError": True,
}
else:
try:
text = run_agent(task.strip())
result = {"content": [{"type": "text", "text": text}], "isError": False}
except Exception as exc:
result = {
"content": [{"type": "text", "text": str(exc)}],
"isError": True,
}
else:
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {"code": -32601, "message": f"Method not found: {method}"},
}
return {"jsonrpc": "2.0", "id": request_id, "result": result}
def serve(
stdin: Any = sys.stdin,
stdout: Any = sys.stdout,
run_agent: Callable[[str, threading.Event], str] = run_local_agent,
) -> None:
active: dict[object, threading.Event] = {}
workers: list[threading.Thread] = []
state_lock = threading.RLock()
output_lock = threading.Lock()
shutdown_started = threading.Event()
def cancel_active() -> None:
with state_lock:
pending = list(active.values())
for cancel_event in pending:
cancel_event.set()
def handle_shutdown(_signum: int, _frame: Any) -> None:
# Claude Code sends SIGINT (possibly repeatedly) to cancel a tool call. Only
# the first unwinds stdin; later ones must not interrupt process-tree cleanup.
first_signal = not shutdown_started.is_set()
shutdown_started.set()
cancel_active()
if first_signal:
raise KeyboardInterrupt
previous_handlers: dict[int, Any] = {}
if threading.current_thread() is threading.main_thread():
for signum in (signal.SIGINT, signal.SIGTERM):
previous_handlers[signum] = signal.signal(signum, handle_shutdown)
def send(response: dict | None) -> None:
if response is None:
return
with output_lock:
stdout.write(json.dumps(response, separators = (",", ":")) + "\n")
stdout.flush()
def call_tool(request: dict, request_id: object, cancel_event: threading.Event) -> None:
try:
response = _response(
request,
run_agent = lambda task: run_agent(task, cancel_event),
)
if not cancel_event.is_set():
send(response)
finally:
with state_lock:
if active.get(request_id) is cancel_event:
active.pop(request_id, None)
try:
for line in stdin:
try:
request = json.loads(line)
if not isinstance(request, dict):
response = None
elif request.get("method") == "notifications/cancelled":
request_id = (request.get("params") or {}).get("requestId")
with state_lock:
cancel_event = active.get(request_id)
if cancel_event is not None:
cancel_event.set()
response = None
elif request.get("method") == "tools/call" and request.get("id") is not None:
request_id = request["id"]
cancel_event = threading.Event()
with state_lock:
active[request_id] = cancel_event
worker = threading.Thread(
target = call_tool,
args = (request, request_id, cancel_event),
name = f"unsloth-agent-{request_id}",
)
workers.append(worker)
worker.start()
response = None
else:
response = _response(request)
except Exception as exc:
response = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": str(exc)},
}
send(response)
except KeyboardInterrupt:
pass
finally:
cancel_active()
for worker in workers:
if worker.ident is not None:
worker.join()
for signum, handler in previous_handlers.items():
signal.signal(signum, handler)
if __name__ == "__main__":
serve()

525
unsloth_cli/commands/start.py
View file

@ -73,11 +73,21 @@ _HERMES_POSIX_INSTALL_HINT = (
# windows and scales the compaction threshold back down to the real window.
_HERMES_MIN_CONTEXT = 65536
_PI_PROVIDER = "unsloth"
# OpenCode selects a model by "<providerID>/<modelID>" and honors a user
# disabled_providers list. Register the session provider under a dedicated id a
# user's disable list would never target, so the model is always selectable
# without the wrapper having to reconstruct (and override) OpenCode's full,
# multi-layer disabled_providers resolution.
_SUBAGENT_NAME = "unsloth"
_SUBAGENT_DESCRIPTION = (
"Local coding subagent powered by Unsloth for debugging, implementation, and codebase "
"research. Use when the user asks to spawn an Unsloth or local agent."
)
_SUBAGENT_INSTRUCTIONS = (
"You are a local coding subagent powered by Unsloth. Complete the assigned task directly, "
"use the available tools when useful, verify your work, and return a concise result to the "
"parent agent."
)
_CLAUDE_SUBAGENT_MCP_MODULE = "unsloth_cli.claude_subagent_mcp"
_CLAUDE_SUBAGENT_TOOL = "mcp__plugin_unsloth-local-agent_unsloth__unsloth_agent"
_PI_SUBAGENT_EXTENSION = Path(__file__).parent.parent / "pi_subagent.ts"
# OpenCode selects a model by "<providerID>/<modelID>". Use a dedicated id to avoid
# colliding with a user's providers; provider filters are set in the launch-time overlay.
_OPENCODE_PROVIDER = "unsloth-studio"
_PROVIDER_HEADER = f"[model_providers.{_CODEX_PROFILE}]"
_PASSTHROUGH = {"allow_extra_args": True, "ignore_unknown_options": True}
@ -158,6 +168,11 @@ _PERSIST_OPTION = typer.Option(
"the agent unchanged."
),
)
_AS_SUBAGENT_OPTION = typer.Option(
False,
"--as-subagent",
help = "Keep the coding agent's current model and add Unsloth as a local subagent.",
)
# Per-agent CLI flag for "run tools without prompting". OpenCode (native --auto is
# command-scoped, handled below) and OpenClaw (config-only) are absent from this prefix map.
@ -334,11 +349,54 @@ def _display_model_spec(model: str, variant: Optional[str]) -> str:
return f"{repo}:{selected_variant}" if selected_variant else model
def _subagent_model_id(
base: str,
key: str,
entry: dict,
requested_model: Optional[str],
requested_variant: Optional[str],
) -> str:
"""Return an API model id that preserves the selected GGUF variant.
Coding-agent model definitions outlive the initial load. If Unsloth later
unloads the model, a bare repository id may resolve to a different cached
quant. Include the explicit or currently loaded variant so an automatic
reload selects the same weights.
"""
model_id = str(entry["id"])
_, inline_variant = _split_repo_variant(requested_model or "")
variant = requested_variant or inline_variant
if not variant:
try:
status = _http_json("GET", f"{base}/api/inference/status", key)
except Exception:
status = {}
typer.echo(
"Warning: could not verify the loaded GGUF variant; a later reload "
"may pick a different cached quant. Pass :variant to pin it.",
err = True,
)
if status.get("is_gguf"):
variant = status.get("gguf_variant")
return (
_display_model_spec(model_id, str(variant))
if variant and _is_hub_model_id(model_id)
else model_id
)
def _fail(message: str) -> NoReturn:
typer.echo(message, err = True)
raise typer.Exit(code = 1)
def _reject_as_subagent(agent: str, args: list) -> None:
# Reject early; otherwise the flag reaches the agent binary and fails after
# Studio has already loaded the model.
if "--as-subagent" in args:
_fail(f"--as-subagent is not supported for {agent}.")
def _http_error_detail(exc: urllib.error.HTTPError) -> str:
try:
body = json.loads(exc.read().decode())
@ -1278,6 +1336,25 @@ def _claude_flags(model_id: str) -> list:
return [_DYNAMIC_SECTIONS_FLAG, "--settings", _claude_settings_overlay(model_id)]
def _claude_local_env(base: str, key: str, entry: dict) -> dict:
"""Build the local endpoint, cache, display, and compaction environment."""
model_id = entry["id"]
env = {
"ANTHROPIC_BASE_URL": base,
"ANTHROPIC_AUTH_TOKEN": key,
"ANTHROPIC_MODEL": model_id,
"CLAUDE_CODE_ATTRIBUTION_HEADER": "0",
"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
"CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1",
"CLAUDE_CODE_NO_FLICKER": "1",
}
window = entry.get("context_length") or entry.get("max_context_length")
if window:
env["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] = str(int(window))
env["CLAUDE_AUTOCOMPACT_PCT_OVERRIDE"] = "90"
return env
def _merge_codex_config(existing: str, base: str) -> str:
chunks = re.split(r"(?m)^(?=\[)", existing) # preamble, then one chunk per table
if not re.search(r"(?m)^\s*oss_provider\s*=", chunks[0]):
@ -1391,6 +1468,206 @@ def write_codex_config(base: str, model: dict, home: Path) -> None:
typer.echo(f"Updated {profile}")
def write_codex_subagent_config(base: str, key: str, model: dict, home: Path) -> Path:
"""Write a session-scoped Codex custom agent without replacing the main model."""
home.mkdir(parents = True, exist_ok = True)
model_id = model["id"]
window = model.get("context_length") or model.get("max_context_length")
catalog_name = "unsloth-model-catalog.json"
text = (
f"name = {json.dumps(_SUBAGENT_NAME)}\n"
f"description = {json.dumps(_SUBAGENT_DESCRIPTION)}\n"
f"developer_instructions = {json.dumps(_SUBAGENT_INSTRUCTIONS)}\n"
f"model_provider = {json.dumps(_CODEX_PROFILE)}\n"
f"model = {json.dumps(model_id)}\n"
)
if _codex_supports_model_catalog() and _CODEX_FALLBACK_PROMPT.is_file():
catalog = home / catalog_name
catalog_text = json.dumps(_codex_model_catalog(model), indent = 2) + "\n"
if not catalog.exists() or catalog.read_text(encoding = "utf-8") != catalog_text:
catalog.write_text(catalog_text, encoding = "utf-8")
typer.echo(f"Updated {catalog}")
text += f"model_catalog_json = {json.dumps(catalog_name)}\n"
if window:
text += f"model_context_window = {int(window)}\n"
credential = home / "unsloth-auth.json"
_write_private_json(credential, {"token": key})
auth_command = sys.executable
auth_args = [
"-c",
"import json,sys; print(json.load(open(sys.argv[1], encoding='utf-8'))['token'])",
str(credential),
]
if _wsl_windows_executable(["codex"]):
auth_command = "wsl.exe"
auth_args = [
"-d",
os.environ["WSL_DISTRO_NAME"],
"--",
sys.executable,
*auth_args,
]
text += (
f"\n{_PROVIDER_HEADER}\n"
'name = "Unsloth Studio"\n'
f"base_url = {json.dumps(base + '/v1')}\n"
'wire_api = "responses"\n'
f"\n{_PROVIDER_HEADER[:-1]}.auth]\n"
f"command = {json.dumps(auth_command)}\n"
f"args = {json.dumps(auth_args)}\n"
"timeout_ms = 5000\n"
)
path = home / f"{_SUBAGENT_NAME}.toml"
if not path.exists() or path.read_text(encoding = "utf-8") != text:
path.write_text(text, encoding = "utf-8")
typer.echo(f"Updated {path}")
return path
def _agent_config_path(path: Path, command: list) -> str:
"""Translate a generated config path when a Windows agent runs through WSL."""
return _wsl_windows_path(path) if _wsl_windows_executable(command) else str(path)
def _opencode_subagent_inline_config(path: Path, permission: dict) -> dict:
"""Keep the local provider visible without hiding the parent's allowed providers."""
inline: dict = {}
inherited = os.environ.get("OPENCODE_CONFIG_CONTENT")
if inherited:
try:
parsed = json.loads(inherited)
except ValueError:
_fail("OPENCODE_CONFIG_CONTENT is not valid JSON.")
if not isinstance(parsed, dict):
_fail("OPENCODE_CONFIG_CONTENT must contain a JSON object.")
inline.update(parsed)
def merge_provider_filters(effective_config: dict) -> None:
enabled = effective_config.get("enabled_providers")
if isinstance(enabled, list):
inline["enabled_providers"] = list(dict.fromkeys([*enabled, _OPENCODE_PROVIDER]))
disabled = effective_config.get("disabled_providers")
if isinstance(disabled, list) and _OPENCODE_PROVIDER in disabled:
inline["disabled_providers"] = [
provider for provider in disabled if provider != _OPENCODE_PROVIDER
]
# The inherited inline layer is already highest priority. Merge it even when
# OpenCode is not installed yet, as in fresh-install and --no-launch flows.
merge_provider_filters(inline)
effective = inline
executable = _which_with_install_dirs("opencode")
if executable is None:
typer.echo(
f"Warning: OpenCode is not installed, so provider filters could not be checked. "
f"The target configuration must allow '{_OPENCODE_PROVIDER}'.",
err = True,
)
else:
env = os.environ.copy()
env["OPENCODE_CONFIG"] = _agent_config_path(path, ["opencode"])
try:
resolved = subprocess.run(
[executable, "debug", "config"],
capture_output = True,
text = True,
timeout = 15,
env = env,
)
except Exception as exc:
_fail(f"Could not inspect OpenCode provider filters: {exc}")
if resolved.returncode != 0:
detail = resolved.stderr.strip() or resolved.stdout.strip()
_fail(f"Could not inspect OpenCode provider filters: {detail or 'unknown error'}")
try:
effective = json.loads(resolved.stdout)
except ValueError:
_fail("Could not inspect OpenCode provider filters: invalid JSON response.")
if not isinstance(effective, dict):
_fail("Could not inspect OpenCode provider filters: expected a JSON object.")
merge_provider_filters(effective)
depth = effective.get("subagent_depth")
inline["subagent_depth"] = (
depth if isinstance(depth, int) and not isinstance(depth, bool) and depth > 0 else 1
)
if permission:
inline["permission"] = permission
return inline
def write_claude_subagent_plugin(path: Path, server_env: dict) -> Path:
"""Write a session plugin that exposes the local Claude child through MCP."""
plugin = path / "unsloth-local-agent"
command = sys.executable
args = ["-m", _CLAUDE_SUBAGENT_MCP_MODULE]
mcp_env = dict(server_env)
if _wsl_windows_executable(["claude"]):
command = "wsl.exe"
args = [
"-d",
os.environ["WSL_DISTRO_NAME"],
"--",
sys.executable,
"-m",
_CLAUDE_SUBAGENT_MCP_MODULE,
]
mcp_env["WSLENV"] = _merge_wslenv(
os.environ.get("WSLENV", ""),
_wsl_bridge_names(server_env, ()),
)
_write_private_json(
plugin / ".claude-plugin" / "plugin.json",
{
"name": "unsloth-local-agent",
"version": "1.0.0",
"description": _SUBAGENT_DESCRIPTION,
"author": {"name": "Unsloth AI"},
},
)
_write_private_json(
plugin / ".mcp.json",
{
"mcpServers": {
"unsloth": {
"type": "stdio",
"command": command,
"args": args,
"env": mcp_env,
}
}
},
)
skill = plugin / "skills" / "local-agent" / "SKILL.md"
skill.parent.mkdir(parents = True, exist_ok = True, mode = 0o700)
skill.write_text(
"---\n"
"description: Delegate a task to the local agent powered by Unsloth. Use when the "
"user asks to spawn an Unsloth agent or local agent.\n"
"---\n\n"
"Call the Unsloth local agent tool once with the complete task. Return its result "
"to the user without claiming that the cloud parent completed the local work.\n",
encoding = "utf-8",
)
return plugin
def _codex_subagent_flags(path: Path) -> list[str]:
config_path = _agent_config_path(path, ["codex"])
return [
"--enable",
"multi_agent",
"-c",
"agents.max_depth=1",
"-c",
f"agents.{_SUBAGENT_NAME}.description={json.dumps(_SUBAGENT_DESCRIPTION)}",
"-c",
f"agents.{_SUBAGENT_NAME}.config_file={json.dumps(config_path)}",
]
def _wsl_windows_executable(command: list) -> Optional[str]:
if os.name == "nt" or not os.environ.get("WSL_DISTRO_NAME"):
return None
@ -1974,6 +2251,7 @@ def write_opencode_config(
model: dict,
path: Path,
yolo: bool = False,
as_subagent: bool = False,
) -> dict:
config = _read_json_object(path)
if config is None:
@ -1985,10 +2263,8 @@ def write_opencode_config(
return {}
before = json.dumps(config, sort_keys = True)
config.setdefault("$schema", "https://opencode.ai/config.json")
# The session provider is registered under a dedicated id (_OPENCODE_PROVIDER)
# that a user's disabled_providers list would never target, so it is always
# selectable without this overlay having to reconstruct or override OpenCode's
# disabled_providers resolution.
# Keep the provider definition in this private session file. The launch path
# adjusts effective provider filters in the higher-priority inline overlay.
model_entry = {"name": model["id"]}
window = model.get("context_length") or model.get("max_context_length")
if window:
@ -2003,15 +2279,36 @@ def write_opencode_config(
"options": {"baseURL": f"{base}/v1", "apiKey": key},
"models": {model["id"]: model_entry},
}
# OpenCode selects a model by "<providerID>/<modelID>".
config["model"] = f"{_OPENCODE_PROVIDER}/{model['id']}"
if window:
# Normal mode pins this as the session model. Subagent mode leaves the user's
# main/small models alone and exposes the local model to @unsloth and /models.
opencode_model = f"{_OPENCODE_PROVIDER}/{model['id']}"
if as_subagent:
for field in ("model", "small_model"):
if str(config.get(field) or "").startswith(f"{_OPENCODE_PROVIDER}/"):
config.pop(field, None)
managed_compaction = {"auto": True, "reserved": max(1, window // 10)} if window else None
if managed_compaction and config.get("compaction") == managed_compaction:
config.pop("compaction", None)
_subdict(config, "agent")[_SUBAGENT_NAME] = {
"description": _SUBAGENT_DESCRIPTION,
"mode": "subagent",
"model": opencode_model,
"prompt": _SUBAGENT_INSTRUCTIONS,
}
else:
config["model"] = opencode_model
agents = config.get("agent")
if isinstance(agents, dict):
agents.pop(_SUBAGENT_NAME, None)
if not agents:
config.pop("agent", None)
if window and not as_subagent:
# Compact with ~10% headroom (near 90% full). The fixed 20k-token default
# buffer over-compacts, or never settles, on a small local context.
compaction = _subdict(config, "compaction")
compaction["auto"] = True
compaction["reserved"] = max(1, window // 10)
tools = ("edit", "bash", "webfetch")
tools = ("edit", "bash", "webfetch", *(("task",) if as_subagent else ()))
if yolo:
# Fallback for commands without native --auto and for the append-safe bare
# --no-launch command (subcommand unknown yet). Rides inline (OPENCODE_CONFIG_CONTENT)
@ -2140,6 +2437,22 @@ def write_pi_config(base: str, key: str, model: dict, path: Path) -> None:
typer.echo(f"Updated {path}")
def write_pi_subagent_config(base: str, key: str, model: dict, path: Path) -> None:
"""Write private bootstrap data for the bundled Pi extension."""
window = model.get("context_length") or model.get("max_context_length")
window = int(window) if window else 32768
_write_private_json(
path,
{
"baseUrl": f"{base}/v1",
"apiKey": key,
"model": model["id"],
"contextWindow": window,
"maxTokens": min(window // 4, 8192),
},
)
@start_app.command("claude", context_settings = _PASSTHROUGH)
def claude(
ctx: typer.Context,
@ -2153,6 +2466,7 @@ def claude(
serve: bool = _SERVE_OPTION,
yolo: bool = _YOLO_OPTION,
persist: bool = _PERSIST_OPTION,
as_subagent: bool = _AS_SUBAGENT_OPTION,
):
"""Point Claude Code at the running Unsloth server and start it."""
base, key, entry = _connect(
@ -2163,37 +2477,52 @@ def claude(
launch = launch,
)
model_id = entry["id"]
install_hint = (
"irm https://claude.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://claude.ai/install.sh | bash"
)
if as_subagent:
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
window = subagent_model.get("context_length") or subagent_model.get("max_context_length")
server_env = {
"UNSLOTH_CLAUDE_SUBAGENT_BASE_URL": base,
"UNSLOTH_CLAUDE_SUBAGENT_API_KEY": key,
"UNSLOTH_CLAUDE_SUBAGENT_MODEL": subagent_id,
"UNSLOTH_CLAUDE_SUBAGENT_BYPASS_PERMISSIONS": "1" if yolo else "0",
}
if window:
server_env["UNSLOTH_CLAUDE_SUBAGENT_CONTEXT_WINDOW"] = str(int(window))
with _session_config("claude-subagent", launch, persist = persist) as config:
plugin = write_claude_subagent_plugin(config, server_env)
command = [
"claude",
"--plugin-dir",
_agent_config_path(plugin, ["claude"]),
# Before ctx.args: a forwarded `--` would turn later flags positional.
"--allowedTools",
_CLAUDE_SUBAGENT_TOOL,
*_yolo_command_flags("claude", yolo),
*ctx.args,
]
typer.echo(
"Unsloth is available as a local agent. "
"Ask Claude to spawn an Unsloth or local agent."
)
_run(
base,
subagent_model,
{},
command,
launch = launch,
install_hint = install_hint,
)
return
env = {
"ANTHROPIC_BASE_URL": base,
"ANTHROPIC_AUTH_TOKEN": key,
"ANTHROPIC_MODEL": model_id,
# Session-only (no ~/.claude write): suppress the attribution header so
# llama.cpp KV-cache reuse is preserved; --settings below reinforces it.
"CLAUDE_CODE_ATTRIBUTION_HEADER": "0",
# Update checks, beta features, and other background requests either
# stall against a local server or evict the conversation from
# llama-server's KV-cache slots, so turn off everything nonessential.
"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
"CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1",
# A local server streams in bursts; disable the full-screen TUI redraw so the
# terminal doesn't flicker between tokens.
"CLAUDE_CODE_NO_FLICKER": "1",
}
# Claude Code auto-compacts against its native (~600k token) window; a local
# model's context is usually far smaller, so size the window to the loaded
# model's real context length. Otherwise the conversation overflows the
# server's window (silent truncation) long before Claude decides to compact.
# codex/openclaw get the same value through their config (model_context_window
# / contextWindow); Claude has no config file, so it rides on the env var.
window = entry.get("context_length") or entry.get("max_context_length")
if window:
env["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] = str(int(window))
# Compact at 90% of that window; the override only takes effect once the
# window is set, and it can only lower the threshold, so it just guarantees
# headroom before the server's context limit instead of relying on Claude's
# default (which is tuned for its native 200K/1M window).
env["CLAUDE_AUTOCOMPACT_PCT_OVERRIDE"] = "90"
env = _claude_local_env(base, key, entry)
# Claude Code auto-compacts against its native context window. The local env
# above supplies the loaded model's real window and a 90% threshold instead.
# --yolo (or its aliases) maps to Claude's own --dangerously-skip-permissions.
# IS_SANDBOX is left unset on purpose: Claude refuses bypass mode as root unless a
# sandbox is detected, and we don't want to falsely claim one on the user's host.
@ -2208,11 +2537,6 @@ def claude(
*_yolo_command_flags("claude", yolo),
*ctx.args,
]
install_hint = (
"irm https://claude.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://claude.ai/install.sh | bash"
)
_run(
base,
entry,
@ -2237,6 +2561,7 @@ def codex(
serve: bool = _SERVE_OPTION,
yolo: bool = _YOLO_OPTION,
persist: bool = _PERSIST_OPTION,
as_subagent: bool = _AS_SUBAGENT_OPTION,
):
"""Point OpenAI Codex at the running Unsloth server and start it."""
base, key, entry = _connect(
@ -2254,6 +2579,30 @@ def codex(
except BaseException:
_shutdown_auto_served()
raise
if as_subagent:
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
with _session_config("codex-subagent", launch, persist = persist) as home:
agent_config = write_codex_subagent_config(base, key, subagent_model, home)
command = [
"codex",
*_codex_subagent_flags(agent_config),
*_yolo_command_flags("codex", yolo),
*ctx.args,
]
typer.echo(
"Unsloth is available as the `unsloth` local agent. "
"Ask Codex to spawn an Unsloth or local agent."
)
_run(
base,
subagent_model,
{},
command,
launch = launch,
install_hint = "npm install -g @openai/codex",
)
return
command = [
"codex",
"--oss",
@ -2283,6 +2632,7 @@ def openclaw(
persist: bool = _PERSIST_OPTION,
):
"""Point OpenClaw at the running Unsloth server and start it."""
_reject_as_subagent("openclaw", ctx.args)
base, key, entry = _connect(
api_key,
model,
@ -2338,6 +2688,7 @@ def opencode(
serve: bool = _SERVE_OPTION,
yolo: bool = _YOLO_OPTION,
persist: bool = _PERSIST_OPTION,
as_subagent: bool = _AS_SUBAGENT_OPTION,
):
"""Point OpenCode at the running Unsloth server and start it."""
base, key, entry = _connect(
@ -2347,6 +2698,50 @@ def opencode(
serve = serve,
launch = launch,
)
if as_subagent:
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
# Stay append-safe for a bare no-launch recipe: a later `run <prompt>` would make
# `opencode --auto run ...` parse as the TUI, so keep yolo in the inline fallback.
route_native_auto = yolo and _opencode_supports_native_auto() and (launch or bool(ctx.args))
opencode_args, native_auto = _opencode_native_auto_args(list(ctx.args), route_native_auto)
command = ["opencode", *opencode_args]
with _session_config("opencode-subagent", launch, persist = persist) as cfg:
config_path = cfg / "opencode.json"
session_permission = write_opencode_config(
base,
key,
subagent_model,
config_path,
yolo = yolo and not native_auto,
as_subagent = True,
)
env = {"OPENCODE_CONFIG": str(config_path)}
if launch and _which_with_install_dirs("opencode") is None:
# Provider-filter inspection needs the binary; offer the install now so
# a global/project allowlist is honored on this first launch instead of
# being read only after _launch installs OpenCode.
_install_agent("opencode", "npm install -g opencode-ai")
inline_config = _opencode_subagent_inline_config(config_path, session_permission)
# A project opencode.json outranks the session file and could field-merge its
# own agent.unsloth over ours. Pin ours in the inline overlay so it wins.
inline_config.setdefault("agent", {})[_SUBAGENT_NAME] = {
"description": _SUBAGENT_DESCRIPTION,
"mode": "subagent",
"model": f"{_OPENCODE_PROVIDER}/{subagent_model['id']}",
"prompt": _SUBAGENT_INSTRUCTIONS,
}
env["OPENCODE_CONFIG_CONTENT"] = json.dumps(inline_config)
typer.echo("Unsloth is available as @unsloth and in /models.")
_run(
base,
subagent_model,
env,
command,
launch = launch,
install_hint = "npm install -g opencode-ai",
)
return
opencode_model = f"{_OPENCODE_PROVIDER}/{entry['id']}"
# The inline OPENCODE_CONFIG_CONTENT below pins the model in the highest-priority
# layer, so the session model is forced without a --model flag. Only add --model for
@ -2433,6 +2828,7 @@ def hermes(
persist: bool = _PERSIST_OPTION,
):
"""Point Hermes (Nous Research) at the running Unsloth server and start it."""
_reject_as_subagent("hermes", ctx.args)
native_args = [*_yolo_command_flags("hermes", yolo), *ctx.args]
command = ["hermes", *_hermes_resume_oneshot_args(native_args)]
base, key, entry = _connect(
@ -2464,6 +2860,7 @@ def pi(
serve: bool = _SERVE_OPTION,
yolo: bool = _YOLO_OPTION,
persist: bool = _PERSIST_OPTION,
as_subagent: bool = _AS_SUBAGENT_OPTION,
):
"""Point Pi (coding agent) at the running Unsloth server and start it."""
base, key, entry = _connect(
@ -2473,6 +2870,37 @@ def pi(
serve = serve,
launch = launch,
)
install_hint = "npm install -g --ignore-scripts @earendil-works/pi-coding-agent"
if as_subagent:
if not _PI_SUBAGENT_EXTENSION.is_file():
_fail(f"Missing Pi subagent extension: {_PI_SUBAGENT_EXTENSION}")
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
extension = _agent_config_path(_PI_SUBAGENT_EXTENSION, ["pi"])
with _session_config("pi-subagent", launch, persist = persist) as config:
config_path = config / "subagent.json"
write_pi_subagent_config(base, key, subagent_model, config_path)
command = [
"pi",
"--extension",
extension,
*_yolo_command_flags("pi", yolo),
*ctx.args,
]
typer.echo(
"Unsloth is available as a local agent and in /model. "
"Ask Pi to spawn an Unsloth or local agent."
)
_run(
base,
subagent_model,
{"UNSLOTH_PI_SUBAGENT_CONFIG": str(config_path)},
command,
launch = launch,
install_hint = install_hint,
clear_screen = True,
)
return
# Pi defaults to the google provider, so pin our provider/model on the command
# line; the custom OpenAI-compatible endpoint itself is only configurable via
# ~/.pi/agent/models.json.
@ -2487,7 +2915,6 @@ def pi(
]
# --ignore-scripts matches Pi's documented install recipe (its README notes Pi needs
# no install scripts), so accepting the prompt skips dependency lifecycle scripts.
install_hint = "npm install -g --ignore-scripts @earendil-works/pi-coding-agent"
with _session_config("pi", launch, persist = persist) as home:
# Pi resolves its config dir from PI_CODING_AGENT_DIR first (getAgentDir() prefers
# it over $HOME/.pi/agent), so pin it at the session dir: an inherited

241
unsloth_cli/pi_subagent.ts Normal file
View file

@ -0,0 +1,241 @@
import { spawn, type ChildProcess } from "node:child_process";
import * as fs from "node:fs";
import * as path from "node:path";
import { fileURLToPath } from "node:url";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { Type } from "typebox";
const provider = "unsloth";
const maxResultCharacters = 100_000;
const cancelGraceMilliseconds = 2_000;
const configPath = process.env.UNSLOTH_PI_SUBAGENT_CONFIG || "";
delete process.env.UNSLOTH_PI_SUBAGENT_CONFIG;
let config: Record<string, unknown> = {};
if (configPath) {
try {
const parsed = JSON.parse(fs.readFileSync(configPath, "utf8"));
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
throw new Error("expected a JSON object");
}
config = parsed;
} catch (error) {
throw new Error(`Could not read Unsloth subagent configuration: ${error}`);
}
}
const model = typeof config.model === "string" ? config.model : "";
const baseUrl = typeof config.baseUrl === "string" ? config.baseUrl : "";
const apiKey = typeof config.apiKey === "string" ? config.apiKey : "";
const contextWindow = positiveInt(config.contextWindow, 32768);
const maxTokens = positiveInt(config.maxTokens, Math.min(Math.floor(contextWindow / 4), 8192));
function positiveInt(value: unknown, fallback: number): number {
const parsed = Number.parseInt(typeof value === "string" ? value : String(value || ""), 10);
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
}
function finalText(message: any): string {
if (message?.role !== "assistant" || !Array.isArray(message.content)) return "";
return message.content
.filter((part: any) => part?.type === "text" && typeof part.text === "string")
.map((part: any) => part.text)
.join("\n")
.trim();
}
function boundedResult(text: string): string {
if (text.length <= maxResultCharacters) return text;
return `${text.slice(0, maxResultCharacters)}\n\n[Local agent output truncated]`;
}
function piInvocation(args: string[]): { command: string; args: string[] } {
const currentScript = process.argv[1];
const bunVirtualScript = currentScript?.startsWith("/$bunfs/root/");
if (currentScript && !bunVirtualScript && fs.existsSync(currentScript)) {
return { command: process.execPath, args: [currentScript, ...args] };
}
const executable = path.basename(process.execPath).toLowerCase();
if (!/^(node|bun)(\.exe)?$/.test(executable)) return { command: process.execPath, args };
return { command: "pi", args };
}
function signalProcessGroup(child: ChildProcess, signal: NodeJS.Signals): void {
if (!child.pid) return;
try {
process.kill(-child.pid, signal);
} catch {
try {
child.kill(signal);
} catch {
// The process tree already exited.
}
}
}
async function stopChildTree(child: ChildProcess): Promise<void> {
if (!child.pid) return;
if (process.platform === "win32") {
await new Promise<void>((resolve) => {
const killer = spawn("taskkill", ["/PID", String(child.pid), "/T", "/F"], {
shell: false,
stdio: "ignore",
windowsHide: true,
});
killer.once("error", () => {
try {
child.kill("SIGKILL");
} catch {
// The child already exited.
}
resolve();
});
killer.once("close", (code) => {
if (code !== 0) {
try {
child.kill("SIGKILL");
} catch {
// The child already exited.
}
}
resolve();
});
});
return;
}
signalProcessGroup(child, "SIGTERM");
await new Promise((resolve) => setTimeout(resolve, cancelGraceMilliseconds));
signalProcessGroup(child, "SIGKILL");
}
export default function unslothSubagent(pi: ExtensionAPI): void {
if (!model || !baseUrl || !apiKey || !configPath) {
throw new Error("Unsloth subagent configuration is incomplete.");
}
pi.registerProvider(provider, {
name: "Unsloth Studio",
baseUrl,
apiKey,
api: "openai-completions",
authHeader: true,
models: [
{
id: model,
name: `${model} via Unsloth`,
reasoning: false,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow,
maxTokens,
},
],
});
if (process.env.UNSLOTH_PI_SUBAGENT_CHILD === "1") return;
pi.registerTool({
name: "unsloth_agent",
label: "Unsloth agent",
description:
"Local coding subagent powered by Unsloth for debugging, implementation, and codebase research. Use when the user asks to spawn an Unsloth or local agent.",
parameters: Type.Object({
task: Type.String({ description: "The complete task for the local Unsloth agent." }),
}),
async execute(_toolCallId, params, signal, _onUpdate, ctx) {
const extension = fileURLToPath(import.meta.url);
const args = [
"--mode",
"json",
"--print",
"--no-session",
"--provider",
provider,
"--model",
model,
"--no-extensions",
"--extension",
extension,
`Task: ${params.task}`,
];
const invocation = piInvocation(args);
let output = "";
let stderr = "";
let lastResponse = "";
let childError = "";
let aborted = false;
const processLine = (line: string) => {
try {
const event = JSON.parse(line);
if (event.type !== "message_end") return;
const message = event.message;
// Pi reports model/API failures as message_end events while still
// exiting 0, so the exit status alone cannot surface them.
if (message?.stopReason === "error" || message?.stopReason === "aborted") {
childError =
(typeof message.errorMessage === "string" && message.errorMessage) ||
`The local Unsloth agent stopped: ${message.stopReason}.`;
return;
}
const response = finalText(message);
if (response) {
lastResponse = boundedResult(response);
childError = "";
}
} catch {
// Ignore non-JSON diagnostic lines. The exit status still reports failures.
}
};
const exitCode = await new Promise<number>((resolve, reject) => {
const child = spawn(invocation.command, invocation.args, {
cwd: ctx.cwd,
detached: process.platform !== "win32",
shell: false,
stdio: ["ignore", "pipe", "pipe"],
env: {
...process.env,
UNSLOTH_PI_SUBAGENT_CHILD: "1",
UNSLOTH_PI_SUBAGENT_CONFIG: configPath,
},
});
let cleanup: Promise<void> | undefined;
const cancel = () => {
if (aborted) return;
aborted = true;
cleanup = stopChildTree(child);
};
child.on("error", (error) => {
signal?.removeEventListener("abort", cancel);
reject(error);
});
child.stdout.on("data", (chunk) => {
output += chunk.toString();
const lines = output.split("\n");
output = lines.pop() || "";
for (const line of lines) processLine(line);
});
child.stderr.on("data", (chunk) => {
stderr = (stderr + chunk.toString()).slice(-100_000);
});
child.on("close", async (code) => {
signal?.removeEventListener("abort", cancel);
await cleanup;
if (output.trim()) processLine(output);
resolve(code ?? 1);
});
signal?.addEventListener("abort", cancel, { once: true });
if (signal?.aborted) cancel();
});
if (aborted) throw new Error("The local Unsloth agent was cancelled.");
if (exitCode !== 0) {
throw new Error(stderr.trim() || `The local Unsloth agent exited with code ${exitCode}.`);
}
if (childError) throw new Error(boundedResult(childError));
return {
content: [{ type: "text", text: lastResponse || "The local agent returned no text." }],
details: { provider, model },
};
},
});
}

338
unsloth_cli/tests/test_claude_subagent_mcp.py Normal file
View file

@ -0,0 +1,338 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
from __future__ import annotations
import io
import json
import os
import subprocess
import sys
import time
import pytest
import unsloth_cli.claude_subagent_mcp as bridge
def test_protocol_lists_and_calls_local_agent():
initialized = bridge._response(
{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
)
assert initialized["result"]["serverInfo"]["name"] == "unsloth-local-agent"
listed = bridge._response({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
tool = listed["result"]["tools"][0]
assert tool["name"] == "unsloth_agent"
assert "spawn an Unsloth or local agent" in tool["description"]
assert tool["inputSchema"]["required"] == ["task"]
assert tool["_meta"]["anthropic/maxResultSizeChars"] == 100_000
called = bridge._response(
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": "unsloth_agent", "arguments": {"task": " inspect this "}},
},
run_agent = lambda task: f"completed: {task}",
)
assert called["result"] == {
"content": [{"type": "text", "text": "completed: inspect this"}],
"isError": False,
}
def test_protocol_returns_tool_errors_to_parent():
response = bridge._response(
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": "unsloth_agent", "arguments": {"task": "test"}},
},
run_agent = lambda task: (_ for _ in ()).throw(RuntimeError("local failure")),
)
assert response["result"]["isError"] is True
assert response["result"]["content"][0]["text"] == "local failure"
def test_stdio_server_ignores_notifications_and_answers_requests():
requests = "\n".join(
[
json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}),
json.dumps({"jsonrpc": "2.0", "id": 4, "method": "ping"}),
]
)
output = io.StringIO()
bridge.serve(io.StringIO(requests), output)
assert json.loads(output.getvalue()) == {"jsonrpc": "2.0", "id": 4, "result": {}}
def test_stdio_cancellation_reaches_the_running_local_agent():
requests = "\n".join(
[
json.dumps(
{
"jsonrpc": "2.0",
"id": "call-1",
"method": "tools/call",
"params": {"name": "unsloth_agent", "arguments": {"task": "wait"}},
}
),
json.dumps(
{
"jsonrpc": "2.0",
"method": "notifications/cancelled",
"params": {"requestId": "call-1", "reason": "user cancelled"},
}
),
]
)
output = io.StringIO()
cancelled = []
def run_agent(task, cancel_event):
assert task == "wait"
assert cancel_event.wait(timeout = 1)
cancelled.append(task)
raise RuntimeError("The local Claude agent was cancelled.")
bridge.serve(io.StringIO(requests), output, run_agent = run_agent)
assert cancelled == ["wait"]
assert output.getvalue() == ""
def test_stdio_sigint_stops_the_running_local_agent(monkeypatch):
request = json.dumps(
{
"jsonrpc": "2.0",
"id": "call-1",
"method": "tools/call",
"params": {"name": "unsloth_agent", "arguments": {"task": "wait"}},
}
)
handlers = {}
started = bridge.threading.Event()
cancelled = []
def set_handler(signum, handler):
previous = handlers.get(signum, bridge.signal.SIG_DFL)
handlers[signum] = handler
return previous
monkeypatch.setattr(bridge.signal, "signal", set_handler)
class InterruptingInput:
def __init__(self):
self.sent = False
def __iter__(self):
return self
def __next__(self):
if not self.sent:
self.sent = True
return request + "\n"
assert started.wait(timeout = 1)
handlers[bridge.signal.SIGINT](bridge.signal.SIGINT, None)
raise AssertionError("SIGINT handler must unwind the stdin loop")
def run_agent(task, cancel_event):
assert task == "wait"
started.set()
assert cancel_event.wait(timeout = 1)
# Real Claude Code sends SIGINT twice. The second one must not abort cleanup.
handlers[bridge.signal.SIGINT](bridge.signal.SIGINT, None)
cancelled.append(task)
raise RuntimeError("The local Claude agent was cancelled.")
output = io.StringIO()
bridge.serve(InterruptingInput(), output, run_agent = run_agent)
assert cancelled == ["wait"]
assert output.getvalue() == ""
@pytest.mark.parametrize(
("bypass", "permission"),
[("0", "acceptEdits"), ("1", "bypassPermissions")],
)
def test_local_child_uses_unsloth_without_overwriting_parent_auth(
monkeypatch, tmp_path, bypass, permission
):
captured = {}
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_BASE_URL", "http://127.0.0.1:8888")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_API_KEY", "sk-unsloth-test")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_MODEL", "unsloth/model-GGUF:Q4_K_M")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_CONTEXT_WINDOW", "32768")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_BYPASS_PERMISSIONS", bypass)
monkeypatch.setenv("ANTHROPIC_API_KEY", "cloud-key")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "cloud-oauth")
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path))
monkeypatch.setattr(bridge.shutil, "which", lambda _: "/usr/local/bin/claude")
monkeypatch.setattr(bridge, "_claude_flags", lambda model: ["--settings", "{}"])
class Process:
pid = 1234
returncode = 0
def communicate(self, timeout):
captured["timeout"] = timeout
return json.dumps({"is_error": False, "result": "LOCAL_OK"}), ""
def poll(self):
return self.returncode
def popen(command, **kwargs):
captured["command"] = command
captured.update(kwargs)
return Process()
monkeypatch.setattr(bridge.subprocess, "Popen", popen)
assert bridge.run_local_agent("reply exactly LOCAL_OK") == "LOCAL_OK"
command = captured["command"]
assert command[:3] == ["/usr/local/bin/claude", "--model", "unsloth/model-GGUF:Q4_K_M"]
assert command[command.index("--permission-mode") + 1] == permission
assert "--no-session-persistence" in command
assert captured["cwd"] == str(tmp_path)
assert captured["stdin"] is bridge.subprocess.DEVNULL
assert captured["stdout"] is bridge.subprocess.PIPE
assert captured["stderr"] is bridge.subprocess.PIPE
if os.name == "nt":
assert captured["creationflags"] == bridge.subprocess.CREATE_NEW_PROCESS_GROUP
else:
assert captured["start_new_session"] is True
child_env = captured["env"]
assert child_env["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:8888"
assert child_env["ANTHROPIC_AUTH_TOKEN"] == "sk-unsloth-test"
assert child_env["ANTHROPIC_MODEL"] == "unsloth/model-GGUF:Q4_K_M"
assert child_env["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] == "32768"
assert child_env["CLAUDE_AUTOCOMPACT_PCT_OVERRIDE"] == "90"
assert "ANTHROPIC_API_KEY" not in child_env
assert "CLAUDE_CODE_OAUTH_TOKEN" not in child_env
def test_local_child_process_is_stopped_on_cancellation(monkeypatch, tmp_path):
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_BASE_URL", "http://127.0.0.1:8888")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_API_KEY", "sk-unsloth-test")
monkeypatch.setenv("UNSLOTH_CLAUDE_SUBAGENT_MODEL", "unsloth/model-GGUF:Q4_K_M")
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path))
monkeypatch.setattr(bridge.shutil, "which", lambda _: "/usr/local/bin/claude")
monkeypatch.setattr(bridge, "_claude_flags", lambda model: [])
cancel_event = bridge.threading.Event()
stopped = []
class Process:
pid = 1234
returncode = None
def communicate(self, timeout):
cancel_event.set()
raise bridge.subprocess.TimeoutExpired("claude", timeout)
def poll(self):
return self.returncode
process = Process()
monkeypatch.setattr(bridge.subprocess, "Popen", lambda *args, **kwargs: process)
def stop(child):
stopped.append(child)
child.returncode = -15
monkeypatch.setattr(bridge, "_stop_child", stop)
with pytest.raises(RuntimeError, match = "cancelled"):
bridge.run_local_agent("wait", cancel_event)
assert stopped == [process]
def test_windows_cancellation_stops_the_child_process_tree(monkeypatch):
monkeypatch.setattr(bridge.os, "name", "nt")
captured = {}
class Process:
pid = 4321
returncode = None
def poll(self):
return self.returncode
def wait(self, timeout = None):
captured["wait_timeout"] = timeout
self.returncode = 1
def terminate(self):
raise AssertionError("taskkill should handle the process tree")
def run(command, **kwargs):
captured["command"] = command
captured.update(kwargs)
return bridge.subprocess.CompletedProcess(command, 0)
monkeypatch.setattr(bridge.subprocess, "run", run)
bridge._stop_child(Process())
assert captured["command"] == ["taskkill", "/PID", "4321", "/T", "/F"]
assert captured["capture_output"] is True
assert captured["check"] is False
assert captured["wait_timeout"] == bridge._CANCEL_GRACE_SECONDS
def test_windows_failed_taskkill_still_terminates_the_child(monkeypatch):
monkeypatch.setattr(bridge.os, "name", "nt")
captured = {}
class Process:
pid = 4321
returncode = None
def poll(self):
return self.returncode
def wait(self, timeout = None):
self.returncode = 1
def terminate(self):
captured["terminated"] = True
self.returncode = 1
monkeypatch.setattr(
bridge.subprocess,
"run",
lambda command, **kwargs: bridge.subprocess.CompletedProcess(command, 1),
)
bridge._stop_child(Process())
assert captured.get("terminated") is True
@pytest.mark.skipif(os.name == "nt", reason = "POSIX process groups")
def test_stop_child_kills_survivors_after_leader_exit(monkeypatch, tmp_path):
monkeypatch.setattr(bridge, "_CANCEL_GRACE_SECONDS", 0.2)
marker = tmp_path / "grandchild-survived"
grandchild = (
"import pathlib, sys, time; time.sleep(1.0); "
"pathlib.Path(sys.argv[1]).write_text('alive')"
)
process = subprocess.Popen(
[
sys.executable,
"-c",
"import subprocess, sys; "
"subprocess.Popen([sys.executable, '-c', sys.argv[1], sys.argv[2]])",
grandchild,
str(marker),
],
start_new_session = True,
)
process.wait()
bridge._stop_child(process)
time.sleep(1.2)
assert not marker.exists()
def test_result_parser_accepts_diagnostics_before_json():
output = "connector warning\n" + json.dumps({"is_error": False, "result": "OK"})
assert bridge._result_text(output) == "OK"

191
unsloth_cli/tests/test_pi_subagent.py Normal file
View file

@ -0,0 +1,191 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
import os
from pathlib import Path
import json
import shutil
import subprocess
import pytest
@pytest.mark.skipif(os.name == "nt", reason = "POSIX process group regression test")
def test_pi_cancel_kills_child_process_group(tmp_path):
bun = shutil.which("bun")
if bun is None:
pytest.skip("Bun is required to execute the bundled Pi extension")
ready = tmp_path / "grandchild-ready"
marker = tmp_path / "grandchild-survived"
config = tmp_path / "subagent.json"
config.write_text(
json.dumps(
{
"baseUrl": "http://127.0.0.1:8000/v1",
"apiKey": "private-token",
"model": "local-model",
"contextWindow": 32768,
"maxTokens": 8192,
}
),
encoding = "utf-8",
)
driver = tmp_path / "pi-driver.js"
driver.write_text(
"""
import { spawn } from "node:child_process";
spawn(
process.execPath,
[
"-e",
`
const fs = require("node:fs");
process.on("SIGTERM", () => {});
fs.writeFileSync(process.env.PI_CHILD_READY, "ready");
setTimeout(() => fs.writeFileSync(process.env.PI_CANCEL_MARKER, "alive"), 3000);
setInterval(() => {}, 1000);
`,
],
{ stdio: "inherit" },
);
process.on("SIGTERM", () => {});
setInterval(() => {}, 1000);
""",
encoding = "utf-8",
)
extension = Path(__file__).parents[1] / "pi_subagent.ts"
test_file = tmp_path / "pi-cancel.test.ts"
test_file.write_text(
f"""
import {{ expect, mock, test }} from "bun:test";
import {{ existsSync }} from "node:fs";
import {{ pathToFileURL }} from "node:url";
mock.module("typebox", () => ({{
Type: {{ Object: (value) => value, String: (value) => value }},
}}));
test("cancellation stops the Pi child process group", async () => {{
process.env.UNSLOTH_PI_SUBAGENT_CONFIG = {str(config)!r};
process.env.PI_CHILD_READY = {str(ready)!r};
process.env.PI_CANCEL_MARKER = {str(marker)!r};
process.argv[1] = {str(driver)!r};
const loaded = await import(pathToFileURL({str(extension)!r}).href);
let tool;
let provider;
loaded.default({{
registerProvider(_name, value) {{ provider = value; }},
registerTool(value) {{ tool = value; }},
}});
expect(process.env.UNSLOTH_PI_SUBAGENT_CONFIG).toBeUndefined();
expect(process.env.UNSLOTH_PI_SUBAGENT_API_KEY).toBeUndefined();
expect(provider.apiKey).toBe("private-token");
const controller = new AbortController();
const execution = tool.execute(
"call",
{{ task: "wait" }},
controller.signal,
undefined,
{{ cwd: {str(tmp_path)!r} }},
);
for (let attempt = 0; attempt < 100 && !existsSync({str(ready)!r}); attempt++) {{
await Bun.sleep(20);
}}
expect(existsSync({str(ready)!r})).toBe(true);
controller.abort();
await expect(execution).rejects.toThrow("cancelled");
await Bun.sleep(3200);
expect(existsSync({str(marker)!r})).toBe(false);
}}, 10_000);
""",
encoding = "utf-8",
)
completed = subprocess.run(
[bun, "test", str(test_file)],
capture_output = True,
text = True,
timeout = 15,
)
assert completed.returncode == 0, completed.stdout + completed.stderr
@pytest.mark.skipif(os.name == "nt", reason = "POSIX driver script")
def test_pi_child_error_events_fail_the_tool_call(tmp_path):
bun = shutil.which("bun")
if bun is None:
pytest.skip("Bun is required to execute the bundled Pi extension")
config = tmp_path / "subagent.json"
config.write_text(
json.dumps(
{
"baseUrl": "http://127.0.0.1:8000/v1",
"apiKey": "private-token",
"model": "local-model",
"contextWindow": 32768,
"maxTokens": 8192,
}
),
encoding = "utf-8",
)
# Pi reports model/API failures as message_end events while exiting 0.
driver = tmp_path / "pi-driver.js"
driver.write_text(
"""
const event = {
type: "message_end",
message: { role: "assistant", stopReason: "error", errorMessage: "backend unreachable", content: [] },
};
console.log(JSON.stringify(event));
""",
encoding = "utf-8",
)
extension = Path(__file__).parents[1] / "pi_subagent.ts"
test_file = tmp_path / "pi-error.test.ts"
test_file.write_text(
f"""
import {{ expect, mock, test }} from "bun:test";
import {{ pathToFileURL }} from "node:url";
mock.module("typebox", () => ({{
Type: {{ Object: (value) => value, String: (value) => value }},
}}));
test("child error events fail the tool call", async () => {{
process.env.UNSLOTH_PI_SUBAGENT_CONFIG = {str(config)!r};
process.argv[1] = {str(driver)!r};
const loaded = await import(pathToFileURL({str(extension)!r}).href);
let tool;
loaded.default({{
registerProvider() {{}},
registerTool(value) {{ tool = value; }},
}});
const execution = tool.execute(
"call",
{{ task: "fail" }},
undefined,
undefined,
{{ cwd: {str(tmp_path)!r} }},
);
await expect(execution).rejects.toThrow("backend unreachable");
}}, 10_000);
""",
encoding = "utf-8",
)
completed = subprocess.run(
[bun, "test", str(test_file)],
capture_output = True,
text = True,
timeout = 15,
)
assert completed.returncode == 0, completed.stdout + completed.stderr

490
unsloth_cli/tests/test_start.py
View file

@ -619,6 +619,104 @@ def test_write_codex_config_omits_catalog_for_old_codex(tmp_path, monkeypatch):
assert not (tmp_path / "model-catalog.json").exists()
def test_write_codex_subagent_config_keeps_parent_model_out(tmp_path, monkeypatch):
monkeypatch.setattr(start, "_codex_supports_model_catalog", lambda: True)
local = {**MODEL, "id": MODEL["id"] + ":UD-Q4_K_XL"}
path = start.write_codex_subagent_config(BASE, "private-token", local, tmp_path)
agent = _parse_toml(path.read_text())
assert agent["name"] == "unsloth"
assert "local agent" in agent["description"].lower()
assert agent["model_provider"] == start._CODEX_PROFILE
assert agent["model"] == local["id"]
assert agent["model_context_window"] == MODEL["context_length"]
assert agent["model_providers"][start._CODEX_PROFILE] == {
"name": "Unsloth Studio",
"base_url": f"{BASE}/v1",
"wire_api": "responses",
"auth": {
"command": sys.executable,
"args": [
"-c",
"import json,sys; print(json.load(open(sys.argv[1], encoding='utf-8'))['token'])",
str(tmp_path / "unsloth-auth.json"),
],
"timeout_ms": 5000,
},
}
assert json.loads((tmp_path / "unsloth-auth.json").read_text()) == {"token": "private-token"}
catalog = json.loads((tmp_path / agent["model_catalog_json"]).read_text())
assert catalog["models"][0]["slug"] == local["id"]
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
def test_codex_subagent_auth_uses_wsl_for_windows_codex(monkeypatch, tmp_path):
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setattr(start, "_codex_supports_model_catalog", lambda: False)
monkeypatch.setattr(
start.shutil,
"which",
lambda _: "/mnt/c/Users/x/AppData/Roaming/npm/codex.exe",
)
path = start.write_codex_subagent_config(BASE, "private-token", MODEL, tmp_path)
auth = _parse_toml(path.read_text())["model_providers"][start._CODEX_PROFILE]["auth"]
assert auth["command"] == "wsl.exe"
assert auth["args"][:5] == ["-d", "Ubuntu", "--", sys.executable, "-c"]
assert auth["args"][-1] == str(tmp_path / "unsloth-auth.json")
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
def test_agent_config_path_translates_for_windows_agent(monkeypatch, tmp_path):
windows_path = r"\\wsl.localhost\Ubuntu\tmp\unsloth.toml"
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setattr(
start.shutil,
"which",
lambda _: "/mnt/c/Users/x/AppData/Roaming/npm/codex",
)
monkeypatch.setattr(start.subprocess, "check_output", lambda *args, **kwargs: windows_path)
assert start._agent_config_path(tmp_path / "unsloth.toml", ["codex"]) == windows_path
def test_subagent_model_id_preserves_explicit_variant(monkeypatch):
monkeypatch.setattr(
start,
"_http_json",
lambda *args, **kwargs: pytest.fail("explicit variant should not need status"),
)
assert (
start._subagent_model_id(BASE, "key", MODEL, MODEL["id"], "UD-Q4_K_XL")
== MODEL["id"] + ":UD-Q4_K_XL"
)
def test_subagent_model_id_uses_loaded_variant(monkeypatch):
monkeypatch.setattr(
start,
"_http_json",
lambda *args, **kwargs: {"is_gguf": True, "gguf_variant": "Q5_K_M"},
)
assert start._subagent_model_id(BASE, "key", MODEL, None, None) == MODEL["id"] + ":Q5_K_M"
def test_subagent_model_id_warns_when_status_unavailable(monkeypatch, capsys):
def raise_error(*args, **kwargs):
raise OSError("connection refused")
monkeypatch.setattr(start, "_http_json", raise_error)
assert start._subagent_model_id(BASE, "key", MODEL, None, None) == MODEL["id"]
assert "could not verify the loaded GGUF variant" in capsys.readouterr().err
@pytest.mark.parametrize("agent", ["openclaw", "hermes"])
def test_unsupported_agents_reject_as_subagent(agent):
result = CliRunner().invoke(start.start_app, [agent, "--as-subagent"])
assert result.exit_code == 1
assert f"--as-subagent is not supported for {agent}." in result.output
@pytest.fixture()
def fake_studio(tmp_path, monkeypatch):
calls = []
@ -690,6 +788,82 @@ def test_connect_claude_no_launch(fake_studio):
assert ".claude/settings.json" not in result.output
def test_connect_claude_as_subagent_preserves_cloud_parent(fake_studio, tmp_path):
result = CliRunner().invoke(
start.start_app,
[
"claude",
"--as-subagent",
"--no-launch",
"--model",
MODEL["id"] + ":UD-Q4_K_XL",
"hello",
],
)
assert result.exit_code == 0, result.output
command = _launch_command(result.output)
plugin = tmp_path / "agents" / "claude-subagent" / "unsloth-local-agent"
assert command == [
"claude",
"--plugin-dir",
str(plugin),
"--allowedTools",
start._CLAUDE_SUBAGENT_TOOL,
"hello",
]
assert "--model" not in command
parent_base = "$env:ANTHROPIC_BASE_URL" if os.name == "nt" else "export ANTHROPIC_BASE_URL="
parent_token = (
"$env:ANTHROPIC_AUTH_TOKEN" if os.name == "nt" else "export ANTHROPIC_AUTH_TOKEN="
)
assert parent_base not in result.output
assert parent_token not in result.output
assert "unset ANTHROPIC_API_KEY" not in result.output
assert "UNSLOTH_CLAUDE_SUBAGENT_API_KEY" not in result.output
assert "sk-unsloth-feedfacefeedface" not in result.output
assert json.loads((plugin / ".claude-plugin" / "plugin.json").read_text())["name"] == (
"unsloth-local-agent"
)
mcp = json.loads((plugin / ".mcp.json").read_text())["mcpServers"]["unsloth"]
assert mcp["command"] == sys.executable
assert mcp["args"] == ["-m", start._CLAUDE_SUBAGENT_MCP_MODULE]
assert mcp["env"] == {
"UNSLOTH_CLAUDE_SUBAGENT_BASE_URL": BASE,
"UNSLOTH_CLAUDE_SUBAGENT_API_KEY": "sk-unsloth-feedfacefeedface",
"UNSLOTH_CLAUDE_SUBAGENT_MODEL": MODEL["id"] + ":UD-Q4_K_XL",
"UNSLOTH_CLAUDE_SUBAGENT_BYPASS_PERMISSIONS": "0",
"UNSLOTH_CLAUDE_SUBAGENT_CONTEXT_WINDOW": "4096",
}
skill = (plugin / "skills" / "local-agent" / "SKILL.md").read_text()
assert "spawn an Unsloth agent or local agent" in skill
assert "Ask Claude to spawn an Unsloth or local agent." in result.output
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
def test_claude_subagent_plugin_uses_wsl_for_windows_claude(monkeypatch, tmp_path):
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setenv("WSLENV", "EXISTING")
monkeypatch.setattr(
start.shutil,
"which",
lambda _: "/mnt/c/Users/x/AppData/Local/Programs/claude.exe",
)
server_env = {"UNSLOTH_CLAUDE_SUBAGENT_API_KEY": "secret"}
plugin = start.write_claude_subagent_plugin(tmp_path, server_env)
mcp = json.loads((plugin / ".mcp.json").read_text())["mcpServers"]["unsloth"]
assert mcp["command"] == "wsl.exe"
assert mcp["args"] == [
"-d",
"Ubuntu",
"--",
sys.executable,
"-m",
start._CLAUDE_SUBAGENT_MCP_MODULE,
]
assert mcp["env"]["UNSLOTH_CLAUDE_SUBAGENT_API_KEY"] == "secret"
assert mcp["env"]["WSLENV"].split(":") == ["EXISTING", "UNSLOTH_CLAUDE_SUBAGENT_API_KEY"]
def test_connect_claude_compact_window_omitted_without_context(fake_studio, monkeypatch):
# A model that doesn't report a context length -> leave Claude's default window
# rather than guessing one.
@ -814,6 +988,38 @@ def test_connect_codex_no_launch(fake_studio, tmp_path):
assert (home / "unsloth_api.config.toml").exists()
def test_connect_codex_as_subagent_preserves_cloud_parent(fake_studio, tmp_path, monkeypatch):
monkeypatch.setattr(start, "_codex_supports_model_catalog", lambda: True)
result = CliRunner().invoke(
start.start_app,
[
"codex",
"--as-subagent",
"--no-launch",
"--model",
MODEL["id"] + ":UD-Q4_K_XL",
],
)
assert result.exit_code == 0, result.output
command = _launch_command(result.output)
assert command[0] == "codex"
assert command[1:3] == ["--enable", "multi_agent"]
assert "agents.max_depth=1" in command
assert "--oss" not in command
assert "--profile" not in command
assert "--model" not in command
assert "CODEX_HOME" not in result.output
assert start._CODEX_ENV_KEY not in result.output
assert "sk-unsloth-feedfacefeedface" not in result.output
home = tmp_path / "agents" / "codex-subagent"
agent_path = home / "unsloth.toml"
agent = _parse_toml(agent_path.read_text())
assert agent["model"] == MODEL["id"] + ":UD-Q4_K_XL"
assert "env_key" not in agent["model_providers"][start._CODEX_PROFILE]
assert f"agents.unsloth.config_file={json.dumps(str(agent_path))}" in command
assert "Ask Codex to spawn an Unsloth or local agent." in result.output
def test_connect_codex_matches_requested_model_case_insensitively(fake_studio, tmp_path):
result = CliRunner().invoke(
start.start_app,
@ -2467,8 +2673,7 @@ def test_write_opencode_config_fresh(tmp_path):
MODEL["id"]: {"name": MODEL["id"], "limit": {"context": 131072, "output": 8192}}
}
assert config["model"] == f"{start._OPENCODE_PROVIDER}/{MODEL['id']}"
# The overlay never writes disabled_providers; the dedicated provider id is one a
# user's disable list would not target, so nothing needs re-enabling.
# Provider filters belong to the launch-time inline overlay, not this config writer.
assert "disabled_providers" not in config
# Compaction buffer scaled to ~10% of the window (compact near 90%).
assert config["compaction"] == {"auto": True, "reserved": 131072 // 10}
@ -2509,6 +2714,109 @@ def test_write_opencode_config_keeps_foreign_disabled_providers(tmp_path):
assert config["disabled_providers"] == ["openai", "gemini"]
def test_write_opencode_config_as_subagent_preserves_parent_model(tmp_path):
path = tmp_path / "opencode.json"
path.write_text(
json.dumps(
{
"model": "anthropic/claude-sonnet-4-5",
"small_model": "anthropic/claude-haiku-4-5",
"compaction": {"auto": False},
}
)
)
local = {**MODEL, "id": MODEL["id"] + ":UD-Q4_K_XL"}
start.write_opencode_config(
BASE,
"sk-unsloth-abc",
local,
path,
as_subagent = True,
)
config = json.loads(path.read_text())
assert config["model"] == "anthropic/claude-sonnet-4-5"
assert config["small_model"] == "anthropic/claude-haiku-4-5"
assert config["compaction"] == {"auto": False}
agent = config["agent"]["unsloth"]
assert agent["mode"] == "subagent"
assert agent["model"] == f"{start._OPENCODE_PROVIDER}/{local['id']}"
assert "local agent" in agent["description"].lower()
assert local["id"] in config["provider"][start._OPENCODE_PROVIDER]["models"]
def test_opencode_subagent_inline_keeps_parent_provider_filters(monkeypatch, tmp_path):
config_path = tmp_path / "opencode.json"
inherited = {"theme": "tokyonight"}
monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", json.dumps(inherited))
monkeypatch.setattr(start, "_which_with_install_dirs", lambda _: "/usr/bin/opencode")
captured = {}
def run(command, **kwargs):
captured["command"] = command
captured.update(kwargs)
return SimpleNamespace(
returncode = 0,
stdout = json.dumps(
{
"enabled_providers": ["opencode-go"],
"disabled_providers": ["ollama", start._OPENCODE_PROVIDER],
"subagent_depth": 0,
}
),
stderr = "",
)
monkeypatch.setattr(start.subprocess, "run", run)
permission = {"edit": "allow"}
inline = start._opencode_subagent_inline_config(config_path, permission)
assert captured["command"] == ["/usr/bin/opencode", "debug", "config"]
assert captured["env"]["OPENCODE_CONFIG"] == str(config_path)
assert inline == {
"theme": "tokyonight",
"enabled_providers": ["opencode-go", start._OPENCODE_PROVIDER],
"disabled_providers": ["ollama"],
"subagent_depth": 1,
"permission": permission,
}
def test_opencode_subagent_inline_preserves_positive_depth(monkeypatch, tmp_path):
monkeypatch.setattr(start, "_which_with_install_dirs", lambda _: "/usr/bin/opencode")
monkeypatch.setattr(
start.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(
returncode = 0,
stdout = json.dumps({"subagent_depth": 3}),
stderr = "",
),
)
inline = start._opencode_subagent_inline_config(tmp_path / "opencode.json", {})
assert inline["subagent_depth"] == 3
def test_opencode_subagent_inline_merges_inherited_filters_without_binary(monkeypatch, tmp_path):
monkeypatch.setenv(
"OPENCODE_CONFIG_CONTENT",
json.dumps(
{
"enabled_providers": ["opencode-go"],
"disabled_providers": ["ollama", start._OPENCODE_PROVIDER],
}
),
)
monkeypatch.setattr(start, "_which_with_install_dirs", lambda _: None)
inline = start._opencode_subagent_inline_config(tmp_path / "opencode.json", {})
assert inline["enabled_providers"] == ["opencode-go", start._OPENCODE_PROVIDER]
assert inline["disabled_providers"] == ["ollama"]
assert inline["subagent_depth"] == 1
def _opencode_inline_config(output: str) -> dict:
# --no-launch prints OPENCODE_CONFIG_CONTENT as a POSIX `export NAME=<shell-quoted>`
# line on Unix/WSL and a PowerShell `$env:NAME = "<escaped>"` line on native Windows;
@ -2597,6 +2905,130 @@ def test_connect_opencode_no_launch(fake_studio, tmp_path):
assert not any(c[1].endswith("/api/inference/status") for c in fake_studio)
def test_connect_opencode_as_subagent_preserves_cloud_parent(fake_studio, tmp_path, monkeypatch):
monkeypatch.setattr(start, "_opencode_subagent_inline_config", lambda path, permission: {})
result = CliRunner().invoke(
start.start_app,
[
"opencode",
"--as-subagent",
"--no-launch",
"--model",
MODEL["id"] + ":UD-Q4_K_XL",
],
)
assert result.exit_code == 0, result.output
assert _launch_command(result.output) == ["opencode"]
expected_model = f"{start._OPENCODE_PROVIDER}/{MODEL['id']}:UD-Q4_K_XL"
# The agent rides in the inline overlay; nothing else comes from the empty base.
assert _opencode_inline_config(result.output) == {
"agent": {
"unsloth": {
"description": start._SUBAGENT_DESCRIPTION,
"mode": "subagent",
"model": expected_model,
"prompt": start._SUBAGENT_INSTRUCTIONS,
}
}
}
path = tmp_path / "agents" / "opencode-subagent" / "opencode.json"
config = json.loads(path.read_text())
assert "model" not in config
assert "small_model" not in config
assert "compaction" not in config
agent = config["agent"]["unsloth"]
assert agent["model"] == expected_model
assert "Unsloth is available as @unsloth and in /models." in result.output
def test_claude_subagent_allowed_tools_precede_forwarded_delimiter(fake_studio):
# A forwarded `--` makes everything after it positional; the tool pre-approval
# must be parsed as an option, so it rides before ctx.args.
result = CliRunner().invoke(
start.start_app,
["claude", "--as-subagent", "--no-launch", "--", "--resume", "abc123"],
)
assert result.exit_code == 0, result.output
command = _launch_command(result.output)
assert command.index("--allowedTools") < command.index("--resume")
def test_opencode_subagent_installs_binary_before_filter_inspection(fake_studio, monkeypatch):
# The effective-config inspection needs the opencode binary; a first launch must
# offer the install before building the overlay, or a global allowlist read only
# after _launch installs OpenCode would filter out the new provider.
installed = {}
monkeypatch.setattr(
start,
"_which_with_install_dirs",
lambda name: "/usr/local/bin/opencode" if installed.get("done") else None,
)
def install(name, hint):
installed["done"] = True
installed["name"] = name
return "/usr/local/bin/opencode"
monkeypatch.setattr(start, "_install_agent", install)
inspected = {}
def inline(path, permission):
inspected["binary"] = start._which_with_install_dirs("opencode")
return {}
monkeypatch.setattr(start, "_opencode_subagent_inline_config", inline)
monkeypatch.setattr(start, "_run", lambda *a, **k: None)
result = CliRunner().invoke(start.start_app, ["opencode", "--as-subagent"])
assert result.exit_code == 0, result.output
assert installed["name"] == "opencode"
assert inspected["binary"] == "/usr/local/bin/opencode"
def test_opencode_subagent_pins_agent_in_inline_overlay(fake_studio, monkeypatch):
# A project opencode.json outranks the session file, so the agent must ride in
# OPENCODE_CONFIG_CONTENT where a repo's own agent.unsloth cannot field-merge over it.
monkeypatch.setattr(start, "_opencode_subagent_inline_config", lambda path, permission: {})
result = CliRunner().invoke(
start.start_app,
["opencode", "--as-subagent", "--no-launch", "--model", MODEL["id"] + ":UD-Q4_K_XL"],
)
assert result.exit_code == 0, result.output
agent = _opencode_inline_config(result.output)["agent"]["unsloth"]
assert agent["mode"] == "subagent"
assert agent["model"] == f"{start._OPENCODE_PROVIDER}/{MODEL['id']}:UD-Q4_K_XL"
assert agent["prompt"] == start._SUBAGENT_INSTRUCTIONS
assert agent["description"] == start._SUBAGENT_DESCRIPTION
def test_connect_opencode_subagent_yolo_no_launch_stays_append_safe(fake_studio, monkeypatch):
monkeypatch.setattr(start, "_opencode_supports_native_auto", lambda: True)
captured = {}
def inline(path, permission):
captured["permission"] = permission
return {"permission": permission}
monkeypatch.setattr(start, "_opencode_subagent_inline_config", inline)
result = CliRunner().invoke(
start.start_app,
["opencode", "--as-subagent", "--no-launch", "--yolo"],
)
assert result.exit_code == 0, result.output
assert _launch_command(result.output) == ["opencode"]
assert "--auto" not in result.output
assert captured["permission"] == {
"edit": "allow",
"bash": "allow",
"webfetch": "allow",
"task": "allow",
"external_directory": {"*": "allow"},
}
assert _opencode_inline_config(result.output)["permission"] == captured["permission"]
# ── Hermes (OpenAI /v1/chat/completions, key via env) ────────────────
@ -2739,6 +3171,39 @@ def test_connect_pi_no_launch(fake_studio, tmp_path):
assert not any(c[1].endswith("/api/inference/status") for c in fake_studio)
def test_connect_pi_as_subagent_preserves_cloud_parent(fake_studio, tmp_path):
result = CliRunner().invoke(
start.start_app,
[
"pi",
"--as-subagent",
"--no-launch",
"--model",
MODEL["id"] + ":UD-Q4_K_XL",
],
)
assert result.exit_code == 0, result.output
command = _launch_command(result.output)
assert command[:2] == ["pi", "--extension"]
assert command[2].endswith("unsloth_cli/pi_subagent.ts")
assert "--provider" not in command
assert "--model" not in command
assert "PI_CODING_AGENT_DIR" not in result.output
assert "export HOME=" not in result.output
assert "UNSLOTH_PI_SUBAGENT_API_KEY" not in result.output
assert "sk-unsloth-feedfacefeedface" not in result.output
config_path = tmp_path / "agents" / "pi-subagent" / "subagent.json"
_assert_env_set(result.output, "UNSLOTH_PI_SUBAGENT_CONFIG", str(config_path))
assert json.loads(config_path.read_text()) == {
"baseUrl": f"{BASE}/v1",
"apiKey": "sk-unsloth-feedfacefeedface",
"model": MODEL["id"] + ":UD-Q4_K_XL",
"contextWindow": 4096,
"maxTokens": 1024,
}
assert "Ask Pi to spawn an Unsloth or local agent." in result.output
def test_connect_pi_no_launch_windows_relocates_userprofile(fake_studio, tmp_path, monkeypatch):
# On native Windows Node resolves ~/.pi via USERPROFILE, not HOME, so the session
# must point USERPROFILE at the relocated home or Pi reads the user's real ~/.pi.
@ -3282,6 +3747,27 @@ def test_opencode_non_yolo_flips_only_explicit_allow(tmp_path):
assert session == {} # a non-yolo session carries no permission inline
def test_opencode_subagent_non_yolo_clears_yolo_task_permission(tmp_path):
path = tmp_path / "opencode.json"
start.write_opencode_config(
BASE,
"sk-unsloth-abc",
MODEL,
path,
yolo = True,
as_subagent = True,
)
start.write_opencode_config(
BASE,
"sk-unsloth-abc",
MODEL,
path,
as_subagent = True,
)
assert json.loads(path.read_text())["permission"]["task"] == "ask"
def test_opencode_non_yolo_leaves_string_permission(tmp_path):
# A global string rule ("deny") is a user-managed catch-all; leave it untouched and
# carry no inline override.