import asyncio import sys import os from typing import Optional from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.zai_client import ZAiClient, ZAiResponse, MODEL_CAPABILITIES from core.model_executor import ModelExecutor, ExecutionResult from core.model_pool import SLOT_SUMMARY, ThinkingMode class HarnessCLI: """Interactive CLI for SAL-v4 Harness.""" def __init__(self, api_key: str): self.api_key = api_key self.executor = ModelExecutor(api_key) self.history: list = [] self.model = "glm-4.7-flash" self.thinking_enabled = False self.thinking_preserved = False self.running = True def banner(self): print("\n" + "=" * 60) print(" SAL-v4 Harness CLI - Z.AI GLM Interface") print("=" * 60) print(f" Model: {self.model}") print(f" Thinking: {'ENABLED' if self.thinking_enabled else 'disabled'}") print(f" Preserved: {'YES' if self.thinking_preserved else 'no'}") print("-" * 60) print(" Commands: /model, /thinking, /clear, /status, /quit") print("=" * 60 + "\n") def cmd_model(self, args: str): """Switch model.""" models = ["glm-4.7-flash", "glm-4.7-flashx", "glm-4.7", "glm-5"] if args in models: self.model = args print(f"[OK] Model set to: {args}") else: print(f"Available models: {', '.join(models)}") def cmd_thinking(self, args: str): """Toggle thinking mode.""" parts = args.split() if not parts: print(f"Thinking: {'ENABLED' if self.thinking_enabled else 'disabled'}") print(f"Preserved: {'YES' if self.thinking_preserved else 'no'}") print("Usage: /thinking [on|off] [preserve]") return if "on" in parts: self.thinking_enabled = True elif "off" in parts: self.thinking_enabled = False if "preserve" in parts: self.thinking_preserved = True self.thinking_enabled = True print( f"[OK] Thinking: {'ENABLED' if self.thinking_enabled else 'disabled'}, Preserved: {'YES' if self.thinking_preserved else 'no'}" ) def cmd_clear(self, args: str): """Clear history.""" self.history = [] print("[OK] History cleared") def cmd_status(self, args: str): """Show system status.""" print("\n--- System Status ---") print(f"Model: {self.model}") print(f"Thinking: {self.thinking_enabled}") print(f"Preserved: {self.thinking_preserved}") print(f"History turns: {len(self.history)}") print(SLOT_SUMMARY) def chat(self, message: str): """Send message to model.""" thinking_config = None if self.thinking_enabled: thinking_config = { "type": "enabled", "clear_thinking": not self.thinking_preserved, } messages = self.history.copy() messages.append({"role": "user", "content": message}) print(f"\n[{self.model}] ", end="", flush=True) try: with ZAiClient(self.api_key) as client: reasoning_parts = [] content_parts = [] for reasoning, content in client.chat_stream( model=self.model, messages=messages, thinking=thinking_config, max_tokens=2048, temperature=0.7, ): if reasoning: reasoning_parts.append(reasoning) if content: content_parts.append(content) print(content, end="", flush=True) print() full_reasoning = "".join(reasoning_parts) full_content = "".join(content_parts) if full_reasoning: print(f"\n[Thinking ({len(full_reasoning)} chars)]") preview = ( full_reasoning[:500] + "..." if len(full_reasoning) > 500 else full_reasoning ) print(f"\033[90m{preview}\033[0m") if self.thinking_preserved and full_reasoning: self.history.append({"role": "user", "content": message}) self.history.append( { "role": "assistant", "content": full_content, "reasoning_content": full_reasoning, } ) else: self.history.append({"role": "user", "content": message}) self.history.append({"role": "assistant", "content": full_content}) except Exception as e: print(f"\n[ERROR] {e}") def run(self): """Main loop.""" self.banner() while self.running: try: line = input("You> ").strip() if not line: continue if line.startswith("/"): parts = line[1:].split(maxsplit=1) cmd = parts[0].lower() args = parts[1] if len(parts) > 1 else "" if cmd in ("quit", "exit", "q"): print("Goodbye!") self.running = False elif cmd == "model": self.cmd_model(args) elif cmd == "thinking": self.cmd_thinking(args) elif cmd == "clear": self.cmd_clear(args) elif cmd == "status": self.cmd_status(args) elif cmd == "help": self.banner() else: print(f"Unknown command: /{cmd}") else: self.chat(line) except KeyboardInterrupt: print("\n[Interrupted]") except EOFError: break self.executor.close() def health_check(api_key: str) -> dict: """Quick health check.""" print("=== Z.AI Health Check ===\n") results = {} with ZAiClient(api_key, timeout=10.0) as client: models_to_test = ["glm-4.7-flash"] for model in models_to_test: try: print(f"Testing {model}...") start = datetime.now() response = client.chat_completion( model=model, messages=[{"role": "user", "content": "ping"}], max_tokens=10, ) latency = (datetime.now() - start).total_seconds() * 1000 results[model] = { "status": "OK", "latency_ms": int(latency), "model": response.model, } print(f" [OK] {latency:.0f}ms - {response.content[:50]}") except Exception as e: results[model] = {"status": "FAIL", "error": str(e)} print(f" [FAIL] {e}") return results if __name__ == "__main__": api_key = os.environ.get("ZAI_API_KEY") if not api_key: print("ERROR: ZAI_API_KEY not set") print("Usage: ZAI_API_KEY=your-key python interface.py") sys.exit(1) if len(sys.argv) > 1 and sys.argv[1] == "--health": health_check(api_key) else: cli = HarnessCLI(api_key) cli.run()