SHA256
518 lines
18 KiB
Python
Executable File
518 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""sjcode client.
|
|
|
|
Terminal coding-agent front end for the sjcode daemon (server/server.py).
|
|
|
|
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. New servers are saved too.
|
|
|
|
Flow:
|
|
coder -> banner, pick saved server (or add one),
|
|
login, pick model, chat REPL
|
|
coder 10.0.0.5:8420 -> skip the server picker
|
|
coder --model qwen3-coder -> skip the model picker
|
|
|
|
REPL commands:
|
|
/help /server /model /models /limit /clear /quit
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import getpass
|
|
import json
|
|
import os
|
|
import re
|
|
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 = 300
|
|
HISTORY_TURNS = 10 # prior turns sent with each message for context
|
|
|
|
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,
|
|
)
|
|
|
|
# Makes the model behave like a terminal agent: short lines, check marks.
|
|
AGENT_PREAMBLE = (
|
|
"You are sj, a coding agent working in the user's terminal.\n"
|
|
"Answer concisely and practically.\n"
|
|
"Describe what you do as short lines; put every action you performed "
|
|
"or verified on its own line starting with '\u2713 ' (check mark).\n"
|
|
"End your reply with the single line 'Done.'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# small terminal helpers (ANSI, no dependencies)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Style:
|
|
def __init__(self, enabled: bool) -> None:
|
|
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} (\u2018/help\u2019 for commands){ST.reset}")
|
|
|
|
|
|
def shorten_home(path: Path) -> str:
|
|
home = Path.home()
|
|
try:
|
|
return "~" / 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 = " (last)" if server_label(entry) == state.get("last") else ""
|
|
user = f" as {entry['username']}" 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, probe it 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 http_json(method: str, base: str, path: str, payload: dict | None = None,
|
|
timeout: int = REQUEST_TIMEOUT) -> 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:
|
|
return resp.status, json.loads(resp.read().decode())
|
|
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) -> tuple[int, dict]:
|
|
base = f"http://{entry['host']}:{entry['port']}"
|
|
return http_json("POST", base, "/chat",
|
|
{"token": token, "message": message, "model": model})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the agent look
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def print_agent_reply(text: str) -> None:
|
|
print(f"{ST.bold}Agent:{ST.reset}")
|
|
for line in text.rstrip().splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("\u2713"):
|
|
print(f" {ST.green}\u2713{ST.reset} {stripped.lstrip('\u2713').strip()}")
|
|
elif stripped.lower() == "done.":
|
|
print(f" {ST.bold}{stripped}{ST.reset}")
|
|
else:
|
|
print(f" {line.strip() if stripped else ''}")
|
|
|
|
|
|
def build_prompt(history: list[tuple[str, str]], message: str) -> str:
|
|
"""Plain-text transcript + new user message (server endpoint is stateless)."""
|
|
parts = [AGENT_PREAMBLE]
|
|
for user_text, answer in history[-HISTORY_TURNS:]:
|
|
parts.append(f"User: {user_text}\nAssistant: {answer}")
|
|
parts.append(f"User: {message}\nAssistant:")
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# chat REPL
|
|
# ---------------------------------------------------------------------------
|
|
|
|
HELP = """
|
|
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)
|
|
/clear forget the conversation history
|
|
/quit exit (also Ctrl+D)
|
|
anything else is sent to the model.
|
|
"""
|
|
|
|
|
|
def chat_loop(entry: dict, token: str, model: str, remaining: float | None) -> 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 == "/clear":
|
|
history.clear()
|
|
print("history cleared")
|
|
else:
|
|
print(f"unknown command {cmd} - try /help")
|
|
continue
|
|
|
|
try:
|
|
status, body = api_chat(entry, token, build_prompt(history, message), model)
|
|
except KeyboardInterrupt:
|
|
print(f"\n{ST.yellow}request cancelled{ST.reset}")
|
|
continue
|
|
|
|
if status != 200:
|
|
print(f"{ST.red}error: {body.get('error', 'unknown')}{ST.reset}")
|
|
if status == 403:
|
|
return # daily limit reached
|
|
continue
|
|
|
|
answer = str(body.get("response", "")).strip() or "(empty reply)"
|
|
history.append((message, answer))
|
|
print()
|
|
print_agent_reply(answer)
|
|
remaining = body.get("remaining_minutes")
|
|
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)")
|
|
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(Path.cwd())
|
|
banner(model or "(server default)", str(project), label)
|
|
chat_loop(entry, token, model, None)
|
|
print("bye")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\nbye")
|