#!/usr/bin/env python3 """ SAL-v4 Harness Advanced TUI Split-screen interface with live updates. """ import os import sys import time import threading from datetime import datetime from typing import List, Dict from queue import Queue, Empty from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) ENV_FILE = Path(__file__).parent / ".env" if ENV_FILE.exists(): with open(ENV_FILE) as f: for line in f: line = line.strip() if line and "=" in line and not line.startswith("#"): key, val = line.split("=", 1) os.environ.setdefault(key, val) from rich.console import Console, Group from rich.panel import Panel from rich.layout import Layout from rich.live import Live from rich.text import Text from rich.table import Table from rich.progress import Progress, SpinnerColumn, TextColumn from core.zai_client import ZAiClient from core.salience import SalienceEngine from core.blackboard import Blackboard from core.scheduler import Scheduler from agents.router import Router from agents.tlp import TLPGateLoop, Phase class AgentSession: """Agent state manager.""" def __init__(self, api_key: str): self.api_key = api_key self.bb = Blackboard() self.engine = SalienceEngine() self.gov = Scheduler(self.bb, self.engine) self.router = Router(self.bb, self.gov) self.tlp = TLPGateLoop(self.bb) self.model = "glm-4.7-flash" self.thinking_enabled = True self.preserve_thinking = True self.show_thinking = True self.messages: List[Dict] = [] self.current_salience = 0.0 self.current_phase = Phase.RECEIVE self.tokens_used = 0 self.cost = 0.0 self.errors = 0 self.pending_input = "" self.is_processing = False self.status_message = "Ready" def chat(self, message: str) -> tuple: """Send message and get response.""" self.is_processing = True self.status_message = "Processing..." thinking_config = None if self.thinking_enabled: thinking_config = { "type": "enabled", "clear_thinking": not self.preserve_thinking, } msgs = [] for m in self.messages: if m["role"] == "user": msgs.append({"role": "user", "content": m["content"]}) else: msg_dict = {"role": "assistant", "content": m["content"]} if self.preserve_thinking and m.get("reasoning"): msg_dict["reasoning_content"] = m["reasoning"] msgs.append(msg_dict) msgs.append({"role": "user", "content": message}) try: reasoning_parts = [] content_parts = [] with ZAiClient(self.api_key, timeout=60.0) as client: for r, c in client.chat_stream( model=self.model, messages=msgs, thinking=thinking_config, max_tokens=2048, ): if r: reasoning_parts.append(r) if c: content_parts.append(c) full_reasoning = "".join(reasoning_parts) full_content = "".join(content_parts) self.messages.append( {"role": "user", "content": message, "time": datetime.now()} ) self.messages.append( { "role": "assistant", "content": full_content, "reasoning": full_reasoning, "time": datetime.now(), } ) self.tokens_used += 100 self.cost += 0.01 salience = self.engine.calculate_l_sal( delta_a=0.5 if len(full_reasoning) > 100 else 0.2, r=0.1, m=0.3, c=0.8, phi=0.05, t=len(self.messages) * 0.1, ) self.current_salience = salience self.tlp.step() self.current_phase = self.tlp.current_phase self.status_message = "Ready" self.is_processing = False return full_content, full_reasoning, None except Exception as e: self.errors += 1 self.status_message = f"Error: {e}" self.is_processing = False return "", "", str(e) def clear(self): self.messages = [] self.current_salience = 0.0 self.tlp = TLPGateLoop(self.bb) self.current_phase = Phase.RECEIVE self.status_message = "Cleared" class TUI: """Advanced TUI with split-screen layout.""" def __init__(self, api_key: str): self.console = Console() self.session = AgentSession(api_key) self.input_queue = Queue() self.running = True self.layout = Layout() self._setup_layout() def _setup_layout(self): self.layout.split( Layout(name="header", size=3), Layout(name="body"), Layout(name="input", size=3), ) self.layout["body"].split_row( Layout(name="chat", ratio=3), Layout(name="sidebar", ratio=1), ) self.layout["sidebar"].split( Layout(name="status", size=10), Layout(name="tlp", size=10), Layout(name="help"), ) def _make_header(self) -> Panel: t = Text() t.append("SAL-v4 ", style="bold cyan") t.append(f"[{self.session.model}] ", style="green") thinking = "THINK" if self.session.thinking_enabled else "DIRECT" preserve = "+PRESERVE" if self.session.preserve_thinking else "" t.append(f"[{thinking}{preserve}] ", style="yellow") status = self.session.status_message if self.session.is_processing: t.append(f"[{status}...]", style="yellow blink") else: t.append(f"[{status}]", style="dim") return Panel(t, style="on black", height=3) def _make_chat(self) -> Panel: lines = [] for msg in self.session.messages[-15:]: role = msg["role"] content = msg["content"] reasoning = msg.get("reasoning", "") time_str = ( msg["time"].strftime("%H:%M:%S") if isinstance(msg["time"], datetime) else "" ) if role == "user": lines.append(Text()) lines.append(Text(f"[{time_str}] You:", style="bold blue")) lines.append( Text(f" {content[:400]}{'...' if len(content) > 400 else ''}") ) else: if reasoning and self.session.show_thinking: lines.append(Text()) lines.append(Text(f"[{time_str}] Thinking:", style="dim italic")) r_preview = reasoning[:200] + ( "..." if len(reasoning) > 200 else "" ) lines.append(Text(f" {r_preview}", style="dim")) lines.append(Text()) lines.append(Text(f"[{time_str}] Agent:", style="bold green")) lines.append( Text(f" {content[:400]}{'...' if len(content) > 400 else ''}") ) if not lines: lines.append(Text("No messages yet. Type something to start.", style="dim")) return Panel(Group(*lines), title="Conversation", border_style="blue") def _make_status(self) -> Panel: table = Table.grid(padding=(0, 1)) table.add_column("k", style="dim", width=10) table.add_column("v", style="white", width=14) table.add_row("Model:", self.session.model) table.add_row("Thinking:", "ON" if self.session.thinking_enabled else "OFF") table.add_row("Preserve:", "YES" if self.session.preserve_thinking else "NO") table.add_row("Show:", "YES" if self.session.show_thinking else "NO") table.add_row("Salience:", f"{self.session.current_salience:.2f}") table.add_row("Tokens:", str(self.session.tokens_used)) table.add_row("Errors:", str(self.session.errors)) return Panel(table, title="Status", border_style="green") def _make_tlp(self) -> Panel: phases = [ ("RCV", Phase.RECEIVE), ("WRN", Phase.WARN), ("PLAN", Phase.TENTATIVE_PLAN), ("MOVE", Phase.START_MOVEMENT), ("REC", Phase.RECON), ("DONE", Phase.COMPLETE_PLAN), ("ORD", Phase.ISSUE_ORDER), ("REF", Phase.REFINE_SUPERVISE), ] t = Text() for i, (name, p) in enumerate(phases): if p == self.session.current_phase: t.append(f"[{name}]", style="bold yellow") elif p.value < self.session.current_phase.value: t.append(f"[{name}]", style="dim green") else: t.append(f"[{name}]", style="dim") if i < len(phases) - 1: t.append(" ") return Panel(t, title="TLP", border_style="cyan") def _make_help(self) -> Panel: help_text = "/thinking /preserve /show\n/model NAME /clear /quit" return Panel(help_text, title="Commands", border_style="dim") def _make_input(self) -> Panel: prompt = Text() prompt.append("Input> ", style="bold white") if self.session.pending_input: prompt.append(self.session.pending_input, style="white") else: prompt.append("(type message or /command)", style="dim") return Panel(prompt, style="on black", height=3) def render(self) -> Layout: self.layout["header"].update(self._make_header()) self.layout["chat"].update(self._make_chat()) self.layout["status"].update(self._make_status()) self.layout["tlp"].update(self._make_tlp()) self.layout["help"].update(self._make_help()) self.layout["input"].update(self._make_input()) return self.layout def handle_input(self, line: str): if line.startswith("/"): self._handle_command(line[1:]) else: self._handle_chat(line) def _handle_command(self, cmd_line: str): parts = cmd_line.split(maxsplit=1) cmd = parts[0].lower() args = parts[1] if len(parts) > 1 else "" if cmd in ("quit", "exit", "q"): self.running = False self.session.status_message = "Exiting..." elif cmd == "thinking": self.session.thinking_enabled = not self.session.thinking_enabled self.session.status_message = ( f"Thinking: {'ON' if self.session.thinking_enabled else 'OFF'}" ) elif cmd == "preserve": self.session.preserve_thinking = not self.session.preserve_thinking self.session.status_message = ( f"Preserve: {'YES' if self.session.preserve_thinking else 'NO'}" ) elif cmd == "show": self.session.show_thinking = not self.session.show_thinking self.session.status_message = ( f"Show thinking: {'YES' if self.session.show_thinking else 'NO'}" ) elif cmd == "model": models = ["glm-4.7-flash", "glm-4.7-flashx", "glm-4.7", "glm-5"] if args in models: self.session.model = args self.session.status_message = f"Model: {args}" else: self.session.status_message = f"Models: {', '.join(models)}" elif cmd == "clear": self.session.clear() elif cmd == "help": self.session.status_message = ( "Commands: /thinking /preserve /show /model /clear /quit" ) else: self.session.status_message = f"Unknown: /{cmd}" def _handle_chat(self, message: str): def process(): content, reasoning, error = self.session.chat(message) if error: self.session.status_message = f"Error: {error[:50]}" thread = threading.Thread(target=process, daemon=True) thread.start() def run(self): def input_reader(): while self.running: try: line = input() self.input_queue.put(line) except EOFError: self.running = False break reader = threading.Thread(target=input_reader, daemon=True) reader.start() with Live(self.render(), console=self.console, refresh_per_second=4) as live: while self.running: try: line = self.input_queue.get(timeout=0.25) self.handle_input(line) except Empty: pass live.update(self.render()) self.console.print("\n[yellow]Goodbye![/yellow]") def 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 tui.py") sys.exit(1) tui = TUI(api_key) tui.run() if __name__ == "__main__": main()