#!/usr/bin/env python3
"""
Mortdecai Chat App — interactive dev console for testing gateway and model calls.
Features:
- Mode switch: Gateway call (full pipeline) vs Model call (direct Ollama)
- Template selector powered by training/prompts/ JSONL files
- Editable JSON payload with live trace log (SSE streaming)
- Training data export — save interactions for audit/training
- Branding: Rajdhani Bold, Sethian orange (#D35400), dark theme
Usage:
python3 web/chat_app.py --port 8097
Deploy on CT 650 @ pve112, serve as chat.api.mortdec.ai behind Caddy + Google OAuth.
"""
import argparse
import json
import os
import sys
import time
import threading
import traceback
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from urllib.parse import urlparse, parse_qs
from datetime import datetime
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
# Lazy imports — agent modules may not be on every machine
def _get_assistant(config):
from agent.serve import MinecraftAssistant
return MinecraftAssistant(config)
def _get_prompts_module():
from training.prompts.loader import load_manifest, load_prompts, load_prompt_entries, expand_template, TEMPLATE_VARS
return load_manifest, load_prompts, load_prompt_entries, expand_template, TEMPLATE_VARS
PORT = 8097
LOG_DIR = PROJECT_ROOT / "data" / "chat_logs"
TRAINING_LOG = LOG_DIR / "training_export.jsonl"
# Default call configs
DEFAULT_GATEWAY_CONFIG = {
"mode": "sudo",
"player": "slingshooter08",
"query": "",
"ollama_url": "http://192.168.0.141:11434",
"model": "mortdecai:0.5.0-f16",
"rcon_host": "192.168.0.244",
"rcon_port": 25578,
"rcon_password": "REDACTED_RCON",
"temperature": 0.7,
"max_tokens": 800,
}
DEFAULT_MODEL_CONFIG = {
"model": "mortdecai:0.5.0-f16",
"ollama_url": "http://192.168.0.141:11434",
"messages": [
{"role": "system", "content": "You are a Minecraft 1.21 command translator for a Paper server.\nPlugins: FastAsyncWorldEdit, WorldGuard, CoreProtect, EssentialsX, Vault, LuckPerms.\nReturn JSON: {\"commands\": [...], \"reasoning\": \"...\", \"message\": \"...\"}\nUse /no_think mode."},
{"role": "user", "content": "Player slingshooter08: sudo give me diamond armor"}
],
"temperature": 0.7,
"max_tokens": 800,
"format": "json",
}
# ── HTML / CSS / JS ──────────────────────────────────────────────────────────
HTML_PAGE = r"""
Mortdecai Chat
MORTDECAICHAT
Call
Mode
Template
Model: —
Last: —
Cmds: —
Session: 0 calls
chat.api.mortdec.ai
"""
# ── HTTP Handler ──────────────────────────────────────────────────────────────
class ChatHandler(BaseHTTPRequestHandler):
"""HTTP handler for the chat app."""
def log_message(self, fmt, *args):
"""Suppress default request logging."""
pass
def _cors_headers(self):
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
def _json_response(self, data, status=200):
self.send_response(status)
self.send_header('Content-Type', 'application/json')
self._cors_headers()
self.end_headers()
self.wfile.write(json.dumps(data).encode())
def _html_response(self, html):
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode())
def _sse_start(self):
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self._cors_headers()
self.end_headers()
def _sse_event(self, data: dict):
msg = f"data: {json.dumps(data)}\n\n"
self.wfile.write(msg.encode())
self.wfile.flush()
def do_OPTIONS(self):
self.send_response(204)
self._cors_headers()
self.end_headers()
def do_GET(self):
path = urlparse(self.path).path
if path in ('/', '/index.html'):
defaults_json = json.dumps({
'gateway': DEFAULT_GATEWAY_CONFIG,
'model': DEFAULT_MODEL_CONFIG,
})
html = HTML_PAGE.replace('__DEFAULTS_JSON__', defaults_json)
self._html_response(html)
elif path == '/api/manifest':
try:
load_manifest, *_ = _get_prompts_module()
self._json_response(load_manifest())
except Exception as e:
self._json_response({'error': str(e)}, 500)
elif path.startswith('/api/prompts/'):
category = path.split('/api/prompts/', 1)[1]
try:
*_, load_prompt_entries, _, _ = _get_prompts_module()
entries = load_prompt_entries(category)
self._json_response(entries)
except Exception as e:
self._json_response({'error': str(e)}, 500)
elif path == '/api/template-vars':
try:
*_, TEMPLATE_VARS = _get_prompts_module()
self._json_response(TEMPLATE_VARS)
except Exception as e:
self._json_response({'error': str(e)}, 500)
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
path = urlparse(self.path).path
content_len = int(self.headers.get('Content-Length', 0))
body = json.loads(self.rfile.read(content_len)) if content_len else {}
if path == '/api/send':
self._handle_send(body)
elif path == '/api/save-training':
self._handle_save_training(body)
else:
self.send_response(404)
self.end_headers()
def _handle_send(self, body):
"""Handle a gateway or model call with SSE trace streaming."""
call_type = body.get('call_type', 'gateway')
payload = body.get('payload', {})
self._sse_start()
try:
if call_type == 'gateway':
self._run_gateway_call(payload)
else:
self._run_model_call(payload)
except Exception as e:
self._sse_event({'type': 'error', 'text': f'Exception: {e}'})
traceback.print_exc()
def _run_gateway_call(self, payload):
"""Execute a full gateway call with trace events."""
import requests as req
mode = payload.get('mode', 'sudo')
player = payload.get('player', 'slingshooter08')
query = payload.get('query', '')
ollama_url = payload.get('ollama_url', DEFAULT_GATEWAY_CONFIG['ollama_url'])
model = payload.get('model', DEFAULT_GATEWAY_CONFIG['model'])
rcon_host = payload.get('rcon_host', DEFAULT_GATEWAY_CONFIG['rcon_host'])
rcon_port = payload.get('rcon_port', DEFAULT_GATEWAY_CONFIG['rcon_port'])
rcon_pass = payload.get('rcon_password', DEFAULT_GATEWAY_CONFIG['rcon_password'])
temp = payload.get('temperature', 0.7)
max_tokens = payload.get('max_tokens', 800)
self._sse_event({'type': 'system', 'text': f'Gateway call: mode={mode} player={player} model={model}'})
self._sse_event({'type': 'system', 'text': f'Query: {query}', 'model': model})
start = time.time()
# 1. Context gathering via RCON
self._sse_event({'type': 'context', 'text': f'Gathering context from {rcon_host}:{rcon_port}...'})
context_parts = []
try:
from agent.tools.persistent_rcon import get_rcon
rcon = get_rcon(rcon_host, rcon_port, rcon_pass)
# Player info
player_list = rcon.command('list')
context_parts.append(f'Online: {player_list}')
self._sse_event({'type': 'context', 'text': f'Server: {player_list}'})
if player:
pos_result = rcon.command(f'data get entity {player} Pos')
context_parts.append(f'Position: {pos_result}')
self._sse_event({'type': 'context', 'text': f'Player pos: {pos_result[:100]}'})
except Exception as e:
self._sse_event({'type': 'context', 'text': f'RCON context failed: {e} (continuing without context)'})
ctx_ms = int((time.time() - start) * 1000)
self._sse_event({'type': 'context', 'text': f'Context gathered in {ctx_ms}ms'})
# 2. Build system prompt
try:
from agent.prompts.system_prompts import get_prompt
system_prompt = get_prompt(mode)
except ImportError:
system_prompt = f"You are a Minecraft 1.21 command translator. Mode: {mode}."
context_str = '\n'.join(context_parts) if context_parts else 'No context available'
user_message = f"Request from {player}: {query}\n\nContext:\n{context_str}"
messages = [
{'role': 'system', 'content': '/no_think\n' + system_prompt},
{'role': 'user', 'content': user_message},
]
# 3. LLM call
ollama_payload = {
'model': model,
'messages': messages,
'stream': False,
'format': 'json',
'options': {'temperature': temp, 'num_predict': max_tokens},
}
self._sse_event({
'type': 'llm',
'text': f'Calling {model} on {ollama_url}...',
'api_request': {'url': f'{ollama_url}/api/chat', 'method': 'POST', 'body': ollama_payload},
})
llm_start = time.time()
try:
import re
r = req.post(f"{ollama_url}/api/chat", json=ollama_payload, timeout=120)
r.raise_for_status()
api_response = r.json()
raw_content = api_response['message']['content']
# Strip thinking tags
raw_content = re.sub(r'[\s\S]*?\s*', '', raw_content)
llm_ms = int((time.time() - llm_start) * 1000)
self._sse_event({
'type': 'llm',
'text': f'LLM responded in {llm_ms}ms',
'api_response': {'raw': raw_content, 'model': api_response.get('model'), 'eval_count': api_response.get('eval_count'), 'eval_duration': api_response.get('eval_duration')},
})
try:
parsed = json.loads(raw_content)
except json.JSONDecodeError:
parsed = {'commands': [], 'message': raw_content, 'reasoning': 'parse failed'}
self._sse_event({'type': 'error', 'text': f'JSON parse failed, raw: {raw_content[:200]}'})
commands = parsed.get('commands', [])
message = parsed.get('message', '')
reasoning = parsed.get('reasoning', '')
if reasoning:
self._sse_event({'type': 'llm', 'text': f'Reasoning: {reasoning}'})
if message:
self._sse_event({'type': 'llm', 'text': f'Message: {message}'})
self._sse_event({'type': 'llm', 'text': f'Commands ({len(commands)}): {json.dumps(commands)}', 'parsed': parsed})
except Exception as e:
self._sse_event({'type': 'error', 'text': f'LLM call failed: {e}'})
commands = []
message = ''
reasoning = ''
raw_content = ''
llm_ms = 0
# 4. Guardrails
self._sse_event({'type': 'guard', 'text': f'Running guardrails on {len(commands)} commands...'})
guardrail_results = []
safe_commands = []
try:
from agent.guardrails.command_filter import filter_commands
safe_commands, guardrail_results = filter_commands(commands)
blocked = len(commands) - len(safe_commands)
if blocked:
self._sse_event({'type': 'guard', 'text': f'Blocked {blocked} commands'})
for gr in guardrail_results:
if gr.get('warnings'):
self._sse_event({'type': 'guard', 'text': f" {gr['command']}: {', '.join(gr['warnings'])}"})
except ImportError:
safe_commands = commands
self._sse_event({'type': 'guard', 'text': 'Guardrails not available, passing all commands'})
self._sse_event({'type': 'guard', 'text': f'Passed: {len(safe_commands)} commands'})
# 5. Execute via RCON
rcon_results = []
if safe_commands:
self._sse_event({'type': 'rcon', 'text': f'Executing {len(safe_commands)} commands via RCON...'})
try:
rcon = get_rcon(rcon_host, rcon_port, rcon_pass)
for cmd in safe_commands[:12]:
if not isinstance(cmd, str) or not cmd.strip():
continue
try:
result = rcon.command(cmd)
is_error = any(e in result for e in ('<--[HERE]', 'Unknown', 'Incorrect', 'Expected'))
rcon_entry = {'cmd': cmd, 'result': result[:200], 'ok': not is_error}
rcon_results.append(rcon_entry)
trace_type = 'rcon' if not is_error else 'rcon-fail'
self._sse_event({'type': trace_type, 'text': f'{cmd} -> {result[:120]}',
'rcon_command': cmd, 'rcon_result': result, 'rcon_ok': not is_error})
except Exception as e:
rcon_results.append({'cmd': cmd, 'result': str(e), 'ok': False})
self._sse_event({'type': 'rcon-fail', 'text': f'{cmd} -> ERROR: {e}',
'rcon_command': cmd, 'rcon_error': str(e)})
except Exception as e:
self._sse_event({'type': 'error', 'text': f'RCON execution failed: {e}'})
# 6. Final result
total_ms = int((time.time() - start) * 1000)
ok_count = sum(1 for r in rcon_results if r.get('ok'))
fail_count = sum(1 for r in rcon_results if not r.get('ok'))
full_result = {
'timestamp': datetime.utcnow().isoformat() + 'Z',
'source': 'chat_app',
'mode': mode,
'player': player,
'input': {
'user_message': (('pray ' if mode == 'god' else 'sudo ') + query) if query else '',
'server_context': {'server_type': 'paper', 'version': '1.21.x'},
},
'output': {
'commands_generated': commands,
'commands_executed': safe_commands,
'message': message,
'reasoning': reasoning,
},
'rcon_results': rcon_results,
'metadata': {
'model': model,
'ollama_url': ollama_url,
'temperature': temp,
'duration_ms': total_ms,
'context_ms': ctx_ms,
'llm_ms': llm_ms,
},
}
self._sse_event({
'type': 'result',
'text': f'Complete: {total_ms}ms total, {ok_count} ok / {fail_count} fail',
'cmd_count': f'{ok_count}/{ok_count + fail_count}',
'model': model,
'full_result': full_result,
})
def _run_model_call(self, payload):
"""Execute a direct model call (bypass gateway pipeline)."""
import requests as req
model = payload.get('model', DEFAULT_MODEL_CONFIG['model'])
ollama_url = payload.get('ollama_url', DEFAULT_MODEL_CONFIG['ollama_url'])
messages = payload.get('messages', DEFAULT_MODEL_CONFIG['messages'])
temp = payload.get('temperature', 0.7)
max_tokens = payload.get('max_tokens', 800)
fmt = payload.get('format', 'json')
self._sse_event({'type': 'system', 'text': f'Direct model call: {model} on {ollama_url}', 'model': model})
self._sse_event({'type': 'llm', 'text': f'Sending {len(messages)} messages, temp={temp}'})
start = time.time()
try:
import re
api_payload = {
'model': model,
'messages': messages,
'stream': False,
'options': {'temperature': temp, 'num_predict': max_tokens},
}
if fmt:
api_payload['format'] = fmt
r = req.post(f"{ollama_url}/api/chat", json=api_payload, timeout=120)
r.raise_for_status()
raw_content = r.json()['message']['content']
raw_content = re.sub(r'[\s\S]*?\s*', '', raw_content)
llm_ms = int((time.time() - start) * 1000)
self._sse_event({'type': 'llm', 'text': f'Response in {llm_ms}ms'})
try:
parsed = json.loads(raw_content)
commands = parsed.get('commands', [])
message = parsed.get('message', '')
reasoning = parsed.get('reasoning', '')
if reasoning:
self._sse_event({'type': 'llm', 'text': f'Reasoning: {reasoning}'})
if message:
self._sse_event({'type': 'llm', 'text': f'Message: {message}'})
self._sse_event({'type': 'llm', 'text': f'Commands ({len(commands)}): {json.dumps(commands)}',
'cmd_count': str(len(commands))})
except json.JSONDecodeError:
self._sse_event({'type': 'llm', 'text': f'Raw output: {raw_content[:500]}'})
commands = []
message = raw_content
reasoning = ''
full_result = {
'timestamp': datetime.utcnow().isoformat() + 'Z',
'source': 'chat_app_model',
'input': {'messages': messages},
'output': {
'commands_generated': commands,
'message': message,
'reasoning': reasoning,
'raw': raw_content,
},
'metadata': {
'model': model,
'ollama_url': ollama_url,
'temperature': temp,
'duration_ms': llm_ms,
},
}
self._sse_event({
'type': 'result',
'text': f'Complete: {llm_ms}ms, {len(commands)} commands',
'model': model,
'full_result': full_result,
})
except Exception as e:
self._sse_event({'type': 'error', 'text': f'Model call failed: {e}'})
def _handle_save_training(self, body):
"""Save an interaction to the training export log."""
LOG_DIR.mkdir(parents=True, exist_ok=True)
with open(TRAINING_LOG, 'a') as f:
f.write(json.dumps(body) + '\n')
self._json_response({'ok': True, 'file': str(TRAINING_LOG.name)})
# ── Main ──────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description="Mortdecai Chat App")
parser.add_argument("--port", type=int, default=PORT)
parser.add_argument("--host", default="0.0.0.0")
args = parser.parse_args()
LOG_DIR.mkdir(parents=True, exist_ok=True)
server = HTTPServer((args.host, args.port), ChatHandler)
print(f"Mortdecai Chat App")
print(f" http://{args.host}:{args.port}")
print(f" Prompts: {PROJECT_ROOT / 'training' / 'prompts'}")
print(f" Training log: {TRAINING_LOG}")
print()
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nShutdown.")
server.server_close()
if __name__ == "__main__":
main()