WHC
2d1052fd5f
Streaming: Chinese reasoning showed up garbled in the WebUI. The agent loop now feeds visible/thinking text in runs instead of byte by byte, and the WebUI holds an incomplete UTF-8 tail until the next chunk (cst_utf8_hold_feed), so every think_delta/assistant_delta is whole. src/provider_switch.c: model list fetch/cache, key verify-and-save and provider switching moved out of tui.c and shared by TUI and WebUI. WebUI API: - GET /api/providers, GET /api/models[?refresh=1] (401 on a rejected key, still listing built-ins), POST /api/provider, POST /api/key (key checked against /v1/models, saved to credentials 0600, never echoed in replies, events or logs) - /api/config validates everything before applying, persists model, effort and approval (not yolo), warns when effort does not apply; plan_mode no longer forces read-only - state: turn_usage, api_base, key_set/key_from, model_info, effort_applies, sees_images, max_steps, tool_profile, perm_timeout; permission events carry their timeout Frontend: searchable model picker with refresh and manual entry, provider picker and key dialog, effort/vision hints, toasts instead of silent failures, Chinese error text, real protocol in the settings and usage panels, approval countdown from the server's timeout. Tests: utf8 hold unit test, scripts/e2e_webui_settings.py (fake /v1/models, wrong key, switching, persistence, Chinese thinking) run from e2e_webui.sh, scripts/tui_keys/provider.py.
66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""TUI /provider and /key against a fake API whose /models checks the key.
|
|
|
|
nix-shell -p python3Packages.pyte --run "python3 scripts/tui_keys/provider.py"
|
|
|
|
A wrong key is asked for again, a good one is saved and switches the
|
|
provider; /key replaces it. Same code path as the WebUI (provider_switch.c).
|
|
"""
|
|
import json, os, sys, threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from harness import Term
|
|
from keys import check, base
|
|
|
|
GOOD = {"sk-good", "sk-good2"}
|
|
|
|
|
|
class B(BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
auth = self.headers.get("Authorization", "")
|
|
ok = auth.startswith("Bearer ") and auth[7:] in GOOD
|
|
body = json.dumps({"data": [{"id": "fake-one"}]} if ok else {"error": {"message": "bad key"}}).encode()
|
|
self.send_response(200 if ok else 401)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), B)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
|
|
t = Term(args=("-b", base, "-k", "sk-test"))
|
|
cfg = os.path.join(t.home, ".config", "cst-agent")
|
|
os.makedirs(cfg, exist_ok=True)
|
|
t.close()
|
|
with open(os.path.join(cfg, "models.json"), "w") as f:
|
|
json.dump({"providers": {"fakeb": {"api_base": "http://127.0.0.1:%d/v1" % srv.server_address[1],
|
|
"key_env": "FAKEB_API_KEY", "default_model": ""}}}, f)
|
|
t = Term(args=("-b", base, "-k", "sk-test"), home=t.home, env={"FAKEB_API_KEY": ""})
|
|
creds = os.path.join(cfg, "credentials")
|
|
|
|
|
|
def screen():
|
|
return "\n".join(t.lines())
|
|
|
|
|
|
t.send("/provider fakeb\r", 0.8)
|
|
check("a key prompt for fakeb", "API key for fakeb" in screen(), t)
|
|
t.send("sk-bad\r", 1.5)
|
|
check("a wrong key is asked for again", "rejected that key" in screen(), t)
|
|
check("the wrong key is not saved", not os.path.exists(creds) or "sk-bad" not in open(creds).read(), t)
|
|
t.send("sk-good\r", 2.0)
|
|
check("switched to fakeb", "provider → fakeb" in screen(), t)
|
|
check("the good key is saved", os.path.exists(creds) and "fakeb.api_key=sk-good" in open(creds).read(), t)
|
|
check("no key on screen", "sk-good" not in screen() and "sk-bad" not in screen(), t)
|
|
t.send("/key\r", 0.8)
|
|
t.send("sk-good2\r", 2.0)
|
|
check("/key replaces it", "fakeb.api_key=sk-good2" in open(creds).read(), t)
|
|
t.close()
|
|
from keys import FAILS
|
|
print("FAILED:", len(FAILS))
|
|
sys.exit(min(len(FAILS), 100))
|