package main import ( "context" "encoding/json" "fmt" "os" "sort" "strings" "sync" "time" "github.com/godbus/dbus/v5" "github.com/godbus/dbus/v5/introspect" "ollie/agent" "ollie/backend" "olliesrv/fs" "olliesrv/session" ) const ( busName = "org.ollie.SessionManager" busPath = "/org/ollie/SessionManager" busIface = "org.ollie.SessionManager" ) // DBusAdapter exposes sessions over D-Bus. type DBusAdapter struct { conn *dbus.Conn tree *fs.Tree mu sync.RWMutex watchers map[string]context.CancelFunc // sessionID -> cancel for state watcher } // startDBus connects to the session bus, claims the well-known name, and // exports the adapter. Returns nil (no-op) if the bus is unavailable. func startDBus(tree *fs.Tree) *DBusAdapter { conn, err := dbus.ConnectSessionBus() if err != nil { fmt.Fprintf(os.Stderr, "dbus: session bus unavailable: %v\n", err) return nil } reply, err := conn.RequestName(busName, dbus.NameFlagDoNotQueue) if err != nil || reply != dbus.RequestNameReplyPrimaryOwner { fmt.Fprintf(os.Stderr, "dbus: cannot claim %s (ollied already running?)\n", busName) conn.Close() return nil } a := &DBusAdapter{ conn: conn, tree: tree, watchers: make(map[string]context.CancelFunc), } conn.Export(a, busPath, busIface) conn.Export(introspect.Introspectable(introspectXML), busPath, "org.freedesktop.DBus.Introspectable") fmt.Printf("dbus: exported %s\n", busName) return a } // WatchExisting starts watchers for all sessions that already exist. // Called after the adapter is initialized to cover sessions restored at startup. func (a *DBusAdapter) WatchExisting() { if a == nil { return } for _, sess := range session.All(a.tree) { id := sess.RunnableID() a.startWatcher(id, sess) // Emit SessionCreated so GUI clients that connected before restore // (via serviceRegistered) learn about pre-existing sessions. a.conn.Emit(busPath, busIface+".SessionCreated", id) } } func (a *DBusAdapter) Close() { if a == nil || a.conn == nil { return } a.mu.Lock() for _, cancel := range a.watchers { cancel() } a.mu.Unlock() a.conn.Close() } // --- Lifecycle callbacks --- func (a *DBusAdapter) OnSessionCreated(id string, sess *session.Session) { if a == nil { return } a.startWatcher(id, sess) go a.conn.Emit(busPath, busIface+".SessionCreated", id) } func (a *DBusAdapter) OnSessionKilled(id string) { if a == nil { return } a.stopWatcher(id) go a.conn.Emit(busPath, busIface+".SessionKilled", id) } func (a *DBusAdapter) OnSessionRenamed(oldID, newID string) { if a == nil { return } a.stopWatcher(oldID) // The session object is the same, just re-lookup sess := session.Lookup(a.tree, newID) if sess != nil { a.startWatcher(newID, sess) } go a.conn.Emit(busPath, busIface+".SessionRenamed", oldID, newID) } // startWatcher launches goroutines for StateChanged and ChatUpdated signals. func (a *DBusAdapter) startWatcher(id string, sess *session.Session) { ctx, cancel := context.WithCancel(context.Background()) a.mu.Lock() a.watchers[id] = cancel a.mu.Unlock() // State watcher go func() { current := sess.Core.Agent().State() for { next, ok := sess.Core.WaitChange(ctx, agent.WatchState, current) if !ok { return } current = next a.conn.Emit(busPath, busIface+".StateChanged", id, current) if current == "idle" { a.sendNotification("Ollie", "Session "+id+" finished") } } }() // Agent watcher — emits ConfigChanged when the agent name changes go func() { current := sess.Core.Agent().Name() for { next, ok := sess.Core.WaitChange(ctx, "agent", current) if !ok { return } current = next a.conn.Emit(busPath, busIface+".ConfigChanged", id, "agent", current) } }() // Chat watcher (polls log version) go func() { var lastLen int var lastVers uint32 for { select { case <-ctx.Done(): return case <-time.After(150 * time.Millisecond): } l, v := sess.LogInfo() if v == lastVers { continue } if l > lastLen { // Read the new chunk sess.Mu().RLock() chunk := string(sess.Log()[lastLen:]) sess.Mu().RUnlock() a.conn.Emit(busPath, busIface+".ChatUpdated", id, int64(lastLen), chunk) } lastLen = l lastVers = v } }() // Detached process exit watcher go func() { exitedPIDs := make(map[int]bool) // already-reported PIDs for { select { case <-ctx.Done(): return case <-time.After(1 * time.Second): } infos := sess.Core.Agent().ListDetached() // Clean up stale entries from exitedPIDs activePIDs := make(map[int]bool, len(infos)) for _, info := range infos { activePIDs[info.PID] = true } for pid := range exitedPIDs { if !activePIDs[pid] { delete(exitedPIDs, pid) } } for _, info := range infos { if info.Exited && !exitedPIDs[info.PID] { exitedPIDs[info.PID] = true a.conn.Emit(busPath, busIface+".ProcessExited", id, int32(info.PID), int32(info.ExitCode)) } } } }() } func (a *DBusAdapter) stopWatcher(id string) { a.mu.Lock() if cancel, ok := a.watchers[id]; ok { cancel() delete(a.watchers, id) } a.mu.Unlock() } // --- D-Bus exported methods --- func (a *DBusAdapter) CreateSession(cwd, backendName, modelName, agentName, systemPrompt, remote string) (string, *dbus.Error) { var args []string if cwd != "" { args = append(args, "cwd="+cwd) } if backendName != "" { args = append(args, "backend="+backendName) } if modelName != "" { args = append(args, "model="+modelName) } if agentName != "" { args = append(args, "agent="+agentName) } if systemPrompt != "" { args = append(args, "system_prompt="+systemPrompt) } if remote != "" { args = append(args, "remote="+remote) } if cwd == "" { dir, _ := os.Getwd() args = append(args, "cwd="+dir) } id, err := session.CreateFromRoot(a.tree, args) if err != nil { return "", dbus.MakeFailedError(err) } return id, nil } func (a *DBusAdapter) ListSessions() ([]string, *dbus.Error) { var result []string for _, sess := range session.All(a.tree) { id := sess.RunnableID() result = append(result, fmt.Sprintf("%s\t%s\t%s\t%s", id, sess.Core.Agent().State(), sess.Core.Agent().ModelName(), sess.Core.Agent().Name())) } return result, nil } func (a *DBusAdapter) KillSession(sessionID string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } session.KillFromRoot(a.tree, sessionID) return true, nil } func (a *DBusAdapter) RenameSession(sessionID, newName string) (bool, *dbus.Error) { if err := session.RenameFromRoot(a.tree, sessionID, newName); err != nil { return false, dbus.MakeFailedError(err) } return true, nil } func (a *DBusAdapter) Submit(sessionID, prompt string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } go sess.Core.Agent().Submit(sess.SessionCtx, prompt) return true, nil } func (a *DBusAdapter) React(sessionID, emoji string) (bool, *dbus.Error) { return a.ReactTo(sessionID, "", emoji) } func (a *DBusAdapter) ReactTo(sessionID, responseID, emoji string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } if err := sess.Core.Agent().React(responseID, emoji); err != nil { return false, dbus.MakeFailedError(err) } return true, nil } func (a *DBusAdapter) Interrupt(sessionID string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } sess.Core.Agent().Interrupt(agent.ErrInterrupted) return true, nil } func (a *DBusAdapter) GetState(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } return sess.Core.Agent().State(), nil } func (a *DBusAdapter) GetChat(sessionID string, offset int64) (string, int64, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", offset, nil } sess.Mu().RLock() log := sess.Log() if offset < 0 { offset = 0 } if offset > int64(len(log)) { offset = int64(len(log)) } text := string(log[offset:]) newOffset := int64(len(log)) sess.Mu().RUnlock() return text, newOffset, nil } func (a *DBusAdapter) GetUsage(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } return sess.Core.Agent().UsageStr(), nil } func (a *DBusAdapter) GetCost(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } return sess.Core.Agent().CostStr(), nil } func (a *DBusAdapter) GetConfig(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } var sb strings.Builder fmt.Fprintf(&sb, "model=%s\n", sess.Core.Agent().ModelName()) fmt.Fprintf(&sb, "backend=%s\n", sess.Core.Agent().BackendName()) fmt.Fprintf(&sb, "agent=%s\n", sess.Core.Agent().Name()) fmt.Fprintf(&sb, "cwd=%s\n", sess.Core.CWD()) fmt.Fprintf(&sb, "state=%s\n", sess.Core.Agent().State()) return sb.String(), nil } func (a *DBusAdapter) SetConfig(sessionID, key, value string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } switch key { case "model": sess.Core.Agent().Submit(sess.SessionCtx, "/model "+value) case "cwd": if err := sess.Core.SetCWD(value); err != nil { return false, nil } default: return false, nil } return true, nil } func (a *DBusAdapter) GetReactions(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "{}", nil } data, err := json.Marshal(sess.Core.Agent().Reactions()) if err != nil { return "", dbus.MakeFailedError(err) } return string(data), nil } func (a *DBusAdapter) GetContext(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } msgs := sess.Core.Agent().Context() var sb strings.Builder enc := json.NewEncoder(&sb) enc.SetEscapeHTML(false) for _, msg := range msgs { enc.Encode(msg) } return sb.String(), nil } func (a *DBusAdapter) ListBackends() ([]string, *dbus.Error) { return backend.Backends(), nil } func (a *DBusAdapter) ListModelsByBackend(backendName string) ([]string, *dbus.Error) { be, err := backend.NewWithName(backendName) if err != nil { return nil, dbus.MakeFailedError(err) } models := be.Models(context.Background()) sort.Strings(models) return models, nil } func (a *DBusAdapter) ListModels(sessionID string) ([]string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return nil, nil } raw := sess.CachedListModels() models := strings.Split(strings.TrimSpace(raw), "\n") if len(models) == 1 && models[0] == "" { return nil, nil } sort.Strings(models) return models, nil } func (a *DBusAdapter) ListAgents() ([]string, *dbus.Error) { var result []string for _, dir := range agent.AgentsDirs() { entries, err := os.ReadDir(dir) if err != nil { continue } for _, e := range entries { name := e.Name() if strings.HasSuffix(name, ".json") { result = append(result, strings.TrimSuffix(name, ".json")) } } } return result, nil } // --- Detached processes --- func (a *DBusAdapter) DetachProcess(sessionID string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } if !sess.Core.Agent().Detach() { return false, dbus.MakeFailedError(fmt.Errorf("no running process to detach")) } a.conn.Emit(busPath, busIface+".ProcessDetached", sessionID) return true, nil } func (a *DBusAdapter) ListDetached(sessionID string) ([]string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return nil, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } infos := sess.Core.Agent().ListDetached() out := make([]string, len(infos)) for i, info := range infos { status := "running" if info.Exited { status = fmt.Sprintf("exited(%d)", info.ExitCode) } out[i] = fmt.Sprintf("%d\t%s\t%d\t%s", info.PID, info.Command, info.Started, status) } return out, nil } func (a *DBusAdapter) SignalDetached(sessionID string, pid int32, signal string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } var sig int switch strings.ToUpper(signal) { case "TERM", "SIGTERM", "15": sig = 15 case "KILL", "SIGKILL", "9": sig = 9 default: return false, dbus.MakeFailedError(fmt.Errorf("unsupported signal: %s", signal)) } if err := sess.Core.Agent().SignalDetached(int(pid), sig); err != nil { return false, dbus.MakeFailedError(err) } return true, nil } func (a *DBusAdapter) GetDetachedOutput(sessionID string, pid int32) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } output, err := sess.Core.Agent().GetDetachedOutput(int(pid)) if err != nil { return "", dbus.MakeFailedError(err) } return output, nil } func (a *DBusAdapter) DismissDetached(sessionID string, pid int32) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } return sess.Core.Agent().DismissDetached(int(pid)), nil } // --- Generate / Route --- func (a *DBusAdapter) Generate(prompt, system, backendName, model string) (string, *dbus.Error) { result, err := backend.Generate(context.Background(), backend.GenerateRequest{ Prompt: prompt, System: system, Backend: backendName, Model: model, }) if err != nil { return "", dbus.MakeFailedError(err) } return result, nil } func (a *DBusAdapter) Route(task, backendName string) (string, *dbus.Error) { result, err := backend.Route(context.Background(), backend.RouteRequest{ Task: task, Backend: backendName, }) if err != nil { return "", dbus.MakeFailedError(err) } return fmt.Sprintf("backend=%s model=%s", result.Backend, result.Model), nil } func (a *DBusAdapter) Complete(cwd, filePath, prefix, suffix, extraContext string) (string, *dbus.Error) { result, err := agent.Complete(cwd, filePath, prefix, suffix, extraContext) if err != nil { return "", dbus.MakeFailedError(err) } return result, nil } func (a *DBusAdapter) GetEnv(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } var sb strings.Builder fmt.Fprintf(&sb, "OLLIE_SESSION_ID=%s\n", sess.RunnableID()) for _, e := range os.Environ() { if strings.HasPrefix(e, "OLLIE_") && !strings.HasPrefix(e, "OLLIE_SESSION_ID=") { sb.WriteString(e) sb.WriteByte('\n') } } return sb.String(), nil } func (a *DBusAdapter) GetPlan(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } sess.Mu().RLock() plan := string(sess.Plan()) sess.Mu().RUnlock() return plan, nil } func (a *DBusAdapter) SetPlan(sessionID, plan string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } sess.Mu().Lock() sess.SetPlan([]byte(plan)) sess.Mu().Unlock() return true, nil } func (a *DBusAdapter) GetSystemPrompt(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } return sess.Core.Agent().SystemPrompt(), nil } func (a *DBusAdapter) GetContextSize(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } return sess.Core.Agent().CtxSz(), nil } func (a *DBusAdapter) GetPreviousPrompt(sessionID string) (string, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } sess.Mu().RLock() p := string(sess.PrevPrompt()) sess.Mu().RUnlock() return p, nil } func (a *DBusAdapter) Compact(sessionID string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } sess.Core.Agent().Submit(sess.SessionCtx, "/compact") return true, nil } func (a *DBusAdapter) ClearContext(sessionID string) (bool, *dbus.Error) { sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } sess.Core.Agent().Submit(sess.SessionCtx, "/clear") return true, nil } // --- Notifications --- func (a *DBusAdapter) sendNotification(title, body string) { go func() { obj := a.conn.Object("org.freedesktop.Notifications", "/org/freedesktop/Notifications") obj.Call("org.freedesktop.Notifications.Notify", 0, "olliesrv", uint32(0), "system-run", title, body, []string{}, map[string]dbus.Variant{}, int32(5000)) }() } // Introspection XML. var introspectXML = ` `