ollie/kde/gui/ollie9pclient.cpp

721 lines
24 KiB
C++

#include "ollie9pclient.h"
#include <QDir>
#include <QSet>
#include <QStandardPaths>
#include <QJsonDocument>
#include <QJsonObject>
#include <QDebug>
Ollie9pClient::Ollie9pClient(QObject *parent)
: QObject(parent)
{
// Initialize native 9P client
m_9p = new Lib9pClient(this);
m_9p->connectDefault();
// Chat stream — guarded, reads agent chat log. Runs continuously while
// an agent is selected; stops on agent switch or daemon disconnect.
m_chat = new NativeStreamer(NativeStreamer::Guarded, this);
m_chat->setGuard([this]() {
return m_daemonConnected && !m_activeSessionId.isEmpty() && !m_agentId.isEmpty();
});
connect(m_chat, &NativeStreamer::dataReady, this, [this](const QByteArray &data) {
if (!data.isEmpty())
emit chatReceived(QString::fromUtf8(data));
});
// State stream — guarded, reads agent statewait. Auto-restarts until
// session/agent is deselected or daemon disconnects (guard returns false).
m_state = new NativeStreamer(NativeStreamer::Guarded, this);
m_state->setGuard([this]() {
return m_daemonConnected && !m_activeSessionId.isEmpty() && !m_agentId.isEmpty();
});
connect(m_state, &NativeStreamer::dataReady, this, [this](const QByteArray &data) {
QString state = QString::fromUtf8(data).trimmed();
if (!state.isEmpty()) {
m_activeState = state;
emit activeStateChanged();
}
});
// Event stream — looping, reads server eventwait for session/agent events.
m_event = new NativeStreamer(NativeStreamer::Looping, this);
connect(m_event, &NativeStreamer::dataReady, this, [this](const QByteArray &data) {
// Events may come in batches separated by newlines
QString text = QString::fromUtf8(data);
for (const QString &line : text.split('\n', Qt::SkipEmptyParts)) {
handleEvent(line.trimmed());
}
});
// Heartbeat timer — detects daemon death
m_heartbeatTimer = new QTimer(this);
m_heartbeatTimer->setInterval(2500); // 2.5 seconds
connect(m_heartbeatTimer, &QTimer::timeout, this, &Ollie9pClient::heartbeat);
m_heartbeatTimer->start();
// Initial connection check and data load
heartbeat();
if (m_daemonConnected) {
refreshSessions();
ensureRootDataLoaded();
startEventStream();
}
}
Ollie9pClient::~Ollie9pClient()
{
if (m_heartbeatTimer) m_heartbeatTimer->stop();
stopAgentStreams();
if (m_event) m_event->stop();
}
void Ollie9pClient::setActiveSessionId(const QString &id)
{
qDebug() << "setActiveSessionId" << id << "prev session" << m_activeSessionId << "prev agent" << m_agentId;
// Session selection must be side-effect free for per-agent streams.
// Even re-selecting the same session should clear agent selection and stop
// statewait/chat so the UI can stay in a pure "session selected" state.
stopStreams();
m_activeSessionId = id;
m_agentId.clear();
m_activeState = "idle";
if (id.isEmpty()) {
emit activeSessionIdChanged();
emit activeStateChanged();
return;
}
emit activeSessionIdChanged();
emit activeAgentIdChanged();
emit activeStateChanged();
}
QString Ollie9pClient::agentKey(const QString &sessionId, const QString &agentId) const
{
return sessionId + "\n" + agentId;
}
void Ollie9pClient::setDaemonConnected(bool connected)
{
if (m_daemonConnected == connected) return;
m_daemonConnected = connected;
emit daemonConnectedChanged();
emit sessionsChanged();
}
void Ollie9pClient::heartbeat()
{
// Try to read a cheap file to verify daemon is alive
if (!m_9p->isConnected()) {
m_9p->connectDefault();
}
QByteArray result = m_9p->read(QStringLiteral("help"));
bool alive = !result.isEmpty();
if (alive && !m_daemonConnected) {
// Daemon came up — refresh state and start event stream
setDaemonConnected(true);
refreshSessions();
ensureRootDataLoaded();
startEventStream();
if (!m_activeSessionId.isEmpty() && !m_agentId.isEmpty()) {
startActiveAgentStreams();
}
} else if (!alive && m_daemonConnected) {
// Daemon died — clear state and stop streams
setDaemonConnected(false);
stopAgentStreams();
if (m_event) m_event->stop();
m_sessions.clear();
m_activeSessionId.clear();
m_agentId.clear();
m_activeState = "idle";
emit sessionsChanged();
emit activeSessionIdChanged();
emit activeAgentIdChanged();
emit activeStateChanged();
m_9p->disconnect();
}
}
void Ollie9pClient::startEventStream()
{
if (!m_daemonConnected) return;
if (m_event->isRunning()) return;
m_event->start(QStringLiteral("eventwait"));
}
QString Ollie9pClient::sessionConnectionColor(const QString &sessionId) const
{
for (const QVariant &v : m_sessions) {
const QVariantMap s = v.toMap();
if (s.value("id").toString() != sessionId) continue;
const bool paused = s.value("paused").toBool();
const bool connected = s.value("connected").toBool();
// Color logic:
// - Gray: paused (intentionally stopped)
// - Green: not paused AND connected
// - Red: not paused AND disconnected (crashed/broken)
// - Yellow: paused BUT still connected (zombie - shouldn't happen)
if (paused && connected) return "yellow"; // zombie
if (paused) return "gray"; // intentionally paused
if (connected) return "green"; // healthy
return "red"; // crashed/broken
}
return "gray";
}
// handleEvent processes a single event from eventwait.
// Event format: "topic payload" where topic is "session.{sid}.{action}" or
// "session.{sid}.agent.{aid}.{action}"
// Optimized for the hot path: agent.state changes
void Ollie9pClient::handleEvent(const QString &eventLine)
{
if (eventLine.isEmpty()) return;
// Split topic and payload (avoid QString allocation where possible)
const int spaceIdx = eventLine.indexOf(QLatin1Char(' '));
const QStringView topic = spaceIdx > 0
? QStringView(eventLine).left(spaceIdx)
: QStringView(eventLine);
const QStringView payload = spaceIdx > 0
? QStringView(eventLine).mid(spaceIdx + 1)
: QStringView();
// Fast path: must start with "session."
if (!topic.startsWith(QLatin1String("session."))) return;
// Parse: session.{sid}.{rest...}
const int firstDot = 7; // length of "session"
const int secondDot = topic.indexOf(QLatin1Char('.'), firstDot + 1);
if (secondDot < 0) return;
const QStringView sessionId = topic.mid(firstDot + 1, secondDot - firstDot - 1);
const QStringView rest = topic.mid(secondDot + 1);
// Check for agent events: "agent.{aid}.{action}"
if (rest.startsWith(QLatin1String("agent."))) {
// Parse agent.{aid}.{action}
const int agentDot = 5; // length of "agent"
const int aidEnd = rest.indexOf(QLatin1Char('.'), agentDot + 1);
if (aidEnd < 0) return;
const QStringView agentId = rest.mid(agentDot + 1, aidEnd - agentDot - 1);
const QStringView agentAction = rest.mid(aidEnd + 1);
if (agentAction == QLatin1String("state")) {
// Hot path: agent state change
if (payload.isEmpty()) return;
// Fast path: check if this is the active agent first (avoids toString for non-active)
const bool isActiveAgent = (sessionId == m_activeSessionId && agentId == m_agentId);
if (isActiveAgent) {
// Super fast path: compare payload directly against current active state
const QString payloadStr = payload.toString();
if (m_activeState == payloadStr) return; // No change
// State changed for active agent
m_activeState = payloadStr;
const QString sidStr = sessionId.toString();
const QString aidStr = agentId.toString();
const QString key = agentKey(sidStr, aidStr);
m_agentStateValues[key] = payloadStr;
emit agentStateChanged(sidStr, aidStr, payloadStr);
emit activeStateChanged();
} else {
// Non-active agent: need to check cache
const QString sidStr = sessionId.toString();
const QString aidStr = agentId.toString();
const QString key = agentKey(sidStr, aidStr);
const QString payloadStr = payload.toString();
// Use iterator to avoid double lookup
auto it = m_agentStateValues.find(key);
if (it != m_agentStateValues.end() && it.value() == payloadStr) return;
m_agentStateValues[key] = payloadStr;
emit agentStateChanged(sidStr, aidStr, payloadStr);
}
} else if (agentAction == QLatin1String("new") || agentAction == QLatin1String("kill")) {
refreshSessions();
}
} else {
// Session-level event
if (rest == QLatin1String("pause") || rest == QLatin1String("resume") ||
rest == QLatin1String("new") || rest == QLatin1String("kill") ||
rest == QLatin1String("rename")) {
refreshSessions();
}
}
}
QString Ollie9pClient::agentState(const QString &sessionId, const QString &agentId) const
{
return m_agentStateValues.value(agentKey(sessionId, agentId), "idle");
}
void Ollie9pClient::refreshSessions()
{
if (!m_daemonConnected) return;
if (!m_9p || !m_9p->isConnected()) return;
QByteArray out = m_9p->read(QStringLiteral("session/idx"));
QString raw = QString::fromUtf8(out);
m_sessions.clear();
// Track if active session/agent still exist during parsing
bool activeSessionFound = false;
bool activeAgentFound = false;
if (!raw.isEmpty()) {
for (const QString &line : raw.split('\n', Qt::SkipEmptyParts)) {
QStringList parts = line.split('\t');
// New format: name\tid\tpaused\tconnected\tcwd\tbackend\tmodel\tagents
// Where agents is semicolon-separated: name:id:state;name:id:state
if (parts.size() < 2) continue;
const QString sessionId = parts.value(1);
const bool isConnected = (parts.value(3) == "true");
const bool isActiveSession = (sessionId == m_activeSessionId);
if (isActiveSession) {
activeSessionFound = true;
}
QVariantMap session;
session["name"] = parts.value(0);
session["id"] = sessionId;
session["paused"] = (parts.value(2) == "true");
session["connected"] = isConnected;
session["cwd"] = parts.value(4);
session["backend"] = parts.value(5);
session["model"] = parts.value(6);
// Parse agents from semicolon-separated list
QVariantList agents;
QString agentStr = parts.value(7);
if (!agentStr.isEmpty()) {
for (const QString &agentEntry : agentStr.split(';', Qt::SkipEmptyParts)) {
QStringList agentParts = agentEntry.split(':');
if (agentParts.size() >= 3) {
const QString agentId = agentParts.value(1);
QVariantMap av;
av["name"] = agentParts.value(0);
av["id"] = agentId;
av["state"] = agentParts.value(2);
agents.append(av);
// Check if this is the active agent
if (isActiveSession && agentId == m_agentId) {
activeAgentFound = true;
}
}
}
}
session["agents"] = agents;
m_sessions.append(session);
}
}
emit sessionsChanged();
// Handle active session/agent removal
if (!m_activeSessionId.isEmpty()) {
if (!activeSessionFound) {
// Session was killed — clear both and stop streams
stopStreams();
m_activeSessionId.clear();
m_agentId.clear();
m_activeState = "idle";
emit activeSessionIdChanged();
emit activeAgentIdChanged();
emit activeStateChanged();
} else if (!m_agentId.isEmpty() && !activeAgentFound) {
// Agent was killed but session still exists — clear agent only
stopStreams();
m_agentId.clear();
m_activeState = "idle";
emit activeAgentIdChanged();
emit activeStateChanged();
}
}
// Auto-select an agent for the active session when none is selected.
// This also handles the first refresh after a session was selected before
// its agent list arrived.
if (!m_activeSessionId.isEmpty() && m_agentId.isEmpty()) {
for (const QVariant &value : m_sessions) {
const QVariantMap session = value.toMap();
if (session.value("id").toString() != m_activeSessionId) continue;
const QVariantList agents = session.value("agents").toList();
if (!agents.isEmpty()) {
const QString aid = agents.first().toMap().value("id").toString();
if (!aid.isEmpty()) switchAgent(m_activeSessionId, aid);
}
break;
}
}
// Auto-select first session and its first agent if none active
if (m_activeSessionId.isEmpty() && !m_sessions.isEmpty()) {
const QVariantMap firstSession = m_sessions.first().toMap();
const QString sid = firstSession.value("id").toString();
const QVariantList agents = firstSession.value("agents").toList();
if (!agents.isEmpty()) {
// Select first agent of first session
const QString aid = agents.first().toMap().value("id").toString();
switchAgent(sid, aid);
} else {
// No agents - just select the session
setActiveSessionId(sid);
}
}
}
QString Ollie9pClient::readLog()
{
if (m_activeSessionId.isEmpty()) return {};
if (!m_9p || !m_9p->isConnected()) return {};
return QString::fromUtf8(m_9p->read(agentPath() + "/log"));
}
QString Ollie9pClient::readLogForSession(const QString &sessionId, const QString &agentId)
{
if (sessionId.isEmpty() || agentId.isEmpty()) return {};
if (!m_9p || !m_9p->isConnected()) return {};
// Use immutable IDs — the 9P namespace resolves them via aliases.
QString path = "session/" + sessionId + "/agent/" + agentId + "/log";
return QString::fromUtf8(m_9p->read(path));
}
bool Ollie9pClient::submit(const QString &prompt)
{
if (m_activeSessionId.isEmpty() || m_agentId.isEmpty() || prompt.trimmed().isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
return m_9p->write(agentPath() + "/prompt", prompt.toUtf8());
}
bool Ollie9pClient::interrupt()
{
if (m_activeSessionId.isEmpty() || m_agentId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
return m_9p->write(agentPath() + "/ctl", "stop");
}
bool Ollie9pClient::kill()
{
if (m_activeSessionId.isEmpty() || m_agentId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
return m_9p->write(agentPath() + "/ctl", "kill");
}
bool Ollie9pClient::killSession(const QString &sessionId)
{
if (sessionId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
// If killing the active session, shut down its streams and deselect.
if (sessionId == m_activeSessionId) {
stopStreams();
m_activeSessionId.clear();
m_agentId.clear();
emit activeSessionIdChanged();
emit activeAgentIdChanged();
}
// Use immutable ID — the 9P namespace resolves it via alias.
QString path = "session/" + sessionId + "/ctl";
return m_9p->write(path, "kill");
}
bool Ollie9pClient::pauseSession(const QString &sessionId)
{
if (sessionId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
QString path = "session/" + sessionId + "/ctl";
if (m_9p->write(path, "pause")) {
refreshSessions();
return true;
}
return false;
}
bool Ollie9pClient::resumeSession(const QString &sessionId)
{
if (sessionId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
QString path = "session/" + sessionId + "/ctl";
if (m_9p->write(path, "resume")) {
refreshSessions();
return true;
}
return false;
}
bool Ollie9pClient::isSessionPaused(const QString &sessionId) const
{
for (const QVariant &v : m_sessions) {
const QVariantMap s = v.toMap();
if (s.value("id").toString() == sessionId)
return s.value("paused").toBool();
}
return false;
}
QString Ollie9pClient::getConfig()
{
if (m_activeSessionId.isEmpty()) return {};
if (!m_9p || !m_9p->isConnected()) return {};
return QString::fromUtf8(m_9p->read(agentPath() + "/cfg"));
}
QStringList Ollie9pClient::getAgents(const QString &sessionId)
{
if (sessionId.isEmpty()) return {};
if (!m_9p || !m_9p->isConnected()) return {};
// Use immutable ID — the 9P namespace resolves it via alias.
QString path = "session/" + sessionId + "/agent";
QStringList all = m_9p->ls(path);
QStringList agents;
for (const QString &a : all) {
QString name = a.trimmed();
// "new" is the agent creation file, not an actual agent
if (name != "new")
agents.append(name);
}
return agents;
}
void Ollie9pClient::setActiveAgentId(const QString &agentId)
{
if (m_agentId == agentId) return;
stopStreams();
m_agentId = agentId;
m_activeState = "idle";
emit activeAgentIdChanged();
emit activeStateChanged();
}
void Ollie9pClient::switchAgent(const QString &sessionId, const QString &agentId)
{
qDebug() << "switchAgent" << sessionId << agentId << "prev session" << m_activeSessionId << "prev agent" << m_agentId;
if (m_activeSessionId == sessionId && m_agentId == agentId) return;
stopStreams();
m_activeSessionId = sessionId;
m_agentId = agentId;
// Use cached state from live tracking if available, otherwise default to idle
m_activeState = m_agentStateValues.value(agentKey(sessionId, agentId), "idle");
emit activeSessionIdChanged();
emit activeAgentIdChanged();
emit activeStateChanged();
startActiveAgentStreams();
}
void Ollie9pClient::loadRootBackends()
{
if (m_rootBackendsLoaded) return;
if (!m_9p || !m_9p->isConnected()) return;
QByteArray out = m_9p->read(QStringLiteral("backends"));
QString raw = QString::fromUtf8(out).trimmed();
if (!raw.isEmpty()) {
m_availableBackends = raw.split('\n', Qt::SkipEmptyParts);
m_rootBackendsLoaded = true;
emit availableBackendsChanged();
}
emit rootBackendsLoadedChanged();
}
void Ollie9pClient::loadRootAgents()
{
if (m_rootAgentsLoaded) return;
if (!m_9p || !m_9p->isConnected()) return;
QByteArray out = m_9p->read(QStringLiteral("agents"));
QString raw = QString::fromUtf8(out).trimmed();
if (!raw.isEmpty()) {
m_availableAgents = raw.split('\n', Qt::SkipEmptyParts);
m_rootAgentsLoaded = true;
emit availableAgentsChanged();
} else {
m_availableAgents = QStringList() << QStringLiteral("default");
m_rootAgentsLoaded = true;
emit availableAgentsChanged();
}
emit rootAgentsLoadedChanged();
}
QStringList Ollie9pClient::getAvailableModels(const QString &backend) const
{
if (!m_rootModelsLoaded) return {};
auto it = m_rootModels.find(backend);
if (it == m_rootModels.end()) return {};
return it.value().toStringList();
}
bool Ollie9pClient::createQuickSession()
{
if (!m_9p || !m_9p->isConnected()) return false;
QByteArray result = m_9p->rdwr(QStringLiteral("session/new"), "\n");
return !result.isEmpty() || m_9p->lastError().isEmpty();
}
QString Ollie9pClient::createSession(const QString &name, const QString &remote)
{
if (!m_9p || !m_9p->isConnected()) return QStringLiteral("not connected");
QStringList args;
if (!name.isEmpty()) args << "name=" + name;
if (!remote.isEmpty()) args << "remote=" + remote;
QByteArray result = m_9p->rdwr(QStringLiteral("session/new"), (args.join(" ") + "\n").toUtf8());
QString err = m_9p->lastError();
if (!err.isEmpty()) {
return err;
}
return QString(); // success
}
QString Ollie9pClient::createAgent(const QString &sessionId, const QString &cwd, const QString &backend, const QString &model, const QString &agent, const QString &remote, const QString &agentAlias)
{
if (sessionId.isEmpty() || cwd.isEmpty()) return QStringLiteral("session ID and directory are required");
if (!m_9p || !m_9p->isConnected()) return QStringLiteral("not connected");
QStringList agentArgs;
agentArgs << "cwd=" + cwd;
if (!backend.isEmpty()) agentArgs << "backend=" + backend;
if (!model.isEmpty()) agentArgs << "model=" + model;
if (!agent.isEmpty()) agentArgs << "agent=" + agent;
if (!remote.isEmpty()) agentArgs << "remote=" + remote;
if (!agentAlias.isEmpty()) agentArgs << "name=" + agentAlias;
// Use immutable ID — the 9P namespace resolves it via alias.
QString path = "session/" + sessionId + "/agent/new";
QByteArray result = m_9p->rdwr(path, (agentArgs.join(" ") + "\n").toUtf8());
QString err = m_9p->lastError();
if (!err.isEmpty()) {
qDebug() << "createAgent failed:" << err;
return err;
}
// Return the new agent's ID (trimmed) so callers can switch to it immediately.
// Format: "agentid\n"
return QString::fromUtf8(result).trimmed();
}
bool Ollie9pClient::killAgent(const QString &sessionId, const QString &agentId)
{
if (sessionId.isEmpty() || agentId.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
// Use immutable IDs — the 9P namespace resolves them via aliases.
QString path = "session/" + sessionId + "/agent/" + agentId + "/ctl";
return m_9p->write(path, "kill");
}
bool Ollie9pClient::renameSession(const QString &sessionId, const QString &newName)
{
if (sessionId.isEmpty() || newName.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
// Session names are mutable display names. Writing the name file triggers
// the rename. Use the immutable ID to address the session.
QString path = "session/" + sessionId + "/name";
if (m_9p->write(path, (newName + "\n").toUtf8())) {
// With immutable-ID paths, renaming no longer invalidates streams or
// the active selection. Just refresh the session list to pick up the
// new display name.
refreshSessions();
return true;
}
qDebug() << "renameSession failed:" << m_9p->lastError();
return false;
}
bool Ollie9pClient::renameAgent(const QString &sessionId, const QString &agentId, const QString &newName)
{
if (sessionId.isEmpty() || agentId.isEmpty() || newName.isEmpty()) return false;
if (!m_9p || !m_9p->isConnected()) return false;
// Agent names are mutable display names. Write the name file directly
// using immutable IDs to address the path.
QString path = "session/" + sessionId + "/agent/" + agentId + "/name";
if (m_9p->write(path, (newName + "\n").toUtf8())) {
// With immutable-ID paths, renaming no longer invalidates streams.
refreshSessions();
return true;
}
qDebug() << "renameAgent failed:" << m_9p->lastError();
return false;
}
// --- Root data loading (lazy) ---
void Ollie9pClient::ensureRootDataLoaded()
{
if (!m_rootBackendsLoaded) loadRootBackends();
if (!m_rootAgentsLoaded) loadRootAgents();
if (!m_rootModelsLoaded && m_9p && m_9p->isConnected()) {
QByteArray out = m_9p->read(QStringLiteral("models"));
QString raw = QString::fromUtf8(out);
if (!raw.isEmpty()) {
for (const QString &line : raw.split('\n')) {
int tab = line.indexOf('\t');
if (tab < 0) continue;
QString be = line.left(tab).trimmed();
QString mo = line.mid(tab + 1).trimmed();
if (!be.isEmpty() && !mo.isEmpty()) {
QStringList list = m_rootModels[be].toStringList();
list.append(mo);
m_rootModels[be] = QVariant(list);
}
}
m_rootModelsLoaded = true;
emit rootModelsLoadedChanged();
}
}
refreshSessions();
}
// --- Streaming ---
void Ollie9pClient::startActiveAgentStreams()
{
if (!m_daemonConnected) return;
if (m_activeSessionId.isEmpty() || m_agentId.isEmpty()) return;
m_chat->start(agentPath() + "/chat.raw");
m_state->start(agentPath() + "/statewait");
}
void Ollie9pClient::stopAgentStreams()
{
if (m_chat) m_chat->stop();
if (m_state) m_state->stop();
}
void Ollie9pClient::stopStreams()
{
// Daemon and per-agent connectivity monitoring are independent of the
// currently selected agent. Selection changes must not tear them down.
stopAgentStreams();
}
// --- Helpers ---
QString Ollie9pClient::agentPath() const
{
// Use immutable IDs — the 9P namespace resolves them via aliases.
return "session/" + m_activeSessionId + "/agent/" + m_agentId;
}