Serverjonas/sjcode
SHA256
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
sjcode/client.py
T
jhkaktus 8f1c3207c9 FIX SOME BUGS
2026-09-12 10:14:55 +02:00

1105 lines
40 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
"""sjcode client.
Terminal coding-agent front end for the sjcode daemon (server/server.py).
This is a real agent: it can run shell commands and change files in the
directory it was started from (the "project directory"). The model emits
tool calls, the client executes them locally and feeds the real output
back to the model until the task is done. You confirm every command that
runs and every file that is written, unless you pass --yes or toggle
/auto in the session.
On start it asks which server to connect to. Servers you used before are
remembered in ~/.config/sjcode/client.json together with username and the
last model, so next time you just press Enter.
Flow:
coder -> banner, pick saved server (or add one),
login, pick model, agent REPL
coder 10.0.0.5:8420 -> skip the server picker
coder --model qwen3-coder -> skip the model picker
coder --yes -> auto-approve every tool call (careful)
REPL commands:
/help /server /model /models /limit /auto /state /cwd /clear /quit
"""
from __future__ import annotations
import argparse
import difflib
import getpass
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "sjcode"
CONFIG_FILE = CONFIG_DIR / "client.json"
DEFAULT_PORT = 8420
REQUEST_TIMEOUT = 600
HISTORY_TURNS = 10 # prior task/summary pairs sent with each task for context
# The project directory the agent works in (where ./coder was started).
CWD = Path.cwd()
MAX_TURNS = 30 # model round-trips per user task
NO_TOOL_LIMIT = 2 # plain-text replies before the turn ends by itself
RUN_TIMEOUT = 300 # seconds per shell command
MAX_RUN_CHARS = 10_000 # shell output fed back / shown, per command
READ_DEFAULT_LIMIT = 250 # lines per read
WRITE_MAX_CHARS = 60_000 # refuse to write bigger files in one go
DIFF_PREVIEW_LINES = 60 # diff lines shown in a confirmation prompt
HOST_RE = re.compile(
r"^(?=.{1,253}\z)" # total length limit
r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+" # domain labels +
r"[a-z]{2,63}\z" # a TLD
r"|^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\z" # or single-label hostname
r"|^(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}\z" # IPv4
r"|^(?:[0-9a-f]{0,4}:){2,7}[0-9a-f]{0,4}\z", # IPv6 (2+ colons)
re.IGNORECASE,
)
# Tool-call protocol: the model answers with normal text plus fenced blocks.
# ```run
# git status --short
# ```
# ```sj:write
# {"path": "x.txt", "content": "hi"}
# ```
# The closing fence must be a line containing only ``` so that content which
# itself contains backticks (inside a JSON string) cannot end the block early.
# (?![a-z]) keeps language fences like ```rust or ```ruby from being read as
# run-tool calls.
TAG_RE = re.compile(
r"(?ms)^[ \t]*```(?:sj:([a-z]+)|run)(?![a-z])[^\n]*\n(.*?)(?=\n[ \t]*```[ \t]*(?:\n|\Z)|\Z)"
)
STRAY_FENCE_RE = re.compile(r"(?m)^[ \t]*```[^\n]*$")
AGENT_PREAMBLE = """\
You are sj, a real coding agent working in the user's project directory on
their machine. You can actually run commands and change files. Work like an
experienced engineer:
- Plan silently, act step by step. After every step, look at the real output
you got before deciding the next step.
- Explore before you change code you have not seen: list files, read them,
then edit. Never guess file contents.
- Keep replies short. Say what you are doing, then do it.
- Every action you performed or verified goes on its own line starting with
"✓ " (check mark).
- If something fails, read the error and change your approach; do not repeat
a failing call unchanged and do not apologize.
- When the whole task is finished, end your reply with the single line "Done."
You act by emitting tool calls. A tool call is a fenced code block tagged
"run" or "sj:<tool>" whose body is plain text (run) or a JSON object (all
other tools). Examples:
```run
git status --short
```
```sj:write
{"path": "notes/hello.txt", "content": "hi\\n"}
```
```sj:edit
{"path": "app.py", "old": "old text", "new": "new text", "all": false}
```
```sj:read
{"path": "app.py", "start": 1, "limit": 200}
```
```sj:files
{"path": ".", "limit": 200}
```
Tools:
- run: run one shell command in the project directory. Every run starts
in the project directory again (cd does not persist).
- read: read a text file. Args: path, start (line number), limit (lines).
- files: list project files. Args: path (subdirectory), limit.
- write: create a new file or overwrite a small one. Args: path, content.
- edit: replace exact text in a file. Args: path, old, new, all (optional).
Rules:
- Emit at most ONE tool call per reply, then stop. You will receive its real
output as the next message and then continue.
- The user confirms dangerous actions. If a result says "cancelled by user",
accept that, adjust, and move on; never repeat the same call.
- Never invent tool output. If you need information, call a tool to get it.
"""
# ---------------------------------------------------------------------------
# small terminal helpers (ANSI, no dependencies)
# ---------------------------------------------------------------------------
class Style:
def __init__(self, enabled: bool) -> None:
self.enabled = enabled
self.dim = "\033[2m" if enabled else ""
self.bold = "\033[1m" if enabled else ""
self.green = "\033[32m" if enabled else ""
self.yellow = "\033[33m" if enabled else ""
self.red = "\033[31m" if enabled else ""
self.cyan = "\033[36m" if enabled else ""
self.reset = "\033[0m" if enabled else ""
def style_enabled() -> bool:
if os.environ.get("NO_COLOR"):
return False
return sys.stdout.isatty()
ST = Style(style_enabled())
def ask(prompt: str) -> str:
"""input() that exits cleanly on Ctrl+D instead of crashing."""
try:
return input(prompt)
except EOFError:
print()
sys.exit(1)
def banner(model: str, project: str, server: str) -> None:
title = "sj Coding Agent"
lines = [f"Model: {model}", f"Project: {project}"]
width = max(len(title), *(len(l) for l in lines)) + 2
print(f"\u256d{'\u2500' * (width + 2)}\u256e")
print(f"\u2502 {title.center(width)} \u2502")
for line in lines:
print(f"\u2502 {line.ljust(width)} \u2502")
print(f"\u2570{'\u2500' * (width + 2)}\u256f")
print(f"{ST.dim}server: {server} ('/help' for commands){ST.reset}")
def shorten_home(path: Path) -> Path:
home = Path.home()
try:
return Path("~") / path.relative_to(home)
except ValueError:
return path
# ---------------------------------------------------------------------------
# saved servers
# ---------------------------------------------------------------------------
def load_state() -> dict:
try:
with open(CONFIG_FILE, encoding="utf-8") as fh:
data = json.load(fh)
if isinstance(data, dict) and isinstance(data.get("servers"), list):
return data
except (OSError, json.JSONDecodeError):
pass
return {"servers": [], "last": ""}
def save_state(state: dict) -> None:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(CONFIG_FILE, "w", encoding="utf-8") as fh:
json.dump(state, fh, indent=2)
def server_label(entry: dict) -> str:
return f"{entry['host']}:{entry['port']}"
def find_server(state: dict, host: str, port: int) -> dict | None:
for entry in state["servers"]:
if entry["host"] == host and int(entry["port"]) == port:
return entry
return None
def parse_address(text: str) -> tuple[str, int]:
"""Accept 'domain', 'ip', 'host:port', IPv6 or a full http(s) URL."""
text = text.strip()
if "://" in text:
parsed = urllib.parse.urlsplit(text)
host, port = parsed.hostname or "", parsed.port or DEFAULT_PORT
elif text.startswith("["): # [ipv6]:port
parsed = urllib.parse.urlsplit("//" + text)
host, port = parsed.hostname or "", parsed.port or DEFAULT_PORT
elif text.count(":") > 1: # bare IPv6 literal (no port part)
host, port = text, DEFAULT_PORT
else:
host, _, port_s = text.rpartition(":")
if not host or not port_s.isdigit() or not (0 < int(port_s) < 65536):
host, port = text, DEFAULT_PORT
else:
port = int(port_s)
host = host.strip("[]").strip().lower()
if not host or not HOST_RE.match(host):
raise ValueError(f"not a valid host: {text!r}")
return host, port
def pick_server(state: dict) -> tuple[dict, bool]:
"""Ask which server to use. Returns (entry, created_new)."""
servers = state["servers"]
print(f"\n{ST.bold}Connect to which server?{ST.reset}")
for i, entry in enumerate(servers, 1):
default = f" {ST.dim}(last){ST.reset}" if server_label(entry) == state.get("last") else ""
user = f" {ST.dim}({entry['username']}){ST.reset}" if entry.get("username") else ""
print(f" {i}) {server_label(entry)}{user}{default}")
print(f" n) add a new server\n")
hint = ""
if servers:
idx = next(
(i for i, e in enumerate(servers, 1) if server_label(e) == state.get("last")),
1,
)
hint = str(idx)
while True:
choice = ask(f"server [1-{len(servers)}/n{', Enter=' + hint if hint else ''}]: ").strip().lower()
if not choice and hint:
entry = servers[int(hint) - 1]
return entry, False
if choice == "n":
entry = add_server(state)
return entry, True
if choice.isdigit() and 1 <= int(choice) <= len(servers):
return servers[int(choice) - 1], False
print(f"{ST.red}please enter 1-{len(servers)}, 'n', or Enter{ST.reset}")
def add_server(state: dict) -> dict:
"""Ask for a new server address, and save it."""
while True:
raw = ask("server address (domain or ip:port): ").strip()
try:
host, port = parse_address(raw)
except ValueError as exc:
print(f"{ST.red}{exc} - use e.g. 192.168.1.10:8420 or code.my.lan{ST.reset}")
continue
entry = find_server(state, host, port)
if entry is None:
entry = {"host": host, "port": port, "username": "", "model": ""}
state["servers"].append(entry)
break
return entry
# ---------------------------------------------------------------------------
# daemon API
# ---------------------------------------------------------------------------
def _feed_chunk(line: str, on_chunk, final: dict) -> dict:
"""Handle one JSON line of a streamed reply: chunk -> callback, else final."""
try:
obj = json.loads(line)
except json.JSONDecodeError:
return final
if isinstance(obj, dict) and "chunk" in obj:
on_chunk(obj)
return final
return obj if isinstance(obj, dict) else final
def http_json(method: str, base: str, path: str, payload: dict | None = None,
timeout: int = REQUEST_TIMEOUT, on_chunk=None) -> tuple[int, dict]:
url = base + path
data = None if payload is None else json.dumps(payload).encode()
req = urllib.request.Request(
url, data=data, method=method,
headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
if on_chunk is None:
return resp.status, json.loads(resp.read().decode())
# streamed reply: one JSON object per line
status = resp.status
final: dict = {}
buf = ""
for raw in resp:
buf += raw.decode(errors="replace")
while "\n" in buf:
line, buf = buf.split("\n", 1)
if line.strip():
final = _feed_chunk(line.strip(), on_chunk, final)
if buf.strip():
final = _feed_chunk(buf.strip(), on_chunk, final)
return status, final
except urllib.error.HTTPError as exc:
try:
return exc.code, json.loads(exc.read().decode())
except Exception:
return exc.code, {"error": f"HTTP {exc.code}"}
except (urllib.error.URLError, TimeoutError, ConnectionError, OSError) as exc:
return 0, {"error": f"cannot reach server: {exc}"}
def probe_server(entry: dict, probe: bool = True) -> str | None:
"""Return an error string, or None if the daemon answers.
With probe=False, skip the check (e.g. HTTP proxies that intercept
the port make a probe useless; login will fail with a clear error).
"""
if not probe:
return None
base = f"http://{entry['host']}:{entry['port']}"
status, body = http_json("GET", base, "/status", timeout=5)
if status != 200:
return str(body.get("error", f"no answer from {server_label(entry)}"))
return None
def api_login(entry: dict, username: str, password: str) -> tuple[int, dict]:
base = f"http://{entry['host']}:{entry['port']}"
return http_json("POST", base, "/login",
{"username": username, "password": password}, timeout=15)
def api_models(entry: dict, token: str) -> list[str]:
base = f"http://{entry['host']}:{entry['port']}"
status, body = http_json(
"GET", base, f"/models?token={urllib.parse.quote(token)}", timeout=15)
return list(body.get("models", [])) if status == 200 else []
def api_chat(entry: dict, token: str, message: str, model: str,
on_chunk=None) -> tuple[int, dict]:
base = f"http://{entry['host']}:{entry['port']}"
payload: dict = {"token": token, "message": message, "model": model}
if on_chunk is not None:
payload["stream"] = True
return http_json("POST", base, "/chat", payload, on_chunk=on_chunk)
return http_json("POST", base, "/chat", payload)
# ---------------------------------------------------------------------------
# login / model selection
# ---------------------------------------------------------------------------
def login(entry: dict) -> str | None:
"""Ask credentials, return the session token (or None)."""
default = entry.get("username") or ""
while True:
username = ask(f"username [{default}]: ").strip() or default
if not username:
print(f"{ST.red}username required{ST.reset}")
continue
if sys.stdin.isatty():
try:
password = getpass.getpass("password: ")
except EOFError:
print()
sys.exit(1)
else: # piped stdin (scripted runs): plain input
password = ask("password: ")
status, body = api_login(entry, username, password)
if status == 200:
entry["username"] = username
left = body.get("remaining_minutes")
time_left = "unlimited" if left is None else f"{left:g} min left today"
print(f"{ST.green}logged in as {username}{ST.reset} "
f"{ST.dim}({time_left}){ST.reset}\n")
return str(body["token"])
print(f"{ST.red}login failed: {body.get('error', 'unknown')}{ST.reset}")
if status == 403: # daily limit reached - no point retrying
return None
def pick_model(entry: dict, models: list[str]) -> str:
"""Ask which model to use; preselects the saved one."""
if not models:
print(f"{ST.yellow}server lists no models, using daemon default{ST.reset}")
return ""
if len(models) == 1:
return models[0]
saved = entry.get("model") or ""
print(f"{ST.bold}Models on {server_label(entry)}:{ST.reset}")
for i, name in enumerate(models, 1):
mark = f" {ST.dim}(last used){ST.reset}" if name == saved else ""
print(f" {i}) {name}{mark}")
print()
while True:
choice = ask(f"model [1-{len(models)}, Enter="
f"{models.index(saved) + 1 if saved in models else 1}]: ").strip()
if not choice:
return saved if saved in models else models[0]
if choice in models:
return choice
if choice.isdigit() and 1 <= int(choice) <= len(models):
return models[int(choice) - 1]
print(f"{ST.red}pick 1-{len(models)} or a model name{ST.reset}")
# ---------------------------------------------------------------------------
# tools the agent can use (executed locally, in the project directory)
# ---------------------------------------------------------------------------
class ToolError(Exception):
"""A tool refused to run; the message goes back to the model."""
class ToolAbort(Exception):
"""The user aborted the whole agent turn at a confirmation prompt."""
ABORT = "!!abort"
EXCLUDED_DIRS = {
".git", "__pycache__", "node_modules", ".venv", "venv", ".mypy_cache",
".ruff_cache", ".pytest_cache", ".tox", ".idea", ".vscode", ".eggs",
"dist", "build", ".hg", ".svn",
}
def _resolve(path: str) -> Path:
"""Resolve a path against the project dir; refuse anything outside it."""
if not path:
raise ToolError("path required")
p = Path(path).expanduser()
if not p.is_absolute():
p = CWD / p
try:
rp = p.resolve()
except OSError as exc:
raise ToolError(f"cannot resolve path: {exc}")
inside = rp == CWD or str(rp).startswith(str(CWD) + os.sep)
if not inside:
raise ToolError(
f"outside the project directory ({CWD}): {path!r}")
return rp
def _display(p: Path) -> str:
try:
rel = str(p.relative_to(CWD))
return rel or "."
except ValueError:
return str(p)
def _unified_diff(old: str, new: str, display: str) -> list[str]:
diff = difflib.unified_diff(
old.splitlines(), new.splitlines(),
fromfile=f"{display} (old)", tofile=f"{display} (new)", lineterm="",
)
lines = list(diff)
if len(lines) > DIFF_PREVIEW_LINES:
lines = lines[:DIFF_PREVIEW_LINES]
lines.append(f"... (+{len(lines)} more diff lines)")
return lines
def _run_args(args: dict) -> tuple[str, int, Path]:
"""Validate run-tool args -> (command, timeout, working dir)."""
cmd_text = str(args.get("command") or args.get("cmd") or "").strip()
if not cmd_text:
raise ToolError("'command' required")
timeout = args.get("timeout")
try:
timeout = min(max(int(timeout), 5), RUN_TIMEOUT) if timeout else RUN_TIMEOUT
except (TypeError, ValueError):
timeout = RUN_TIMEOUT
wd = _resolve(str(args.get("cwd") or "."))
if not wd.is_dir():
raise ToolError(f"not a directory: {_display(wd)}")
return cmd_text, timeout, wd
def tool_run_confirm(args: dict) -> list[str]:
cmd_text, timeout, wd = _run_args(args)
return [
f"$ {cmd_text}",
f"cwd: {_display(wd)} timeout: {timeout}s",
]
def tool_run_execute(args: dict) -> list[str]:
cmd_text, timeout, wd = _run_args(args)
try:
proc = subprocess.run(
cmd_text, shell=True, cwd=str(wd), capture_output=True,
text=True, timeout=timeout,
)
except subprocess.TimeoutExpired:
return [f"\u2717 $ {cmd_text}", f"timed out after {timeout}s"]
except OSError as exc:
return [f"\u2717 $ {cmd_text}", f"cannot run: {exc}"]
ok = proc.returncode == 0
head = (f"\u2713 $ {cmd_text}" if ok
else f"\u2717 $ {cmd_text} (exit {proc.returncode})")
parts = []
if (proc.stdout or "").strip():
parts.append(proc.stdout.rstrip())
if (proc.stderr or "").strip():
parts.append(proc.stderr.rstrip())
text = "\n".join(parts).strip()
if len(text) > MAX_RUN_CHARS:
text = text[:MAX_RUN_CHARS] + f"\n... (output truncated at {MAX_RUN_CHARS} chars)"
return [head] + (text.splitlines() if text else [])
def tool_read(args: dict) -> list[str]:
path = str(args.get("path", "")).strip()
p = _resolve(path)
if not p.is_file():
raise ToolError(f"not a file: {_display(p)}")
try:
text = p.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
raise ToolError(str(exc))
lines = text.splitlines()
total = len(lines)
try:
start = max(1, int(args.get("start", 1)))
except (TypeError, ValueError):
start = 1
try:
limit = int(args.get("limit", READ_DEFAULT_LIMIT))
except (TypeError, ValueError):
limit = READ_DEFAULT_LIMIT
limit = max(1, min(limit, 1000))
start = min(start, max(total, 1))
sel = lines[start - 1:start - 1 + limit]
out = [f"\u2713 read {_display(p)} ({total} lines)"]
for i, line in enumerate(sel, start):
out.append(f"{i}\t{line[:400]}")
if start - 1 + limit < total:
out.append(f"(... {total - (start - 1 + limit)} more lines - "
f"read again with 'start'/'limit')")
return out
def tool_files(args: dict) -> list[str]:
sub = str(args.get("path") or ".").strip()
base = _resolve(sub)
if not base.is_dir():
raise ToolError(f"not a directory: {_display(base)}")
try:
limit = max(1, min(int(args.get("limit", 200)), 500))
except (TypeError, ValueError):
limit = 200
out: list[str] = []
truncated = False
for root, dirs, fnames in os.walk(base):
dirs[:] = sorted(
d for d in dirs
if d not in EXCLUDED_DIRS and not d.endswith(".egg-info")
)
for fn in sorted(fnames):
if len(out) >= limit:
truncated = True
break
out.append(_display(Path(root) / fn))
if truncated:
break
head = (f"\u2713 files under {_display(base)} ({len(out)} shown"
+ (", truncated - narrow with 'path'" if truncated else "") + ")")
return [head] + out
def tool_write_confirm(args: dict) -> list[str]:
path = str(args.get("path", "")).strip()
if not path:
raise ToolError("path required")
content = args.get("content")
if not isinstance(content, str):
raise ToolError("'content' (string) required")
if len(content) > WRITE_MAX_CHARS:
raise ToolError(
f"content too large ({len(content)} chars, max {WRITE_MAX_CHARS}) - "
f"write it in smaller pieces or use 'run'")
p = _resolve(path)
exists = p.is_file()
if exists:
old = p.read_text(encoding="utf-8", errors="replace")
if len(old) > 2000:
raise ToolError(
f"{_display(p)} exists and is {len(old)} chars; refusing to "
f"overwrite it wholesale - use 'edit' instead")
else:
old = ""
summary = [
f"{'overwrite' if exists else 'create'} {_display(p)} "
f"({len(content.splitlines())} lines)"
]
summary += _unified_diff(old, content, _display(p))
return summary
def tool_write_execute(args: dict) -> list[str]:
p = _resolve(str(args.get("path", "")).strip())
content = str(args.get("content", ""))
p.parent.mkdir(parents=True, exist_ok=True)
text = content if content.endswith("\n") else content + "\n"
p.write_text(text, encoding="utf-8")
return [f"\u2713 wrote {_display(p)} ({len(text.splitlines())} lines)"]
def _flag(args: dict, key: str) -> bool:
return args.get(key) in (True, "true", "True", 1, "1", "yes")
def tool_edit_confirm(args: dict) -> list[str]:
path = str(args.get("path", "")).strip()
if not path:
raise ToolError("path required")
old, new = args.get("old"), args.get("new")
if not isinstance(old, str) or not old:
raise ToolError("'old' (exact existing text) required")
if not isinstance(new, str):
raise ToolError("'new' (replacement text) required")
p = _resolve(path)
if not p.is_file():
raise ToolError(f"not a file: {_display(p)}")
body = p.read_text(encoding="utf-8", errors="replace")
n = body.count(old)
if n == 0:
raise ToolError(
f"search text not found in {_display(p)} - 'read' the file and "
f"copy the text exactly")
all_flag = _flag(args, "all")
if n > 1 and not all_flag:
raise ToolError(
f"'old' matches {n} times in {_display(p)} - include more "
f"surrounding text or set \"all\": true")
updated = body.replace(old, new) if all_flag else body.replace(old, new, 1)
summary = [
f"edit {_display(p)} "
f"({'all' if all_flag else 'first'} of {n} match{'es' if n != 1 else ''})"
]
summary += _unified_diff(body, updated, _display(p))
return summary
def tool_edit_execute(args: dict) -> list[str]:
p = _resolve(str(args.get("path", "")).strip())
old, new = str(args.get("old", "")), str(args.get("new", ""))
all_flag = _flag(args, "all")
body = p.read_text(encoding="utf-8", errors="replace")
updated = body.replace(old, new) if all_flag else body.replace(old, new, 1)
p.write_text(updated, encoding="utf-8")
return [f"\u2713 edited {_display(p)}"]
# name -> {confirm, execute, confirm_first}
TOOLS: dict[str, dict] = {
"run": {"confirm": tool_run_confirm, "execute": tool_run_execute, "confirm_first": True},
"read": {"confirm": tool_read, "execute": tool_read, "confirm_first": False},
"files": {"confirm": tool_files, "execute": tool_files, "confirm_first": False},
"write": {"confirm": tool_write_confirm, "execute": tool_write_execute, "confirm_first": True},
"edit": {"confirm": tool_edit_confirm, "execute": tool_edit_execute, "confirm_first": True},
}
def confirm_tool(name: str, summary: list[str], auto: bool) -> str:
"""Ask the user to approve a tool call.
Returns "y", "n", ABORT, or a free-text instruction from the user.
"""
if auto:
return "y"
print(f"{ST.bold}{ST.yellow}? {name}{ST.reset}")
for line in summary:
print(f" {line}")
try:
answer = input(
f" {ST.bold}[Enter=yes n=no a=abort or type a new instruction]{ST.reset} "
).strip()
except KeyboardInterrupt:
print()
return "n"
except EOFError:
print()
return ABORT
low = answer.lower()
if low in ("", "y", "yes"):
return "y"
if low in ("n", "no", "c", "cancel"):
return "n"
if low in ("a", "abort"):
return ABORT
return answer
def handle_tool_call(name: str, raw_body: str, auto: bool) -> tuple[bool, list[str]]:
"""Validate, confirm and execute one tool call. Returns (ok, lines)."""
spec = TOOLS.get(name)
if spec is None:
return False, [f"\u2717 unknown tool '{name}' - "
f"available: {', '.join(sorted(TOOLS))}"]
if name == "run":
# Accept the plain-text command (as documented) or a JSON object
# with a "command" key - some models emit JSON anyway.
body = (raw_body or "").strip()
args = {"command": body}
if body.startswith("{"):
try:
parsed = json.loads(body)
if isinstance(parsed, dict) and "command" in parsed:
args = parsed
except json.JSONDecodeError:
pass # e.g. a brace group in shell - run it as-is
else:
try:
args = json.loads(raw_body or "{}")
if not isinstance(args, dict):
raise ValueError("arguments must be a JSON object")
except json.JSONDecodeError as exc:
return False, [f"\u2717 bad JSON for '{name}': {exc} - fix it and retry"]
except ValueError as exc:
return False, [f"\u2717 {exc}"]
try:
prepared = spec["confirm"](args) # validates; for read/files it already runs
except ToolError as exc:
return False, [f"\u2717 {exc}"]
except (KeyError, TypeError, ValueError, OSError) as exc:
return False, [f"\u2717 {exc}"]
if not spec["confirm_first"]: # read/files: safe, results are in
return True, prepared
decision = "y" if auto else confirm_tool(name, prepared, auto)
if decision == ABORT:
raise ToolAbort()
if decision != "y":
note = "cancelled by user" if decision == "n" else f"user says: {decision}"
return False, [f"\u2717 {note} - do not repeat this exact call; "
f"adjust your approach or stop"]
try:
lines = spec["execute"](args)
# A failing command is not a tool error, but the model must see it
# as a failure: the first output line starts with the cross mark.
ok = not (lines and lines[0].startswith("\u2717"))
return ok, lines
except ToolError as exc:
return False, [f"\u2717 {exc}"]
except (OSError, ValueError, RuntimeError) as exc:
return False, [f"\u2717 {exc}"]
# ---------------------------------------------------------------------------
# agent turn: model reply -> tool calls -> real output -> model ...
# ---------------------------------------------------------------------------
def read_tool_tags(raw: str) -> list[tuple[str, str]]:
"""Extract tool calls in order: [(name, body), ...]."""
tags: list[tuple[str, str]] = []
for m in TAG_RE.finditer(raw):
name = m.group(1) or "run"
tags.append((name, (m.group(2) or "").strip()))
return tags
def print_agent_reply(text: str) -> None:
print(f"{ST.bold}Agent:{ST.reset}")
for line in text.rstrip().splitlines():
if line.startswith("\u2713"):
print(f"{ST.green}{line}{ST.reset}")
elif line.startswith("\u2717"):
print(f"{ST.red}{line}{ST.reset}")
else:
print(line)
def process_reply(raw: str, auto: bool, show_text: bool = True
) -> tuple[bool, str, list]:
"""Print the reply, run its tool calls. Returns (had_tools, text, results)."""
# Tool blocks out, leftover closing-fence lines out, blank edges trimmed.
reply_text = STRAY_FENCE_RE.sub("", TAG_RE.sub("", raw)).strip()
if reply_text and show_text: # already shown live when streamed
print_agent_reply(reply_text)
tags = read_tool_tags(raw)
if not tags:
return False, reply_text, []
results: list[tuple[str, bool, list[str]]] = []
for name, body in tags:
print(f"{ST.dim}\u2014 tool: {name}{ST.reset}")
ok, lines = handle_tool_call(name, body, auto)
results.append((name, ok, lines))
for i, line in enumerate(lines):
if i == 0 and not ok:
print(f"{ST.red}{line}{ST.reset}")
elif line.startswith("\u2713"):
print(f"{ST.green}{line}{ST.reset}")
else:
print(f" {line}")
return True, reply_text, results
def agent_turn(entry: dict, token: str, model: str,
history: list[tuple[str, str]], task: str, auto: bool
) -> tuple[str, str, float | None]:
"""Run one user task through the agent loop.
Returns (status, final_text, remaining_minutes):
status is "done", "stopped", "cancelled", "limit" or "error".
"""
transcript = ""
for user_text, answer in history[-HISTORY_TURNS:]:
transcript += f"User: {user_text}\nAssistant: {answer}\n\n"
messages = [AGENT_PREAMBLE
+ (("\n\n" + transcript) if transcript else "")
+ f"\n\nUser: {task}\nAssistant:"]
remaining: float | None = None
no_tool = 0
streamed = {"any": False}
def on_chunk(obj: dict) -> None:
piece = obj.get("chunk")
if not piece:
return
if not streamed["any"]:
streamed["any"] = True
print(f"{ST.bold}Agent:{ST.reset}")
sys.stdout.write(f"{ST.dim}{piece}{ST.reset}" if ST.enabled else piece)
sys.stdout.flush()
try:
for turn in range(1, MAX_TURNS + 1):
try:
status, body = api_chat(
entry, token, "\n\n".join(messages), model,
on_chunk=on_chunk)
except KeyboardInterrupt:
print()
return "cancelled", "", remaining
if status != 200 or body.get("error"):
print(f"\n{ST.red}error: "
f"{body.get('error', 'unknown')}{ST.reset}")
if status == 403:
return "limit", "", remaining
return "error", "", remaining
remaining = body.get("remaining_minutes", remaining)
raw = str(body.get("response", ""))
if streamed["any"]:
print()
had_tools, reply_text, results = process_reply(
raw, auto, show_text=not streamed["any"])
streamed["any"] = False
if not had_tools:
no_tool += 1
if no_tool >= NO_TOOL_LIMIT:
return "done", reply_text, remaining
messages.append(
"(no tool call received) If the task is complete, reply "
"with exactly: Done.\nOtherwise continue with the next "
"tool call.")
continue
no_tool = 0
if turn >= MAX_TURNS:
print(f"{ST.yellow}step limit ({MAX_TURNS}) reached - "
f"stopping{ST.reset}")
return "stopped", reply_text, remaining
out = []
for name, ok, lines in results:
out.append(f"{name}: {'ok' if ok else 'FAILED'}")
out.extend(" " + ln for ln in lines)
messages.append(
"Tool results (real output from your last tool call):\n"
+ ("\n".join(out) or "(no output)")
+ "\n\nContinue with the next step.")
except ToolAbort:
print(f"{ST.yellow}stopped by user{ST.reset}")
return "stopped", "", remaining
except KeyboardInterrupt:
print()
return "cancelled", "", remaining
print(f"{ST.yellow}step limit ({MAX_TURNS}) reached - stopping{ST.reset}")
return "stopped", "", remaining
# ---------------------------------------------------------------------------
# chat REPL
# ---------------------------------------------------------------------------
HELP = f"""\
commands:
/help show this help
/server show the connected server
/models list models on the server
/model <name> switch model (also saved for this server)
/limit show remaining minutes today (last known)
/auto toggle confirmations (auto-approve every tool call)
/state show auto mode, project dir, history size
/cwd show the project directory
/clear forget the conversation history
/quit exit (also Ctrl+D)
anything else is a task for the agent. It can run commands and change files
in the project directory; you confirm every action unless auto mode is on.
available tools: {', '.join(sorted(TOOLS))}
"""
def chat_loop(entry: dict, token: str, model: str,
remaining: float | None, auto: bool) -> None:
history: list[tuple[str, str]] = []
while True:
try:
message = input(f"{ST.cyan}you>{ST.reset} ").strip()
except EOFError:
print()
return
except KeyboardInterrupt:
print()
continue
if not message:
continue
if message.startswith("/"):
cmd, _, arg = message.partition(" ")
if cmd in ("/quit", "/exit", "/q"):
return
if cmd == "/help":
print(HELP.strip())
elif cmd == "/server":
print(f"connected to {server_label(entry)} as {entry.get('username')}")
elif cmd == "/models":
models = api_models(entry, token)
print("\n".join(models) or f"{ST.red}no models listed{ST.reset}")
elif cmd == "/model":
if arg.strip():
model = arg.strip()
entry["model"] = model
print(f"model set to {model}")
else:
print(f"current model: {model or '(daemon default)'}")
elif cmd == "/limit":
left = "unlimited" if remaining is None else f"{remaining:g} min"
print(f"time left today: {left}")
elif cmd == "/auto":
auto = not auto
print("auto mode ON - tool calls run without confirmation"
if auto else
"auto mode OFF - every action asks first")
elif cmd == "/state":
print(f"auto: {'on' if auto else 'off'}")
print(f"project dir: {CWD}")
print(f"history: {len(history)} turn(s)")
print(f"model: {model or '(daemon default)'}")
elif cmd == "/cwd":
print(str(CWD))
elif cmd == "/clear":
history.clear()
print("history cleared")
else:
print(f"unknown command {cmd} - try /help")
continue
status, final, remaining = agent_turn(
entry, token, model, history, message, auto)
if status == "limit":
return # daily limit reached
if status == "done" and final:
history.append((message, final))
if remaining is not None:
print(f"{ST.dim} ({remaining:g} min left today){ST.reset}")
print()
# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------
def main() -> None:
if os.name == "nt":
os.system("") # enable ANSI escapes in classic Windows consoles
parser = argparse.ArgumentParser(description="sj coding agent client")
parser.add_argument("server", nargs="?",
help="server address (domain or ip:port), skips the picker")
parser.add_argument("-m", "--model", help="model to use, skips the picker")
parser.add_argument("--no-probe", action="store_true",
help="skip the reachability check before login (use behind "
"HTTP proxies that answer for every port)")
parser.add_argument("-y", "--yes", action="store_true",
help="auto mode: run tool calls without confirmation "
"(the agent edits real files - be careful)")
args = parser.parse_args()
state = load_state()
# -- choose server ------------------------------------------------------
if args.server:
try:
host, port = parse_address(args.server)
except ValueError:
sys.exit(f"error: cannot parse address '{args.server}'")
entry = find_server(state, host, port)
created = entry is None
if entry is None:
entry = {"host": host, "port": port, "username": "", "model": ""}
state["servers"].append(entry)
else:
if not state["servers"]:
print(f"{ST.bold}First start - let's add a server.{ST.reset}")
entry = add_server(state)
created = True
else:
entry, created = pick_server(state)
label = server_label(entry)
if created:
save_state(state) # persist immediately, before anything can go wrong
print(f"\nconnecting to {label} ...")
if (err := probe_server(entry, probe=not args.no_probe)) is not None:
print(f"{ST.red}error: {err}{ST.reset}")
if created:
print(f"{ST.dim}'{label}' was saved - start coder again once it is reachable{ST.reset}")
else:
print(f"{ST.dim}saved under '{label}' - try again later{ST.reset}")
sys.exit(1)
# -- login ----------------------------------------------------------------
token = login(entry)
if token is None:
save_state(state)
sys.exit(1)
# -- choose model ---------------------------------------------------------
models = api_models(entry, token)
model = args.model or pick_model(entry, models)
if model:
entry["model"] = model
state["last"] = label
save_state(state)
project = shorten_home(CWD)
banner(model or "(server default)", str(project), label)
if args.yes:
print(f"{ST.yellow}auto mode: tool calls run without confirmation{ST.reset}")
print(f"{ST.dim}working dir: {CWD}{ST.reset}")
chat_loop(entry, token, model, None, args.yes)
print("bye")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nbye")
Reference in New Issue View Git Blame Copy Permalink