fs: eliminate AgentLog, SessionNode, RootState
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).
This commit is contained in:
Ollie Agent
parent
31d3466b3b
commit
47b460b2da
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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, ""))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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, ""))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
}
|
||||
|
|
@ -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()
|
||||
}
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue