diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index 22dbd76..52e1daa 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -42,7 +42,6 @@ type Agent struct { fifo Fifo // prompt queue toolCallCount atomic.Int64 pendingInject atomic.Pointer[string] - bgProcs bgTracker // background process tracking submitMu sync.Mutex // serializes Submit calls (commands + turns) stateMu sync.RWMutex signalMu sync.Mutex @@ -655,7 +654,6 @@ func (ag *Agent) DismissProc(pid int) error { if !ts.DismissDetached(pid) { return fmt.Errorf("process %d not found", pid) } - ag.bgProcs.Remove(pid) // Also remove from local tracker if present return nil } diff --git a/cmd/olliesrv/internal/agent/background.go b/cmd/olliesrv/internal/agent/background.go deleted file mode 100644 index 5e4b437..0000000 --- a/cmd/olliesrv/internal/agent/background.go +++ /dev/null @@ -1,166 +0,0 @@ -package agent - -import ( - "fmt" - "strings" - "sync" - - "ollie/toolsrv" -) - -// bgProc tracks a background process for interrupt injection. -type bgProc struct { - PID int - Cmd string // human-readable command description - Tool string // tool name that started it - LastOutput string // last seen output (for diff detection) -} - -// bgTracker manages background process state for an agent. -type bgTracker struct { - mu sync.Mutex - procs []bgProc -} - -// Add registers a new background process. -func (bt *bgTracker) Add(pid int, tool, cmd string) { - bt.mu.Lock() - defer bt.mu.Unlock() - bt.procs = append(bt.procs, bgProc{PID: pid, Tool: tool, Cmd: cmd}) -} - -// Remove removes a process by PID (after it's been reported as exited). -func (bt *bgTracker) Remove(pid int) { - bt.mu.Lock() - defer bt.mu.Unlock() - for i, p := range bt.procs { - if p.PID == pid { - bt.procs = append(bt.procs[:i], bt.procs[i+1:]...) - return - } - } -} - -// CollectInterrupts checks all tracked background procs for new output -// and returns formatted interrupt blocks. Removes completed procs after -// reporting their final output. -func (bt *bgTracker) CollectInterrupts(conn *toolsrv.Conn) string { - if conn == nil { - return "" - } - bt.mu.Lock() - defer bt.mu.Unlock() - - if len(bt.procs) == 0 { - return "" - } - - var sb strings.Builder - var gone []int - - for i := range bt.procs { - p := &bt.procs[i] - - // Get current output - output, err := conn.GetDetachedOutput(p.PID) - if err != nil { - // Process no longer exists in toolsrv (GC'd or dismissed) - mark for removal - gone = append(gone, i) - continue - } - - // Only inject if output changed - if output == p.LastOutput { - continue - } - p.LastOutput = output - - // Get process status - status := "running" - exitCode := -1 - - // Check if exited by trying to parse stat - info := readProcStat(conn, p.PID) - if info != nil { - if exited, ok := info["exited"].(bool); ok && exited { - status = "exited" - if code, ok := info["exit_code"].(int); ok { - exitCode = code - } - // Don't remove completed procs - keep them until explicitly dismissed - } - } - - // Tail last 20 lines - lines := tailLines(output, 20) - - sb.WriteString("\n\n= 0 { - sb.WriteString(fmt.Sprintf(" exit=\"%d\"", exitCode)) - } - sb.WriteString(">\n") - sb.WriteString(lines) - sb.WriteString("\n") - } - - // Remove procs that no longer exist in toolsrv (reverse order to preserve indices) - for i := len(gone) - 1; i >= 0; i-- { - idx := gone[i] - bt.procs = append(bt.procs[:idx], bt.procs[idx+1:]...) - } - - return sb.String() -} - -// HasProcs returns true if there are any tracked background processes. -func (bt *bgTracker) HasProcs() bool { - bt.mu.Lock() - defer bt.mu.Unlock() - return len(bt.procs) > 0 -} - -// List returns a copy of all tracked background processes. -func (bt *bgTracker) List() []bgProc { - bt.mu.Lock() - defer bt.mu.Unlock() - out := make([]bgProc, len(bt.procs)) - copy(out, bt.procs) - return out -} - -// readProcStat reads status info for a background proc. -func readProcStat(conn *toolsrv.Conn, pid int) map[string]any { - // Use the ListDetachedRaw to find this proc - procs := conn.ListDetachedRaw() - for _, p := range procs { - if m, ok := p.(map[string]any); ok { - if mpid, ok := m["pid"].(int); ok && mpid == pid { - return m - } - } - } - return nil -} - -// tailLines returns the last n lines of s. -func tailLines(s string, n int) string { - lines := strings.Split(strings.TrimRight(s, "\n"), "\n") - if len(lines) > n { - lines = lines[len(lines)-n:] - } - return strings.Join(lines, "\n") -} - -// escapeAttr escapes a string for use in an XML-like attribute. -func escapeAttr(s string) string { - s = strings.ReplaceAll(s, "&", "&") - s = strings.ReplaceAll(s, "\"", """) - s = strings.ReplaceAll(s, "<", "<") - return s -} diff --git a/cmd/olliesrv/internal/agent/loop.go b/cmd/olliesrv/internal/agent/loop.go index 9f8baa6..262a4b3 100644 --- a/cmd/olliesrv/internal/agent/loop.go +++ b/cmd/olliesrv/internal/agent/loop.go @@ -116,7 +116,6 @@ type TurnCtx struct { Save func() IncrToolCallCount func() int64 ResultCache *sync.Map - BgTracker *bgTracker ToolServer *toolsrv.Conn } @@ -181,15 +180,6 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error { msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls} results, interrupted := execToolCalls(rt, ctx, toolCalls, resultCache) - // Inject background process interrupts alongside tool results. - if ctx.BgTracker != nil && ctx.BgTracker.HasProcs() { - if interrupts := ctx.BgTracker.CollectInterrupts(ctx.ToolServer); interrupts != "" { - if len(results) > 0 { - results[len(results)-1].Content += interrupts - } - } - } - h.update(msg, results) ctx.Save() @@ -575,9 +565,6 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync. result = fmt.Sprintf("error: background exec: %v", err) isErr = true } else { - if ctx.BgTracker != nil { - ctx.BgTracker.Add(pid, call.Name, cmdDesc) - } result = fmt.Sprintf("\nid=%d cmd=%q\n", pid, cmdDesc) } } diff --git a/cmd/olliesrv/internal/agent/turn.go b/cmd/olliesrv/internal/agent/turn.go index f9821b6..59cbfed 100644 --- a/cmd/olliesrv/internal/agent/turn.go +++ b/cmd/olliesrv/internal/agent/turn.go @@ -118,7 +118,6 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string { } turnCtx := TurnCtx{ - BgTracker: &ag.bgProcs, ToolServer: ag.runtime.ToolServer, } diff --git a/cmd/toolsrv/internal/server/proc.go b/cmd/toolsrv/internal/server/proc.go index 3b185f9..b5b1a55 100644 --- a/cmd/toolsrv/internal/server/proc.go +++ b/cmd/toolsrv/internal/server/proc.go @@ -39,6 +39,7 @@ type State struct { // Callbacks OnToolsChanged func() + OnProcExit func(proc *Proc) // called when a background proc exits } // Proc represents a running or completed process. @@ -46,6 +47,9 @@ type Proc struct { ID int Tool string Args map[string]string + Cmd string // human-readable command + AgentID string // agent that owns this proc + SessionID string // session that owns this proc StartTime time.Time EndTime time.Time ExitCode int @@ -59,6 +63,21 @@ type Proc struct { proc *os.Process // for signaling } +// Lock locks the proc mutex. +func (p *Proc) Lock() { + p.mu.Lock() +} + +// Unlock unlocks the proc mutex. +func (p *Proc) Unlock() { + p.mu.Unlock() +} + +// OutputString returns the output buffer as a string (caller must hold lock). +func (p *Proc) OutputString() string { + return p.output.String() +} + // NewState creates a new filesystem state. func NewState(cwd string) *State { return &State{ @@ -222,6 +241,15 @@ func (st *State) NewProc(ctx context.Context, payload string, background bool) ( return "", 0, fmt.Errorf("missing 'agent' in payload") } envCopy["OLLIE_UNAME"] = aid + + // Get session ID from environment + sid := envCopy["OLLIE_SESSION_ID"] + + // Get human-readable command + cmd := args["cmd"] + if cmd == "" { + cmd = toolName + } if reg == nil { return "", 0, fmt.Errorf("tool registry not configured") @@ -245,6 +273,9 @@ func (st *State) NewProc(ctx context.Context, payload string, background bool) ( ID: st.allocID(), Tool: toolName, Args: args, + Cmd: cmd, + AgentID: aid, + SessionID: sid, StartTime: time.Now(), done: make(chan struct{}), cancel: cancel, @@ -280,6 +311,11 @@ func (st *State) NewProc(ctx context.Context, payload string, background bool) ( proc.Exited = true proc.EndTime = time.Now() proc.mu.Unlock() + + // Notify on background proc exit + if background && st.OnProcExit != nil { + st.OnProcExit(proc) + } }() // For background procs, wait for the process to start and store its reference. @@ -628,9 +664,9 @@ func (p *Proc) Stat() string { rt := time.Since(p.StartTime) if p.Exited { rt = p.EndTime.Sub(p.StartTime) - return fmt.Sprintf("id=%d\nexited=true\nexit_code=%d\nruntime=%v\ntool=%s\n", p.ID, p.ExitCode, rt, p.Tool) + return fmt.Sprintf("id=%d\ncommand=%s\nexited=true\nexit_code=%d\nruntime=%v\ntool=%s\n", p.ID, p.Cmd, p.ExitCode, rt, p.Tool) } - return fmt.Sprintf("id=%d\nexited=false\nruntime=%v\ntool=%s\n", p.ID, rt, p.Tool) + return fmt.Sprintf("id=%d\ncommand=%s\nexited=false\nruntime=%v\ntool=%s\n", p.ID, p.Cmd, rt, p.Tool) } // --- Helpers --- diff --git a/cmd/toolsrv/main.go b/cmd/toolsrv/main.go index ad31220..003ff35 100644 --- a/cmd/toolsrv/main.go +++ b/cmd/toolsrv/main.go @@ -19,7 +19,9 @@ import ( "net" "os" "os/signal" + "os/user" "path/filepath" + "strings" "sync/atomic" "syscall" "time" @@ -27,6 +29,7 @@ import ( "ollie/cmd/toolsrv/internal/server" "ollie/cmd/toolsrv/internal/registry" "ollie/env" + lib9p "ollie/lib9p/client" olog "ollie/log" "ollie/paths" @@ -86,6 +89,14 @@ func main() { if toolReg != nil { srv.SetRegistry(toolReg) } + + // Set up callback to push completed proc output to agent prompt + srv.Fs.OnProcExit = func(proc *server.Proc) { + if proc.SessionID == "" || proc.AgentID == "" { + return + } + go pushProcInterrupt(proc) + } // Build the filesystem tree tree := srv.BuildTree() @@ -204,3 +215,55 @@ monitor: } } } + +// pushProcInterrupt pushes the completed proc output to the agent's prompt. +func pushProcInterrupt(proc *server.Proc) { + // Build the interrupt message + proc.Lock() + output := proc.OutputString() + exitCode := proc.ExitCode + cmd := proc.Cmd + pid := proc.ID + sid := proc.SessionID + aid := proc.AgentID + proc.Unlock() + + msg := fmt.Sprintf("\n%s\n", + pid, escapeAttr(cmd), exitCode, output) + + // Connect to olliesrv + u, err := user.Current() + if err != nil { + logger.Warn("pushProcInterrupt: failed to get current user: %v", err) + return + } + display := os.Getenv("DISPLAY") + if display == "" { + display = ":0" + } + sockPath := fmt.Sprintf("/tmp/ns.%s.%s/ollie", u.Username, display) + + fsys, err := lib9p.Dial(sockPath, u.Username, "") + if err != nil { + logger.Warn("pushProcInterrupt: failed to connect to olliesrv: %v", err) + return + } + defer fsys.Close() + + // Write to session/{sid}/agent/{aid}/prompt + promptPath := fmt.Sprintf("session/%s/agent/%s/prompt", sid, aid) + if err := fsys.Write(promptPath, []byte(msg)); err != nil { + logger.Warn("pushProcInterrupt: failed to write prompt: %v", err) + return + } + + logger.Info("pushed proc %d interrupt to %s", pid, promptPath) +} + +// escapeAttr escapes a string for use in an XML-like attribute. +func escapeAttr(s string) string { + s = strings.ReplaceAll(s, "&", "&") + s = strings.ReplaceAll(s, "\"", """) + s = strings.ReplaceAll(s, "<", "<") + return s +}