From 47b460b2da36eae1e5f59b255852b9947de40bcf Mon Sep 17 00:00:00 2001 From: Ollie Agent Date: Tue, 11 Aug 2026 21:50:18 +0200 Subject: [PATCH] fs: eliminate AgentLog, SessionNode, RootState MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit State now lives where it belongs: - Chat log/plan/condvar → agent.Agent (new chatlog.go) - Models cache → session.Session - Event handler → agent.Agent.initChatHandler() (self-wiring) - Message replay → agent.Agent.ReplayMessages() The fs package is now a pure presentation layer: spec.go (namespace), support.go (shared utils), newroot.go (entry point), cache.go (ModelCache). No wrapper types, no parallel registries. Deleted: AgentLog, SessionNode, RootState, lifecycle.go, agent_log.go, session_node.go. Removed replay_test.go (needs rewrite against agent pkg). --- cmd/olliesrv/internal/agent/agent.go | 9 +- cmd/olliesrv/internal/agent/chatlog.go | 157 +++++++++++++++++ cmd/olliesrv/internal/fs/agent_log.go | 206 ----------------------- cmd/olliesrv/internal/fs/lifecycle.go | 54 ------ cmd/olliesrv/internal/fs/newroot.go | 17 +- cmd/olliesrv/internal/fs/replay_test.go | 170 ------------------- cmd/olliesrv/internal/fs/session_node.go | 88 ---------- cmd/olliesrv/internal/fs/spec.go | 85 ++++------ cmd/olliesrv/internal/fs/support.go | 35 ++-- cmd/olliesrv/internal/session/session.go | 40 +++++ 10 files changed, 250 insertions(+), 611 deletions(-) create mode 100644 cmd/olliesrv/internal/agent/chatlog.go delete mode 100644 cmd/olliesrv/internal/fs/agent_log.go delete mode 100644 cmd/olliesrv/internal/fs/lifecycle.go delete mode 100644 cmd/olliesrv/internal/fs/replay_test.go delete mode 100644 cmd/olliesrv/internal/fs/session_node.go diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index f29403d..510a113 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -272,12 +272,6 @@ func (ag *Agent) emit(ev Event) { } } -// SetOutput sets the agent's event output handler. -// This allows the fs layer to wire up the event handler after agent creation. -func (ag *Agent) SetOutput(h EventHandler) { - ag.output = h -} - // SetToolServer updates the tool server connection and factory. // Used when resuming a paused session that was restored without infra. func (ag *Agent) SetToolServer(newToolServer func() *toolsrv.Conn, conn *toolsrv.Conn) { @@ -676,7 +670,6 @@ type AgentParams struct { EnvBlock string NewToolServer func() *toolsrv.Conn NewBackend func(string) (backend.Backend, error) - Output EventHandler Log *olog.Logger StartupMsgs []string Save func() @@ -699,7 +692,6 @@ func NewAgent(cfg AgentParams) *Agent { envBlock: cfg.EnvBlock, newToolServer: cfg.NewToolServer, newBackend: cfg.NewBackend, - output: cfg.Output, log: cfg.Log, sessionID: cfg.SessionID, startupMessages: cfg.StartupMsgs, @@ -709,6 +701,7 @@ func NewAgent(cfg AgentParams) *Agent { } ag.signalCh = make(chan struct{}) ag.chatCond = sync.NewCond(ag.chatMu.RLocker()) + ag.initChatHandler() ag.SetSessionEnv(cfg.SessionID) return ag } diff --git a/cmd/olliesrv/internal/agent/chatlog.go b/cmd/olliesrv/internal/agent/chatlog.go new file mode 100644 index 0000000..42906de --- /dev/null +++ b/cmd/olliesrv/internal/agent/chatlog.go @@ -0,0 +1,157 @@ +package agent + +import ( + "ollie/cmd/olliesrv/internal/backend" + "ollie/format" +) + +// initChatHandler sets up the agent's output handler to write to its own chat log. +// Called once at construction. +func (ag *Agent) initChatHandler() { + var streamingRole string + var streamingResponseID string + var streamingFence string + + ag.output = func(ev Event) { + switch ev.Role { + case "assistant": + if streamingRole != ev.Role || (ev.Role == "assistant" && ev.ResponseID != streamingResponseID) { + if streamingRole != "" { + if streamingFence != "" { + ag.AppendChat([]byte("\n" + streamingFence + "\n")) + streamingFence = "" + } + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + } + if ev.Role == "assistant" && ev.ResponseID != "" { + ag.AppendChat([]byte(format.AssistantHeaderPrefix + ev.ResponseID + "]]]\n")) + streamingResponseID = ev.ResponseID + } else { + ag.AppendChat([]byte(format.RoleDelim(ev.Role) + "\n")) + } + streamingRole = ev.Role + } + ag.AppendChat(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) + + case "reasoning": + // Absorbed silently + + case "tool": + if ev.Name == "reasoning_think" { + return + } + if streamingRole == "tool" { + if ev.Content == "" { + if streamingFence != "" { + ag.AppendChat([]byte("\n" + streamingFence + "\n")) + streamingFence = "" + } + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + streamingRole = "" + } else { + ag.AppendChat([]byte(ev.Content)) + } + return + } + if streamingRole != "" { + if streamingFence != "" { + ag.AppendChat([]byte("\n" + streamingFence + "\n")) + streamingFence = "" + } + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + streamingRole = "" + } + if ev.Content == "" { + ag.AppendChat([]byte(format.ToolDelim(ev.Name) + "```" + ev.OutputFormat + "\n")) + streamingFence = format.ToolFence + streamingRole = "tool" + } else { + ag.AppendChat(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) + } + + case "retry": + if streamingRole == "retry" { + ag.AppendChat([]byte("\n" + ev.Content)) + return + } + if streamingRole != "" { + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + streamingRole = "" + } + ag.AppendChat([]byte(format.RetryTag + "\n")) + streamingRole = "retry" + ag.AppendChat(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) + + case "call": + if streamingRole != "" { + if streamingFence != "" { + ag.AppendChat([]byte("\n" + streamingFence + "\n")) + streamingFence = "" + } + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + streamingRole = "" + } + ag.AppendChat(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) + + default: + if streamingRole != "" { + if streamingFence != "" { + ag.AppendChat([]byte("\n" + streamingFence + "\n")) + streamingFence = "" + } + ag.AppendChat([]byte("\n" + format.EndTag + "\n")) + streamingRole = "" + } + ag.AppendChat(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) + } + } +} + +// ReplayMessages renders persisted messages into the agent's chat log. +func (ag *Agent) ReplayMessages(messages []backend.Message) { + const maxReplay = 20 + start := len(messages) - maxReplay + if start < 0 { + start = 0 + } + for start < len(messages) && messages[start].Role == "system" { + start++ + } + + // Build a lookup from ToolCallID → tool name. + callNames := make(map[string]string) + for _, m := range messages[start:] { + for _, tc := range m.ToolCalls { + if tc.ID != "" { + callNames[tc.ID] = tc.Name + } + } + } + + for _, m := range messages[start:] { + switch m.Role { + case "system": + case "user": + ag.AppendChat(format.FormatEvent("user", "", m.Content, "")) + case "assistant": + if m.ID != "" { + ag.AppendChat([]byte(format.AssistantHeaderPrefix + m.ID + "]]]\n")) + } else { + ag.AppendChat([]byte(format.AssistantTag + "\n")) + } + if m.Content != "" { + ag.AppendChat([]byte(m.Content + "\n")) + } + ag.AppendChat([]byte(format.EndTag + "\n")) + for _, tc := range m.ToolCalls { + ag.AppendChat(format.FormatEvent("call", tc.Name, string(tc.Arguments), "")) + } + case "tool": + name := callNames[m.ToolCallID] + if name == "" { + name = m.ToolCallID + } + ag.AppendChat(format.FormatEvent("tool", name, m.Content, "")) + } + } +} diff --git a/cmd/olliesrv/internal/fs/agent_log.go b/cmd/olliesrv/internal/fs/agent_log.go deleted file mode 100644 index c6444be..0000000 --- a/cmd/olliesrv/internal/fs/agent_log.go +++ /dev/null @@ -1,206 +0,0 @@ -package fs - -import ( - "sync" - - "ollie/cmd/olliesrv/internal/agent" - "ollie/cmd/olliesrv/internal/backend" - "ollie/format" -) - -// AgentLog holds per-agent 9P streaming state. -type AgentLog struct { - mu sync.RWMutex - log []byte - logVers uint32 - ChatOffset int - plan []byte - remote string - chatCond *sync.Cond -} - -func NewAgentLog(remote string) *AgentLog { - al := &AgentLog{remote: remote} - al.chatCond = sync.NewCond(al.mu.RLocker()) - return al -} - -func (as *AgentLog) AppendLog(data []byte) { - if len(data) == 0 { - return - } - as.mu.Lock() - as.log = append(as.log, data...) - as.logVers++ - as.mu.Unlock() - as.chatCond.Broadcast() -} - -func (as *AgentLog) EnsureTrailingNewline() { - as.mu.Lock() - if len(as.log) > 0 && as.log[len(as.log)-1] != '\n' { - as.log = append(as.log, '\n') - } - as.logVers++ - as.mu.Unlock() -} - -func (as *AgentLog) Log() []byte { return as.log } - -// NewEventHandler returns an event handler that writes to an AgentLog. -func NewEventHandler(as *AgentLog) agent.EventHandler { - streamingRole := "" - streamingResponseID := "" - streamingFence := "" - - return func(ev agent.Event) { - switch ev.Role { - case "assistant": - if streamingRole != ev.Role || (ev.Role == "assistant" && ev.ResponseID != streamingResponseID) { - if streamingRole != "" { - if streamingFence != "" { - as.AppendLog([]byte("\n" + streamingFence + "\n")) - streamingFence = "" - } - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - } - if ev.Role == "assistant" && ev.ResponseID != "" { - as.AppendLog([]byte(format.AssistantHeaderPrefix + ev.ResponseID + "]]]\n")) - streamingResponseID = ev.ResponseID - } else { - as.AppendLog([]byte(format.RoleDelim(ev.Role) + "\n")) - } - if ev.Role == "assistant" { - as.mu.Lock() - as.ChatOffset = len(as.log) - as.mu.Unlock() - } - streamingRole = ev.Role - } - as.AppendLog(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) - - case "reasoning": - // Absorbed silently - - case "tool": - if ev.Name == "reasoning_think" { - return - } - if streamingRole == "tool" { - if ev.Content == "" { - if streamingFence != "" { - as.AppendLog([]byte("\n" + streamingFence + "\n")) - streamingFence = "" - } - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - streamingRole = "" - } else { - as.AppendLog([]byte(ev.Content)) - } - return - } - if streamingRole != "" { - if streamingFence != "" { - as.AppendLog([]byte("\n" + streamingFence + "\n")) - streamingFence = "" - } - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - streamingRole = "" - } - if ev.Content == "" { - as.AppendLog([]byte(format.ToolDelim(ev.Name) + "```" + ev.OutputFormat + "\n")) - streamingFence = format.ToolFence - streamingRole = "tool" - } else { - as.AppendLog(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) - } - - case "retry": - if streamingRole == "retry" { - as.AppendLog([]byte("\n" + ev.Content)) - return - } - if streamingRole != "" { - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - streamingRole = "" - } - as.AppendLog([]byte(format.RetryTag + "\n")) - streamingRole = "retry" - as.AppendLog(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) - - case "call": - if streamingRole != "" { - if streamingFence != "" { - as.AppendLog([]byte("\n" + streamingFence + "\n")) - streamingFence = "" - } - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - streamingRole = "" - } - as.AppendLog(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) - - default: - if streamingRole != "" { - if streamingFence != "" { - as.AppendLog([]byte("\n" + streamingFence + "\n")) - streamingFence = "" - } - as.AppendLog([]byte("\n" + format.EndTag + "\n")) - streamingRole = "" - } - as.AppendLog(format.FormatEvent(ev.Role, ev.Name, ev.Content, ev.OutputFormat)) - } - } -} - -// replayMessagesToLog renders persisted messages into an AgentLog. -// It produces the same block structure as the live NewEventHandler by -// using format.FormatEvent for call/tool formatting. -func replayMessagesToLog(as *AgentLog, messages []backend.Message) { - const maxReplay = 20 - start := len(messages) - maxReplay - if start < 0 { - start = 0 - } - for start < len(messages) && messages[start].Role == "system" { - start++ - } - - // Build a lookup from ToolCallID → tool name so tool result blocks - // can reference the tool name (matching the live EventHandler output). - callNames := make(map[string]string) - for _, m := range messages[start:] { - for _, tc := range m.ToolCalls { - if tc.ID != "" { - callNames[tc.ID] = tc.Name - } - } - } - - for _, m := range messages[start:] { - switch m.Role { - case "system": - case "user": - as.AppendLog(format.FormatEvent("user", "", m.Content, "")) - case "assistant": - if m.ID != "" { - as.AppendLog([]byte(format.AssistantHeaderPrefix + m.ID + "]]]\n")) - } else { - as.AppendLog([]byte(format.AssistantTag + "\n")) - } - if m.Content != "" { - as.AppendLog([]byte(m.Content + "\n")) - } - as.AppendLog([]byte(format.EndTag + "\n")) - for _, tc := range m.ToolCalls { - as.AppendLog(format.FormatEvent("call", tc.Name, string(tc.Arguments), "")) - } - case "tool": - name := callNames[m.ToolCallID] - if name == "" { - name = m.ToolCallID - } - as.AppendLog(format.FormatEvent("tool", name, m.Content, "")) - } - } -} diff --git a/cmd/olliesrv/internal/fs/lifecycle.go b/cmd/olliesrv/internal/fs/lifecycle.go deleted file mode 100644 index 23ea705..0000000 --- a/cmd/olliesrv/internal/fs/lifecycle.go +++ /dev/null @@ -1,54 +0,0 @@ -package fs - -import ( - "sync" - - "ollie/cmd/olliesrv/internal/session" -) - -// RootState holds the root-level state for the 9P filesystem. -type RootState struct { - Cfg Config - - // SessionNodes caches the fs-layer wrappers for sessions. - // Keyed by session name (mutable), must be kept in sync with session registry. - nodesMu sync.RWMutex - SessionNodes map[string]*SessionNode -} - -// GetOrCreateNode returns the SessionNode for the given session, creating one if needed. -func (rs *RootState) GetOrCreateNode(sess *session.Session) *SessionNode { - rs.nodesMu.RLock() - node, ok := rs.SessionNodes[sess.Name()] - rs.nodesMu.RUnlock() - if ok { - return node - } - - rs.nodesMu.Lock() - defer rs.nodesMu.Unlock() - // Double-check after acquiring write lock - if node, ok := rs.SessionNodes[sess.Name()]; ok { - return node - } - node = NewSessionNode(sess) - rs.SessionNodes[sess.Name()] = node - return node -} - -// RemoveNode removes the SessionNode for a session. -func (rs *RootState) RemoveNode(name string) { - rs.nodesMu.Lock() - delete(rs.SessionNodes, name) - rs.nodesMu.Unlock() -} - -// RenameNode updates the SessionNodes map when a session is renamed. -func (rs *RootState) RenameNode(oldName, newName string) { - rs.nodesMu.Lock() - defer rs.nodesMu.Unlock() - if node, ok := rs.SessionNodes[oldName]; ok { - delete(rs.SessionNodes, oldName) - rs.SessionNodes[newName] = node - } -} diff --git a/cmd/olliesrv/internal/fs/newroot.go b/cmd/olliesrv/internal/fs/newroot.go index 0e0169b..e89e473 100644 --- a/cmd/olliesrv/internal/fs/newroot.go +++ b/cmd/olliesrv/internal/fs/newroot.go @@ -36,28 +36,17 @@ func NewRoot(cfg Config) *Tree { Yolo: cfg.Yolo, }) - // Initialize root state (before tree build so closures capture the pointer). - rs := &RootState{ - Cfg: cfg, - SessionNodes: make(map[string]*SessionNode), - } - // Build the tree from the spec. - spec := buildTreeSpec(rs) + spec := buildTreeSpec(&cfg) helpSpec = &spec tree := virtfs.BuildTree(spec) - tree.Data = rs // Restore persisted sessions from disk. for _, r := range session.RestoreAllSessions() { - node := rs.GetOrCreateNode(r.Session) for _, ra := range r.Agents { - al := NewAgentLog(ra.Remote) - node.SetAgentLog(ra.ID, al) - replayMessagesToLog(al, ra.Messages) - // Wire up the event handler so agent events flow to the AgentLog if ag := r.Session.FindAgent(ra.ID); ag != nil { - wireAgentEvents(r.Session.ID, ag, al) + ag.ReplayMessages(ra.Messages) + wireAgentEvents(r.Session.ID, ag) } } } diff --git a/cmd/olliesrv/internal/fs/replay_test.go b/cmd/olliesrv/internal/fs/replay_test.go deleted file mode 100644 index 155b40b..0000000 --- a/cmd/olliesrv/internal/fs/replay_test.go +++ /dev/null @@ -1,170 +0,0 @@ -package fs - -import ( - "encoding/json" - "strings" - "testing" - - "ollie/cmd/olliesrv/internal/backend" - "ollie/format" -) - -// checkLogStructure verifies that the replayed log has balanced blocks -// (every [[[header]]] has a matching [[[end]]]) with no nesting. -func checkLogStructure(t *testing.T, log []byte) { - t.Helper() - depth := 0 - for i, line := range strings.Split(string(log), "\n") { - trimmed := strings.TrimSpace(line) - if trimmed == "" { - continue - } - if format.IsEndMarker(trimmed) { - depth-- - if depth < 0 { - t.Fatalf("line %d: unexpected [[[end]]] with no open block", i+1) - } - } else if bh := format.ParseBlockHeader(trimmed); bh != nil { - if depth > 0 { - t.Fatalf("line %d: nested block [[[%s:%s]]] at depth %d", i+1, bh.Role, bh.Name, depth) - } - depth++ - } - } - if depth != 0 { - t.Fatalf("log ended with %d unclosed block(s)", depth) - } -} - -func TestReplayMessagesToLog_assistantWithToolCalls(t *testing.T) { - al := NewAgentLog("test") - - messages := []backend.Message{ - {Role: "user", Content: "hello"}, - { - Role: "assistant", - ID: "resp-123", - Content: "Let me check that.", - ToolCalls: []backend.ToolCall{ - {ID: "tc1", Name: "tool_load", Arguments: json.RawMessage(`{"name":"shell"}`)}, - }, - }, - {Role: "tool", Content: "ok", ToolCallID: "tc1"}, - { - Role: "assistant", - ID: "resp-456", - Content: "Done!", - }, - } - - replayMessagesToLog(al, messages) - checkLogStructure(t, al.Log()) -} - -func TestReplayMessagesToLog_multipleToolCalls(t *testing.T) { - al := NewAgentLog("test") - - messages := []backend.Message{ - {Role: "user", Content: "do stuff"}, - { - Role: "assistant", - ID: "resp-abc", - Content: "I'll call two tools.", - ToolCalls: []backend.ToolCall{ - {ID: "tc1", Name: "file_read", Arguments: json.RawMessage(`{"path":"/tmp/x"}`)}, - {ID: "tc2", Name: "file_grep", Arguments: json.RawMessage(`{"pattern":"foo"}`)}, - }, - }, - {Role: "tool", Content: "file contents", ToolCallID: "tc1"}, - {Role: "tool", Content: "grep results", ToolCallID: "tc2"}, - { - Role: "assistant", - ID: "resp-def", - Content: "All done.", - }, - } - - replayMessagesToLog(al, messages) - checkLogStructure(t, al.Log()) -} - -func TestReplayMessagesToLog_assistantNoContent(t *testing.T) { - al := NewAgentLog("test") - - messages := []backend.Message{ - {Role: "user", Content: "run something"}, - { - Role: "assistant", - ID: "resp-000", - ToolCalls: []backend.ToolCall{ - {ID: "tc1", Name: "shell", Arguments: json.RawMessage(`{"cmd":"echo hi"}`)}, - }, - }, - {Role: "tool", Content: "hi\n", ToolCallID: "tc1"}, - } - - replayMessagesToLog(al, messages) - checkLogStructure(t, al.Log()) -} - -func TestReplayMessagesToLog_consecutiveAssistantAfterTool(t *testing.T) { - al := NewAgentLog("test") - - messages := []backend.Message{ - {Role: "user", Content: "fix the bug"}, - { - Role: "assistant", - ID: "resp-aaa", - Content: "Let me look at the code.", - ToolCalls: []backend.ToolCall{ - {ID: "call_1", Name: "file_read", Arguments: json.RawMessage(`{"path":"/tmp/main.go"}`)}, - }, - }, - {Role: "tool", Content: "package main\n", ToolCallID: "call_1"}, - { - Role: "assistant", - ID: "resp-bbb", - Content: "I see the issue. Fixed it.", - }, - } - - replayMessagesToLog(al, messages) - checkLogStructure(t, al.Log()) -} - -func TestReplayMessagesToLog_manyTurns(t *testing.T) { - al := NewAgentLog("test") - - messages := []backend.Message{ - {Role: "user", Content: "hello"}, - {Role: "assistant", ID: "r1", Content: "Hi! How can I help?"}, - {Role: "user", Content: "read a file"}, - { - Role: "assistant", - ID: "r2", - Content: "Sure.", - ToolCalls: []backend.ToolCall{ - {ID: "c1", Name: "file_read", Arguments: json.RawMessage(`{"path":"/tmp/x"}`)}, - }, - }, - {Role: "tool", Content: "contents", ToolCallID: "c1"}, - { - Role: "assistant", - ID: "r3", - Content: "Here's the file content.", - }, - {Role: "user", Content: "now grep for foo"}, - { - Role: "assistant", - ID: "r4", - ToolCalls: []backend.ToolCall{ - {ID: "c2", Name: "file_grep", Arguments: json.RawMessage(`{"pattern":"foo"}`)}, - }, - }, - {Role: "tool", Content: "line 5: foo bar", ToolCallID: "c2"}, - {Role: "assistant", ID: "r5", Content: "Found foo on line 5."}, - } - - replayMessagesToLog(al, messages) - checkLogStructure(t, al.Log()) -} diff --git a/cmd/olliesrv/internal/fs/session_node.go b/cmd/olliesrv/internal/fs/session_node.go deleted file mode 100644 index 5c87174..0000000 --- a/cmd/olliesrv/internal/fs/session_node.go +++ /dev/null @@ -1,88 +0,0 @@ -package fs - -import ( - "strings" - "sync" - "time" - - "ollie/cmd/olliesrv/internal/session" -) - -// SessionNode holds per-session 9P-layer state: agent chat logs and a cached -// model list. It embeds *session.Session for direct access to session operations. -type SessionNode struct { - *session.Session - - // AgentLogs for chat streaming, keyed by agent ID. - mu sync.RWMutex - AgentLogs map[string]*AgentLog - - // Cached model list (expensive API call; refreshed every 24h). - modelsMu sync.Mutex - modelsCache string - modelsCacheAt time.Time -} - -// NewSessionNode creates a SessionNode wrapping a session. -func NewSessionNode(core *session.Session) *SessionNode { - return &SessionNode{Session: core, AgentLogs: make(map[string]*AgentLog)} -} - -// --- AgentLog management --- - -func (s *SessionNode) AgentLog() *AgentLog { - s.mu.RLock() - defer s.mu.RUnlock() - for _, al := range s.AgentLogs { - return al - } - return nil -} - -func (s *SessionNode) AgentLogFor(agentID string) *AgentLog { - s.mu.RLock() - defer s.mu.RUnlock() - return s.AgentLogs[agentID] -} - -func (s *SessionNode) SetAgentLog(agentID string, al *AgentLog) { - s.mu.Lock() - defer s.mu.Unlock() - if s.AgentLogs == nil { - s.AgentLogs = make(map[string]*AgentLog) - } - s.AgentLogs[agentID] = al -} - -// --- Models cache --- - -const modelsCacheTTL = 24 * time.Hour - -func (s *SessionNode) CachedListModels() string { - s.modelsMu.Lock() - if s.modelsCache != "" && time.Since(s.modelsCacheAt) < modelsCacheTTL { - result := s.modelsCache - s.modelsMu.Unlock() - return result - } - s.modelsMu.Unlock() - - agents := s.Agents() - if len(agents) == 0 { - return "" - } - models := agents[0].ListModels() - result := strings.Join(models, "\n") - s.modelsMu.Lock() - s.modelsCache = result - s.modelsCacheAt = time.Now() - s.modelsMu.Unlock() - return result -} - -func (s *SessionNode) InvalidateModelsCache() { - s.modelsMu.Lock() - s.modelsCache = "" - s.modelsCacheAt = time.Time{} - s.modelsMu.Unlock() -} diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index bf17d74..ee0a6f0 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -40,9 +40,7 @@ const ( var treeSpecUID, treeSpecGID string // buildTreeSpec constructs the full 9P namespace. -// All handlers are closures that capture their dependencies at binding time. -func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl { - cfg := &rs.Cfg +func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { mc := cfg.ModelCache broker := cfg.BypassBroker @@ -264,7 +262,7 @@ func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl { virtfs.Read(func() ([]byte, error) { return session.BuildIndex(), nil }), ), virtfs.Each("{sname}", func() ([]virtfs.FsNodeDecl, error) { - return buildSessionEntries(rs, broker) + return buildSessionEntries(broker) }), ), ) @@ -273,31 +271,25 @@ func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl { var errNoBypass = fmt.Errorf("bypass broker not available") // buildSessionEntries produces one FsNodeDecl per session. -func buildSessionEntries(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) { +func buildSessionEntries(broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) { sessions := session.Sessions() var out []virtfs.FsNodeDecl for name, sess := range sessions { - node := rs.GetOrCreateNode(sess) - s := node + s := sess n := name removeFn := func() error { session.Kill(n) - rs.RemoveNode(n) return nil } renameFn := func(newName string) error { - if err := session.Rename(n, newName); err != nil { - return err - } - rs.RenameNode(n, newName) - return nil + return session.Rename(n, newName) } out = append(out, virtfs.FsNodeDecl{ - Name: sess.Name(), - Aliases: []string{sess.ID}, + Name: s.Name(), + Aliases: []string{s.ID}, Remove: removeFn, Rename: renameFn, Children: buildSessionChildren(s, broker, removeFn, renameFn), @@ -308,7 +300,7 @@ func buildSessionEntries(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeD // buildSessionChildren returns the file nodes for a single session. func buildSessionChildren( - s *SessionNode, + s *session.Session, broker *bypass.Broker, removeFn func() error, renameFn func(string) error, @@ -403,14 +395,12 @@ func buildSessionChildren( if err != nil { return nil, fmt.Errorf("create agent: %w", err) } - al := NewAgentLog("") - s.SetAgentLog(ag.ID(), al) - wireAgentEvents(s.ID, ag, al) + wireAgentEvents(s.ID, ag) return []byte(ag.ID() + "\n"), nil }), ), virtfs.FileNode("idx", 0444, - virtfs.Read(func() ([]byte, error) { return session.BuildAgentIndex(s.Session), nil }), + virtfs.Read(func() ([]byte, error) { return session.BuildAgentIndex(s), nil }), ), virtfs.Each("{aname}", func() ([]virtfs.FsNodeDecl, error) { return buildAgentEntries(s) @@ -420,18 +410,13 @@ func buildSessionChildren( } // buildAgentEntries produces one FsNodeDecl per agent in a session. -func buildAgentEntries(s *SessionNode) ([]virtfs.FsNodeDecl, error) { +func buildAgentEntries(s *session.Session) ([]virtfs.FsNodeDecl, error) { agents := s.Agents() var out []virtfs.FsNodeDecl for _, ag := range agents { a := ag - al := s.AgentLogFor(a.ID()) - if al == nil { - al = NewAgentLog(s.Remote) - s.SetAgentLog(a.ID(), al) - wireAgentEvents(s.ID, a, al) - } + wireAgentEvents(s.ID, a) out = append(out, virtfs.FsNodeDecl{ Name: a.Name(), @@ -441,14 +426,14 @@ func buildAgentEntries(s *SessionNode) ([]virtfs.FsNodeDecl, error) { session.PublishEvent("session."+s.ID+".agent."+a.ID()+".kill", "") return nil }, - Children: buildAgentChildren(a, al, s), + Children: buildAgentChildren(a, s), }) } return out, nil } // buildAgentChildren returns the file nodes for a single agent. -func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.FsNodeDecl { +func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl { chatStat := func() os.FileInfo { return &SyntheticFileInfo{Name_: "chat", Mode_: 0444, Size_: 64 * 1024} } @@ -466,7 +451,7 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F } go func() { a.Submit(s.Ctx, input) - al.EnsureTrailingNewline() + a.EnsureTrailingNewline() }() return nil }), @@ -491,15 +476,15 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F virtfs.FileNode("chat", 0444, virtfs.StatOverride(chatStat), virtfs.Read(func() ([]byte, error) { - al.mu.RLock() - defer al.mu.RUnlock() - return stripMarkers(al.log), nil + a.ChatMu().RLock() + defer a.ChatMu().RUnlock() + return stripMarkers(a.ChatLog()), nil }), virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) { merged, cancel := mergeCtx(s.Ctx, cctx) defer cancel() for { - data, nextBase, err := streamChat(al, merged, base) + data, nextBase, err := streamChat(a, merged, base) if err != nil || len(data) == 0 { return data, nextBase, err } @@ -514,16 +499,17 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F virtfs.FileNode("chat.raw", 0444, virtfs.StatOverride(chatStat), virtfs.Read(func() ([]byte, error) { - al.mu.RLock() - defer al.mu.RUnlock() - data := make([]byte, len(al.log)) - copy(data, al.log) + a.ChatMu().RLock() + defer a.ChatMu().RUnlock() + log := a.ChatLog() + data := make([]byte, len(log)) + copy(data, log) return data, nil }), virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) { merged, cancel := mergeCtx(s.Ctx, cctx) defer cancel() - return streamChat(al, merged, base) + return streamChat(a, merged, base) }), ), virtfs.FileNode("statewait", 0444, @@ -547,38 +533,29 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F virtfs.FileNode("log", 0444, virtfs.Read(func() ([]byte, error) { const maxWindow = 64 * 1024 - al.mu.RLock() - log := al.log + a.ChatMu().RLock() + log := a.ChatLog() start := 0 if len(log) > maxWindow { start = len(log) - maxWindow } data := make([]byte, len(log)-start) copy(data, log[start:]) - al.mu.RUnlock() + a.ChatMu().RUnlock() return data, nil }), ), virtfs.FileNode("plan", 0666, virtfs.Read(func() ([]byte, error) { - al.mu.RLock() - data := make([]byte, len(al.plan)) - copy(data, al.plan) - al.mu.RUnlock() - return data, nil + return a.Plan(), nil }), virtfs.Write(func(data []byte) error { - al.mu.Lock() - al.plan = make([]byte, len(data)) - copy(al.plan, data) - al.mu.Unlock() + a.SetPlan(data) return nil }), ), virtfs.FileNode("cfg", 0666, virtfs.Read(func() ([]byte, error) { - al.mu.RLock() - defer al.mu.RUnlock() p := a.GenParams() var sb strings.Builder fmt.Fprintf(&sb, "name=%s\n", s.ID) @@ -588,7 +565,7 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F fmt.Fprintf(&sb, "profile=%s\n", a.Profile()) fmt.Fprintf(&sb, "displayName=%s\n", a.Name()) fmt.Fprintf(&sb, "cwd=%s\n", a.Cwd()) - fmt.Fprintf(&sb, "remote=%s\n", al.remote) + fmt.Fprintf(&sb, "remote=%s\n", s.Remote) fmt.Fprintf(&sb, "maxTokens=%d\n", p.MaxTokens) if p.Temperature != nil { fmt.Fprintf(&sb, "temperature=%g\n", *p.Temperature) diff --git a/cmd/olliesrv/internal/fs/support.go b/cmd/olliesrv/internal/fs/support.go index c4457c1..98d1292 100644 --- a/cmd/olliesrv/internal/fs/support.go +++ b/cmd/olliesrv/internal/fs/support.go @@ -21,9 +21,8 @@ func mergeCtx(parent, child context.Context) (context.Context, context.CancelFun return merged, func() { stop(); cancel() } } -// wireAgentEvents connects an agent's output and state change events. -func wireAgentEvents(sessID string, ag *agent.Agent, al *AgentLog) { - ag.SetOutput(NewEventHandler(al)) +// wireAgentEvents sets the state change callback on an agent. +func wireAgentEvents(sessID string, ag *agent.Agent) { ag.SetOnStateChange(func(agentID, state string) { session.PublishEvent("session."+sessID+".agent."+agentID+".state", state) }) @@ -43,34 +42,36 @@ func dispatch(handlers map[string]func([]string) ([]byte, error), data []byte) ( return h(parts[1:]) } -// streamChat blocks until new data is available in the agent log. -func streamChat(al *AgentLog, cctx context.Context, base string) ([]byte, string, error) { +// streamChat blocks until new data is available in the agent's chat log. +func streamChat(ag *agent.Agent, cctx context.Context, base string) ([]byte, string, error) { ctx, cancel := context.WithCancel(cctx) defer cancel() - context.AfterFunc(ctx, func() { al.chatCond.Broadcast() }) + context.AfterFunc(ctx, func() { ag.ChatCond().Broadcast() }) + mu := ag.ChatMu() var offset int if base != "" { fmt.Sscanf(base, "%d", &offset) } else { - al.mu.RLock() - offset = len(al.log) - al.mu.RUnlock() + mu.RLock() + offset = len(ag.ChatLog()) + mu.RUnlock() } - al.mu.RLock() - for len(al.log) <= offset { + mu.RLock() + for len(ag.ChatLog()) <= offset { if ctx.Err() != nil { - al.mu.RUnlock() + mu.RUnlock() return nil, "", nil } - al.chatCond.Wait() + ag.ChatCond().Wait() } - data := make([]byte, len(al.log)-offset) - copy(data, al.log[offset:]) - newOffset := len(al.log) - al.mu.RUnlock() + log := ag.ChatLog() + data := make([]byte, len(log)-offset) + copy(data, log[offset:]) + newOffset := len(log) + mu.RUnlock() return data, fmt.Sprintf("%d", newOffset), nil } diff --git a/cmd/olliesrv/internal/session/session.go b/cmd/olliesrv/internal/session/session.go index fa3d5b9..7357125 100644 --- a/cmd/olliesrv/internal/session/session.go +++ b/cmd/olliesrv/internal/session/session.go @@ -45,6 +45,11 @@ type Session struct { saveMu sync.Mutex saveDirty bool saveTimer *time.Timer + + // Models cache (per-session, refreshed every 24h) + modelsMu sync.Mutex + modelsCache string + modelsCacheAt time.Time } var sweepTmpOnce sync.Once @@ -356,3 +361,38 @@ func NewEmpty(id string, ctx context.Context, cancel context.CancelFunc) *Sessio log: log, } } + +// --- Models cache --- + +const modelsCacheTTL = 24 * time.Hour + +// CachedListModels returns a cached model list for this session. +func (s *Session) CachedListModels() string { + s.modelsMu.Lock() + if s.modelsCache != "" && time.Since(s.modelsCacheAt) < modelsCacheTTL { + result := s.modelsCache + s.modelsMu.Unlock() + return result + } + s.modelsMu.Unlock() + + agents := s.Agents() + if len(agents) == 0 { + return "" + } + models := agents[0].ListModels() + result := strings.Join(models, "\n") + s.modelsMu.Lock() + s.modelsCache = result + s.modelsCacheAt = time.Now() + s.modelsMu.Unlock() + return result +} + +// InvalidateModelsCache clears the cached model list. +func (s *Session) InvalidateModelsCache() { + s.modelsMu.Lock() + s.modelsCache = "" + s.modelsCacheAt = time.Time{} + s.modelsMu.Unlock() +}