operator-main
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Robust sidechat management for muse-chat-api.py.
|
|
|
|
Provides:
|
|
- ensure_sidebar(ws): Opens sidebar if closed, with retry
|
|
- list_sidechats(ws): Returns list of side chat names, with retry
|
|
- All operations logged for audit
|
|
|
|
The sidebar UI is flaky — sometimes the open button isn't found,
|
|
sometimes content takes >2s to load. This module handles that.
|
|
"""
|
|
import time
|
|
import json
|
|
|
|
def _ev(ws, js, await_result=True):
|
|
"""Evaluate JS via CDP, return result.
|
|
Matches muse-chat-api.py's ev() exactly (id=1, returnByValue).
|
|
Fixed 2026-10-04: was using id=100, missing returnByValue.
|
|
"""
|
|
import json as _json
|
|
ws.send(_json.dumps({
|
|
"id": 1, "method": "Runtime.evaluate",
|
|
"params": {"expression": js, "returnByValue": True, "awaitPromise": await_result}
|
|
}))
|
|
resp = _json.loads(ws.recv())
|
|
result = resp.get("result", {}).get("result", {})
|
|
return result.get("value")
|
|
|
|
def ensure_sidebar(ws, max_retries=3):
|
|
"""
|
|
Ensure the sidebar is open. Returns True if open, False otherwise.
|
|
Retries with backoff if the open button isn't found immediately.
|
|
"""
|
|
for attempt in range(max_retries):
|
|
# Check if sidebar is already open (Side chats text visible)
|
|
is_open = _ev(ws, """(() => {
|
|
return document.body.innerText.includes('Side chats');
|
|
})()""")
|
|
if is_open:
|
|
return True
|
|
|
|
# Try to open it
|
|
_ev(ws, """(() => {
|
|
const btn = [...document.querySelectorAll('button')].find(b =>
|
|
(b.textContent||'').includes('Open chat and side chats')
|
|
);
|
|
if (btn) btn.click();
|
|
return !!btn;
|
|
})()""")
|
|
time.sleep(2 * (attempt + 1)) # Exponential backoff: 2s, 4s, 6s
|
|
|
|
# Final check
|
|
return _ev(ws, """(() => {
|
|
return document.body.innerText.includes('Side chats');
|
|
})()""")
|
|
|
|
def list_sidechats(ws):
|
|
"""
|
|
List side chat names. Returns list of strings.
|
|
Ensures sidebar is open first.
|
|
"""
|
|
if not ensure_sidebar(ws):
|
|
return []
|
|
|
|
result = _ev(ws, """(() => {
|
|
const text = document.body.innerText;
|
|
const idx = text.indexOf('Side chats');
|
|
if (idx === -1) return JSON.stringify([]);
|
|
// Extract chat names (heuristic: lines after "Side chats" that look like titles)
|
|
const section = text.slice(idx, idx + 2000);
|
|
const lines = section.split('\\n').map(l => l.trim()).filter(l => l.length > 0);
|
|
// Skip header, collect likely chat titles (skip timestamps like "1m", "2m")
|
|
const chats = [];
|
|
for (let i = 1; i < lines.length && chats.length < 20; i++) {
|
|
const line = lines[i];
|
|
if (/^\\d+[mh]$/.test(line)) continue; // timestamp
|
|
if (line === 'Unread updates') continue;
|
|
if (line.length > 5 && line.length < 80) chats.push(line);
|
|
}
|
|
return JSON.stringify(chats);
|
|
})()""")
|
|
try:
|
|
return json.loads(result) if result else []
|
|
except:
|
|
return []
|
|
|
|
def log_sidechat_op(agent, operation, details=""):
|
|
"""Log sidechat operations for audit."""
|
|
import os
|
|
from datetime import datetime, timezone
|
|
log_file = "/home/super/Projects/NetVM/sidechat-log.jsonl"
|
|
entry = {
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
"agent": agent,
|
|
"op": operation,
|
|
"details": details[:200]
|
|
}
|
|
try:
|
|
with open(log_file, "a") as f:
|
|
f.write(json.dumps(entry) + "\n")
|
|
except:
|
|
pass
|