refactor: push proc completion from toolsrv to agent prompt
- toolsrv pushes <system-proc-complete> to agent prompt when bg proc exits - Remove bgTracker and CollectInterrupts from olliesrv - Add Cmd, AgentID, SessionID fields to Proc - Add OnProcExit callback to State - Add command field to Proc.Stat() output
This commit is contained in:
parent
1e2ce1f899
commit
bc7f51a9dc
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<system-proc-interrupt id=\"")
|
||||
sb.WriteString(fmt.Sprintf("%d", p.PID))
|
||||
sb.WriteString("\" cmd=\"")
|
||||
sb.WriteString(escapeAttr(p.Cmd))
|
||||
sb.WriteString("\" status=\"")
|
||||
sb.WriteString(status)
|
||||
sb.WriteString("\"")
|
||||
if exitCode >= 0 {
|
||||
sb.WriteString(fmt.Sprintf(" exit=\"%d\"", exitCode))
|
||||
}
|
||||
sb.WriteString(">\n")
|
||||
sb.WriteString(lines)
|
||||
sb.WriteString("\n</system-proc-interrupt>")
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
|
@ -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("<system-proc-background>\nid=%d cmd=%q\n</system-proc-background>", pid, cmdDesc)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -118,7 +118,6 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
|
|||
}
|
||||
|
||||
turnCtx := TurnCtx{
|
||||
BgTracker: &ag.bgProcs,
|
||||
ToolServer: ag.runtime.ToolServer,
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 ---
|
||||
|
|
|
|||
|
|
@ -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("<system-proc-complete id=\"%d\" cmd=\"%s\" exit=\"%d\">\n%s\n</system-proc-complete>",
|
||||
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
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue