From 842a575669525d77e29f042fae9715c873f15060 Mon Sep 17 00:00:00 2001 From: Jihad Irfansyah Date: Sat, 1 Aug 2026 10:00:29 +0700 Subject: [PATCH] Add managed background terminal processes --- internal/agent/agent.go | 10 +- internal/agent/harness.go | 24 ++ internal/agent/harness_test.go | 24 ++ internal/agent/process_timeout_test.go | 15 + internal/tools/background_process.go | 417 +++++++++++++++++++ internal/tools/background_process_test.go | 357 ++++++++++++++++ internal/tools/background_process_unix.go | 27 ++ internal/tools/background_process_windows.go | 15 + internal/tools/register.go | 2 +- internal/tools/registry.go | 16 +- internal/tools/shell.go | 39 +- 11 files changed, 933 insertions(+), 13 deletions(-) create mode 100644 internal/agent/process_timeout_test.go create mode 100644 internal/tools/background_process.go create mode 100644 internal/tools/background_process_test.go create mode 100644 internal/tools/background_process_unix.go create mode 100644 internal/tools/background_process_windows.go diff --git a/internal/agent/agent.go b/internal/agent/agent.go index f7532cb1..6eec5659 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -232,7 +232,11 @@ func (a *Agent) Interrupt(sessionID string) bool { if ok { cancel() } - return ok + stopped := 0 + if a.shell != nil { + stopped = a.shell.CancelRunning(sessionID) + } + return ok || stopped > 0 } // ActiveCount reports how many turns are running. @@ -861,6 +865,10 @@ func (a *Agent) toolTimeout(name string) time.Duration { switch name { case "terminal": return time.Duration(maxInt(a.cfg.Terminal.Timeout, 60)) * time.Second + case "process": + // process(wait) intentionally blocks for at most 30 seconds. Leave margin + // for scheduling and JSON serialization so the tool can return its state. + return 45 * time.Second case "delegate_task": return 30 * time.Minute default: diff --git a/internal/agent/harness.go b/internal/agent/harness.go index d3346163..eeed8b31 100644 --- a/internal/agent/harness.go +++ b/internal/agent/harness.go @@ -95,6 +95,12 @@ func newRepeatTracker(limit int) *repeatTracker { func (r *repeatTracker) record(calls []llm.ToolCall) []string { var tripped []string for _, c := range calls { + if processObservation(c) { + // Polling a managed process is an observation of changing external + // state, not a stuck model repeating an idempotent call. The process + // tool bounds every wait and terminal jobs have explicit cancellation. + continue + } sum := sha256.Sum256([]byte(c.Name + "\x00" + normaliseArgs(c.Arguments))) key := hex.EncodeToString(sum[:8]) r.seen[key]++ @@ -105,6 +111,24 @@ func (r *repeatTracker) record(calls []llm.ToolCall) []string { return tripped } +func processObservation(c llm.ToolCall) bool { + if c.Name != "process" { + return false + } + var args struct { + Action string `json:"action"` + } + if json.Unmarshal([]byte(c.Arguments), &args) != nil { + return false + } + switch args.Action { + case "poll", "wait", "log", "list": + return true + default: + return false + } +} + // exceeded reports a call repeated far past the limit, where nudging has // already failed and the run should stop. func (r *repeatTracker) exceeded() bool { diff --git a/internal/agent/harness_test.go b/internal/agent/harness_test.go index 7a9bf2ea..2c2b7f5a 100644 --- a/internal/agent/harness_test.go +++ b/internal/agent/harness_test.go @@ -63,6 +63,30 @@ func TestRepeatTrackerExceeded(t *testing.T) { } } +func TestRepeatTrackerAllowsManagedProcessPolling(t *testing.T) { + r := newRepeatTracker(2) + call := llm.ToolCall{Name: "process", Arguments: `{"action":"wait","process_id":"proc_123","timeout":30}`} + for i := 0; i < 20; i++ { + if got := r.record([]llm.ToolCall{call}); len(got) != 0 { + t.Fatalf("managed process wait was treated as a loop after %d calls: %v", i+1, got) + } + } + if r.exceeded() { + t.Fatal("managed process polling tripped the repeat guard") + } +} + +func TestRepeatTrackerStillTracksProcessKill(t *testing.T) { + r := newRepeatTracker(2) + call := llm.ToolCall{Name: "process", Arguments: `{"action":"kill","process_id":"proc_123"}`} + if got := r.record([]llm.ToolCall{call}); len(got) != 0 { + t.Fatal(got) + } + if got := r.record([]llm.ToolCall{call}); len(got) != 1 || got[0] != "process" { + t.Fatalf("repeated process kill was not tracked: %v", got) + } +} + func TestSteeringRequiresARunningSession(t *testing.T) { a := &Agent{active: map[string]context.CancelFunc{}} if a.Steer("nope", "do this instead") { diff --git a/internal/agent/process_timeout_test.go b/internal/agent/process_timeout_test.go new file mode 100644 index 00000000..bcde45e9 --- /dev/null +++ b/internal/agent/process_timeout_test.go @@ -0,0 +1,15 @@ +package agent + +import ( + "testing" + "time" + + "github.com/enowdev/antares/internal/config" +) + +func TestProcessToolTimeoutLeavesWaitMargin(t *testing.T) { + a := &Agent{cfg: &config.Config{}} + if got := a.toolTimeout("process"); got < 45*time.Second { + t.Fatalf("process tool timeout = %s, want at least 45s for a 30s process wait", got) + } +} diff --git a/internal/tools/background_process.go b/internal/tools/background_process.go new file mode 100644 index 00000000..54ae391c --- /dev/null +++ b/internal/tools/background_process.go @@ -0,0 +1,417 @@ +package tools + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "os/exec" + "sort" + "strings" + "sync" + "time" + + "github.com/enowdev/antares/internal/sandbox" +) + +const ( + backgroundLogLimit = 2 << 20 // retain the newest 2 MiB per process + backgroundReadLimit = 60_000 // keep one tool result inside the normal context budget + backgroundJobLimit = 32 + completedJobTTL = time.Hour +) + +type processStatus string + +const ( + processRunning processStatus = "running" + processCompleted processStatus = "completed" + processFailed processStatus = "failed" + processCancelled processStatus = "cancelled" + processTimedOut processStatus = "timed_out" +) + +type processLog struct { + mu sync.Mutex + data []byte + base int64 +} + +func (l *processLog) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.data = append(l.data, p...) + if drop := len(l.data) - backgroundLogLimit; drop > 0 { + copy(l.data, l.data[drop:]) + l.data = l.data[:len(l.data)-drop] + l.base += int64(drop) + } + return len(p), nil +} + +func (l *processLog) read(offset int64) (chunk string, next int64, truncated, hasMore bool) { + l.mu.Lock() + defer l.mu.Unlock() + if offset < l.base { + offset = l.base + truncated = true + } + end := l.base + int64(len(l.data)) + if offset > end { + offset = end + } + next = offset + min(int64(backgroundReadLimit), end-offset) + return string(l.data[offset-l.base : next-l.base]), next, truncated, next < end +} + +type backgroundProcess struct { + mu sync.Mutex + id string + sessionID string + command string + workspace string + cmd *exec.Cmd + log processLog + readAt int64 + status processStatus + exitCode *int + startedAt time.Time + endedAt time.Time + done chan struct{} + stopOnce sync.Once +} + +type processView struct { + ProcessID string `json:"process_id"` + Status processStatus `json:"status"` + Command string `json:"command,omitempty"` + Workspace string `json:"workspace,omitempty"` + PID int `json:"pid,omitempty"` + ExitCode *int `json:"exit_code,omitempty"` + StartedAt time.Time `json:"started_at"` + EndedAt *time.Time `json:"ended_at,omitempty"` + Output string `json:"output,omitempty"` + NextOffset int64 `json:"next_offset"` + Truncated bool `json:"truncated,omitempty"` + HasMore bool `json:"has_more,omitempty"` +} + +func newProcessID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err == nil { + return "proc_" + hex.EncodeToString(b[:]) + } + return fmt.Sprintf("proc_%x", time.Now().UnixNano()) +} + +func (m *ShellManager) startBackground(sessionID, workspace, command string, timeout time.Duration) (*backgroundProcess, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.reapJobsLocked(time.Now()) + active := 0 + for _, job := range m.jobs { + job.mu.Lock() + if job.sessionID == sessionID && job.status == processRunning { + active++ + } + job.mu.Unlock() + } + if active >= backgroundJobLimit { + return nil, fmt.Errorf("session already has %d running background processes", active) + } + + cmd, err := m.backgroundCommand(sessionID, workspace, command) + if err != nil { + return nil, err + } + job := &backgroundProcess{ + id: newProcessID(), + sessionID: sessionID, + command: command, + workspace: workspace, + cmd: cmd, + status: processRunning, + startedAt: time.Now().UTC(), + done: make(chan struct{}), + } + cmd.Stdout = &job.log + cmd.Stderr = &job.log + configureProcessGroup(cmd) + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("start background process: %w", err) + } + + m.jobs[job.id] = job + + go job.wait() + if timeout > 0 { + go func() { + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-timer.C: + job.stop(processTimedOut) + case <-job.done: + } + }() + } + return job, nil +} + +func (m *ShellManager) backgroundCommand(sessionID, workspace, command string) (*exec.Cmd, error) { + shell, _ := defaultShell(m.cfg.Shell) + switch strings.ToLower(m.cfg.Backend) { + case "docker": + image := m.cfg.DockerImage + if image == "" { + image = "debian:bookworm-slim" + } + net := "none" + if m.cfg.AllowNetwork { + net = "bridge" + } + return exec.Command("docker", "run", "--rm", "-i", "--network", net, + "-v", workspace+":/workspace", "-w", "/workspace", image, "/bin/sh", "-c", command), nil + case "ssh": + if m.cfg.SSHHost == "" { + return nil, fmt.Errorf("terminal.ssh_host is not configured") + } + return exec.Command("ssh", "-T", m.cfg.SSHHost, "/bin/sh -c "+shellQuote(command)), nil + default: + mode, note := sandbox.Resolve(sandbox.Mode(m.cfg.Sandbox)) + if note != "" { + m.warnSandboxOnce(note) + } + policy := sandbox.Policy{Workspace: workspace, AllowNetwork: m.cfg.AllowNetwork, Hidden: m.hiddenPaths()} + cmd, err := sandbox.Command(mode, policy, shell, "-c", command) + if err != nil { + m.warnSandboxOnce(err.Error()) + cmd = exec.Command(shell, "-c", command) + } + cmd.Dir = workspace + env := append(os.Environ(), "ANTARES_SESSION="+sessionID, "TERM=dumb", "PAGER=cat", "GIT_PAGER=cat") + if m.httpShim.dir != "" { + env = withShimEnv(env, m.httpShim) + } + cmd.Env = env + return cmd, nil + } +} + +func shellQuote(s string) string { return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" } + +func (j *backgroundProcess) wait() { + err := j.cmd.Wait() + j.mu.Lock() + if j.status == processRunning { + code := j.cmd.ProcessState.ExitCode() + j.exitCode = &code + if err == nil && code == 0 { + j.status = processCompleted + } else { + j.status = processFailed + } + } + j.endedAt = time.Now().UTC() + j.mu.Unlock() + j.stopOnce.Do(func() { close(j.done) }) +} + +func (j *backgroundProcess) stop(status processStatus) { + j.mu.Lock() + if j.status != processRunning { + j.mu.Unlock() + return + } + j.status = status + cmd := j.cmd + j.mu.Unlock() + terminateProcessGroup(cmd) + go func() { + select { + case <-j.done: + case <-time.After(2 * time.Second): + killProcessGroup(cmd) + } + }() +} + +func (j *backgroundProcess) stopAndWait(status processStatus) { + stopProcesses([]*backgroundProcess{j}, status) +} + +func stopProcesses(jobs []*backgroundProcess, status processStatus) { + for _, job := range jobs { + job.stop(status) + } + for _, job := range jobs { + select { + case <-job.done: + case <-time.After(3 * time.Second): + killProcessGroup(job.cmd) + select { + case <-job.done: + case <-time.After(time.Second): + } + } + } +} + +func (j *backgroundProcess) view(offset int64, consume bool) processView { + j.mu.Lock() + defer j.mu.Unlock() + if consume { + offset = j.readAt + } + view := processView{ProcessID: j.id, Status: j.status, Command: j.command, Workspace: j.workspace, + ExitCode: j.exitCode, StartedAt: j.startedAt} + if j.cmd != nil && j.cmd.Process != nil { + view.PID = j.cmd.Process.Pid + } + if !j.endedAt.IsZero() { + ended := j.endedAt + view.EndedAt = &ended + } + view.Output, view.NextOffset, view.Truncated, view.HasMore = j.log.read(offset) + if consume { + j.readAt = view.NextOffset + } + return view +} + +func (m *ShellManager) getJob(sessionID, id string) (*backgroundProcess, bool) { + m.mu.Lock() + defer m.mu.Unlock() + job, ok := m.jobs[id] + if !ok || job.sessionID != sessionID { + return nil, false + } + return job, true +} + +func (m *ShellManager) listJobs(sessionID string) []processView { + m.mu.Lock() + m.reapJobsLocked(time.Now()) + var jobs []*backgroundProcess + for _, job := range m.jobs { + if job.sessionID == sessionID { + jobs = append(jobs, job) + } + } + m.mu.Unlock() + sort.Slice(jobs, func(i, k int) bool { return jobs[i].startedAt.Before(jobs[k].startedAt) }) + out := make([]processView, 0, len(jobs)) + for _, job := range jobs { + out = append(out, job.view(0, false)) + out[len(out)-1].Output = "" + } + return out +} + +// CancelRunning stops every managed process owned by a session. It is used by +// the agent interrupt path so the dashboard Stop button reaches backend process +// groups instead of only cancelling the current model/tool context. +func (m *ShellManager) CancelRunning(sessionID string) int { + m.mu.Lock() + var jobs []*backgroundProcess + for _, job := range m.jobs { + job.mu.Lock() + running := job.sessionID == sessionID && job.status == processRunning + job.mu.Unlock() + if running { + jobs = append(jobs, job) + } + } + m.mu.Unlock() + stopProcesses(jobs, processCancelled) + return len(jobs) +} + +func (m *ShellManager) reapJobsLocked(now time.Time) { + for id, job := range m.jobs { + job.mu.Lock() + stale := job.status != processRunning && !job.endedAt.IsZero() && now.Sub(job.endedAt) > completedJobTTL + job.mu.Unlock() + if stale { + delete(m.jobs, id) + } + } +} + +type processTool struct{} + +func (processTool) Name() string { return "process" } +func (processTool) Description() string { + return "Inspect and control background terminal processes by handle. Prefer wait or poll over shell sleep when completion time is unknown." +} +func (processTool) Schema() map[string]any { + return schema(map[string]any{ + "action": propEnum("Operation: list session processes, poll status and consume new output, read log at an offset, wait for a state change/completion, or kill the process group.", "list", "poll", "log", "wait", "kill"), + "process_id": prop("string", "Handle returned by terminal(background=true). Required except for list."), + "offset": propDefault("integer", "Absolute log byte offset for action=log.", 0), + "timeout": propDefault("integer", "Maximum seconds to block for action=wait (1-30). This is a wait bound, not an assumed job duration.", 10), + }, "action") +} + +func (processTool) Execute(ctx context.Context, in Input) Result { + if in.Deps == nil || in.Deps.Shell == nil { + return Errorf("terminal backend is not available") + } + var args struct { + Action string `json:"action"` + ProcessID string `json:"process_id"` + Offset int64 `json:"offset"` + Timeout int `json:"timeout"` + } + if err := in.Bind(&args); err != nil { + return Errorf("%v", err) + } + if args.Action == "list" { + return processJSON(in.Deps.Shell.listJobs(in.SessionID)) + } + job, ok := in.Deps.Shell.getJob(in.SessionID, args.ProcessID) + if !ok { + return Errorf("background process %q not found in this session", args.ProcessID) + } + switch args.Action { + case "poll": + return processJSON(job.view(0, true)) + case "log": + if args.Offset < 0 { + return Errorf("offset must be non-negative") + } + return processJSON(job.view(args.Offset, false)) + case "wait": + if args.Timeout <= 0 { + args.Timeout = 10 + } + if args.Timeout > 30 { + args.Timeout = 30 + } + select { + case <-job.done: + case <-time.After(time.Duration(args.Timeout) * time.Second): + case <-ctx.Done(): + } + return processJSON(job.view(0, true)) + case "kill": + job.stop(processCancelled) + select { + case <-job.done: + case <-time.After(3 * time.Second): + } + return processJSON(job.view(0, true)) + default: + return Errorf("unknown action %q", args.Action) + } +} + +func processJSON(v any) Result { + b, err := json.MarshalIndent(v, "", " ") + if err != nil { + return Errorf("encode process state: %v", err) + } + return Result{Content: string(b)} +} diff --git a/internal/tools/background_process_test.go b/internal/tools/background_process_test.go new file mode 100644 index 00000000..5cea8202 --- /dev/null +++ b/internal/tools/background_process_test.go @@ -0,0 +1,357 @@ +package tools + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + "github.com/enowdev/antares/internal/config" +) + +func waitProcessDone(t *testing.T, job *backgroundProcess, timeout time.Duration) { + t.Helper() + select { + case <-job.done: + case <-time.After(timeout): + t.Fatalf("process %s did not finish within %s", job.id, timeout) + } +} + +func TestBackgroundProcessIncrementalOutputAndCompletion(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "printf first; sleep 0.15; printf second", 0) + if err != nil { + t.Fatal(err) + } + + time.Sleep(50 * time.Millisecond) + first := job.view(0, true) + if first.Status != processRunning || first.Output != "first" { + t.Fatalf("first poll = %#v, want running with first", first) + } + waitProcessDone(t, job, 2*time.Second) + final := job.view(0, true) + if final.Status != processCompleted || final.Output != "second" { + t.Fatalf("final poll = %#v, want completed with second", final) + } + if final.ExitCode == nil || *final.ExitCode != 0 { + t.Fatalf("exit code = %v, want 0", final.ExitCode) + } + if again := job.view(0, true); again.Output != "" || again.NextOffset != final.NextOffset { + t.Fatalf("consumed output was returned twice: %#v", again) + } +} + +func TestBackgroundProcessLogOffsetDoesNotConsumePollCursor(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "printf abcdef", 0) + if err != nil { + t.Fatal(err) + } + waitProcessDone(t, job, 2*time.Second) + logView := job.view(3, false) + if logView.Output != "def" || logView.NextOffset != 6 { + t.Fatalf("offset log = %#v", logView) + } + poll := job.view(0, true) + if poll.Output != "abcdef" { + t.Fatalf("log read consumed poll cursor: %q", poll.Output) + } +} + +func TestConcurrentPollsConsumeOutputOnce(t *testing.T) { + job := &backgroundProcess{id: "proc_test", status: processCompleted, startedAt: time.Now()} + _, _ = job.log.Write([]byte("once")) + start := make(chan struct{}) + results := make(chan string, 2) + for i := 0; i < 2; i++ { + go func() { + <-start + results <- job.view(0, true).Output + }() + } + close(start) + a, b := <-results, <-results + if a+b != "once" { + t.Fatalf("concurrent poll output = %q + %q, want exactly one copy", a, b) + } +} + +func TestBackgroundProcessTimeoutUsesRealLifecycle(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "printf started; sleep 30", 100*time.Millisecond) + if err != nil { + t.Fatal(err) + } + waitProcessDone(t, job, 3*time.Second) + got := job.view(0, false) + if got.Status != processTimedOut { + t.Fatalf("status = %s, want %s", got.Status, processTimedOut) + } + if !strings.Contains(got.Output, "started") { + t.Fatalf("partial output lost: %q", got.Output) + } +} + +func TestBackgroundProcessReportsNonZeroExit(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "printf failed; exit 7", 0) + if err != nil { + t.Fatal(err) + } + waitProcessDone(t, job, 2*time.Second) + got := job.view(0, false) + if got.Status != processFailed || got.ExitCode == nil || *got.ExitCode != 7 || got.Output != "failed" { + t.Fatalf("failed process result = %#v", got) + } +} + +func TestBackgroundProcessKillTerminatesProcessGroup(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX process-group fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + dir := t.TempDir() + pidFile := filepath.Join(dir, "child.pid") + stoppedFile := filepath.Join(dir, "child.stopped") + job, err := m.startBackground("session-a", dir, "sh -c 'trap \"echo stopped > child.stopped; exit 0\" TERM; echo $$ > child.pid; while :; do sleep 1; done' & wait", 0) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(2 * time.Second) + for { + if _, err := os.Stat(pidFile); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("child pid file was not created") + } + time.Sleep(10 * time.Millisecond) + } + job.stopAndWait(processCancelled) + waitProcessDone(t, job, time.Second) + if got := job.view(0, false).Status; got != processCancelled { + t.Fatalf("status = %s, want cancelled", got) + } + deadline = time.Now().Add(time.Second) + for { + if _, err := os.Stat(stoppedFile); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("child process did not receive the process-group termination signal") + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestProcessLogPaginationAndRetention(t *testing.T) { + var log processLog + payload := strings.Repeat("x", backgroundReadLimit+123) + if _, err := log.Write([]byte(payload)); err != nil { + t.Fatal(err) + } + first, next, truncated, more := log.read(0) + if len(first) != backgroundReadLimit || next != backgroundReadLimit || truncated || !more { + t.Fatalf("first page = len %d, next %d, truncated %v, more %v", len(first), next, truncated, more) + } + second, end, truncated, more := log.read(next) + if len(second) != 123 || end != int64(len(payload)) || truncated || more { + t.Fatalf("second page = len %d, end %d, truncated %v, more %v", len(second), end, truncated, more) + } + + oversize := strings.Repeat("y", backgroundLogLimit+10) + if _, err := log.Write([]byte(oversize)); err != nil { + t.Fatal(err) + } + _, _, truncated, _ = log.read(0) + if !truncated { + t.Fatal("reader behind the retention window was not told output was truncated") + } +} + +func TestBackgroundProcessHandlesAreSessionScoped(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("owner", t.TempDir(), "true", 0) + if err != nil { + t.Fatal(err) + } + waitProcessDone(t, job, 2*time.Second) + if _, ok := m.getJob("other", job.id); ok { + t.Fatal("another session accessed a background process handle") + } + if _, ok := m.getJob("owner", job.id); !ok { + t.Fatal("owner could not access its process handle") + } +} + +func TestProcessToolPollAndWait(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "printf tool-ok", 0) + if err != nil { + t.Fatal(err) + } + args, _ := json.Marshal(map[string]any{"action": "wait", "process_id": job.id, "timeout": 2}) + result := (processTool{}).Execute(context.Background(), Input{ + Args: args, SessionID: "session-a", Deps: &Deps{Shell: m}, Emit: func(Progress) {}, + }) + if result.IsError { + t.Fatalf("process tool failed: %s", result.Content) + } + var got processView + if err := json.Unmarshal([]byte(result.Content), &got); err != nil { + t.Fatal(err) + } + if got.Status != processCompleted || got.Output != "tool-ok" { + t.Fatalf("process result = %#v", got) + } +} + +func TestProcessWaitReturnsRunningWhenBoundExpires(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + job, err := m.startBackground("session-a", t.TempDir(), "sleep 30", 0) + if err != nil { + t.Fatal(err) + } + args, _ := json.Marshal(map[string]any{"action": "wait", "process_id": job.id, "timeout": 1}) + started := time.Now() + result := (processTool{}).Execute(context.Background(), Input{ + Args: args, SessionID: "session-a", Deps: &Deps{Shell: m}, Emit: func(Progress) {}, + }) + if result.IsError { + t.Fatal(result.Content) + } + var got processView + if err := json.Unmarshal([]byte(result.Content), &got); err != nil { + t.Fatal(err) + } + elapsed := time.Since(started) + if got.Status != processRunning || elapsed < 900*time.Millisecond || elapsed > 2*time.Second { + t.Fatalf("bounded wait returned after %s with status %s", elapsed, got.Status) + } +} + +func TestTerminalToolStartsManagedBackgroundProcess(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + cfg := &config.Config{Terminal: config.Terminal{Sandbox: "none"}} + m := NewShellManager(cfg.Terminal) + t.Cleanup(m.CloseAll) + args, _ := json.Marshal(map[string]any{"command": "printf managed", "background": true}) + result := (terminalTool{}).Execute(context.Background(), Input{ + Args: args, SessionID: "session-a", Workspace: t.TempDir(), + Deps: &Deps{Shell: m, Config: cfg}, Emit: func(Progress) {}, + }) + if result.IsError { + t.Fatalf("terminal background start failed: %s", result.Content) + } + var started processView + if err := json.Unmarshal([]byte(result.Content), &started); err != nil { + t.Fatal(err) + } + if started.ProcessID == "" || (started.Status != processRunning && started.Status != processCompleted) { + t.Fatalf("start result = %#v", started) + } + job, ok := m.getJob("session-a", started.ProcessID) + if !ok { + t.Fatal("returned process handle is not registered") + } + waitProcessDone(t, job, 2*time.Second) +} + +func TestProcessToolIsExposedWithTerminalToolsets(t *testing.T) { + for _, set := range []string{"default", "coding", "security", "reverse", "vibecoder"} { + found := false + for _, name := range ExpandToolset(set) { + if name == "process" { + found = true + break + } + } + if !found { + t.Errorf("toolset %q exposes terminal but not process", set) + } + } +} + +func TestRegistryAddsProcessCompanionForExplicitTerminal(t *testing.T) { + r := NewRegistry() + r.Register(terminalTool{}) + r.Register(processTool{}) + resolved := r.Resolve("minimal", []string{"terminal"}, nil) + seen := map[string]bool{} + for _, tool := range resolved { + seen[tool.Name()] = true + } + if !seen["terminal"] || !seen["process"] { + t.Fatalf("explicit terminal resolved to tools %v", seen) + } + + resolved = r.Resolve("default", nil, []string{"process"}) + for _, tool := range resolved { + if tool.Name() == "process" { + t.Fatal("explicit process disable was ignored") + } + } +} + +func TestCancelRunningStopsOnlyOwnedSession(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX command fixture") + } + m := NewShellManager(config.Terminal{Sandbox: "none"}) + t.Cleanup(m.CloseAll) + owned, err := m.startBackground("owner", t.TempDir(), "sleep 30", 0) + if err != nil { + t.Fatal(err) + } + other, err := m.startBackground("other", t.TempDir(), "sleep 30", 0) + if err != nil { + t.Fatal(err) + } + if got := m.CancelRunning("owner"); got != 1 { + t.Fatalf("CancelRunning stopped %d processes, want 1", got) + } + waitProcessDone(t, owned, time.Second) + if got := owned.view(0, false).Status; got != processCancelled { + t.Fatalf("owned status = %s", got) + } + if got := other.view(0, false).Status; got != processRunning { + t.Fatalf("other session status = %s, want running", got) + } +} diff --git a/internal/tools/background_process_unix.go b/internal/tools/background_process_unix.go new file mode 100644 index 00000000..caf7214c --- /dev/null +++ b/internal/tools/background_process_unix.go @@ -0,0 +1,27 @@ +//go:build !windows + +package tools + +import ( + "os/exec" + "syscall" +) + +func configureProcessGroup(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} +} + +func terminateProcessGroup(cmd *exec.Cmd) { + if cmd == nil || cmd.Process == nil { + return + } + // Negative PID addresses the process group created for the background job, + // so child analyzers/build tools do not survive a cancellation. + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM) +} + +func killProcessGroup(cmd *exec.Cmd) { + if cmd != nil && cmd.Process != nil { + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + } +} diff --git a/internal/tools/background_process_windows.go b/internal/tools/background_process_windows.go new file mode 100644 index 00000000..604fcc05 --- /dev/null +++ b/internal/tools/background_process_windows.go @@ -0,0 +1,15 @@ +//go:build windows + +package tools + +import "os/exec" + +func configureProcessGroup(_ *exec.Cmd) {} + +func terminateProcessGroup(cmd *exec.Cmd) { + if cmd != nil && cmd.Process != nil { + _ = cmd.Process.Kill() + } +} + +func killProcessGroup(cmd *exec.Cmd) { terminateProcessGroup(cmd) } diff --git a/internal/tools/register.go b/internal/tools/register.go index 2b1827be..38db4fdd 100644 --- a/internal/tools/register.go +++ b/internal/tools/register.go @@ -12,7 +12,7 @@ func init() { readFileTool{}, writeFileTool{}, editFileTool{}, listFilesTool{}, readDocumentTool{}, globTool{}, grepTool{}, - terminalTool{}, + terminalTool{}, processTool{}, webFetchTool{}, webSearchTool{}, httpRequestTool{}, memoryTool{}, sessionSearchTool{}, todoTool{}, skillTool{}, ragSearchTool{}, ragIndexTool{}, diff --git a/internal/tools/registry.go b/internal/tools/registry.go index d61935c4..eb8c782c 100644 --- a/internal/tools/registry.go +++ b/internal/tools/registry.go @@ -176,6 +176,12 @@ func (r *Registry) Resolve(toolset string, enabled, disabled []string) []Tool { } delete(want, n) } + // process is terminal's job-control companion. Custom roles that explicitly + // enable terminal should not silently lose the only safe way to observe or + // cancel an unknown-duration command. An explicit process disable still wins. + if want["terminal"] && !contains(disabled, "process") { + want["process"] = true + } r.mu.RLock() defer r.mu.RUnlock() @@ -203,7 +209,7 @@ var Toolsets = map[string][]string{ "minimal": {"read_file", "list_files", "grep", "todo"}, "coding": { "read_file", "read_document", "write_file", "edit_file", "list_files", "glob", "grep", - "terminal", "todo", "board", "project_info", "set_soul", "skill", "delegate_task", "task", "list_roles", "diagnostics", "http_request", "ask_user", "schedule", + "terminal", "process", "todo", "board", "project_info", "set_soul", "skill", "delegate_task", "task", "list_roles", "diagnostics", "http_request", "ask_user", "schedule", }, "research": { "read_file", "read_document", "web_search", "web_fetch", "http_request", "browser", "grep", "todo", "memory", @@ -212,7 +218,7 @@ var Toolsets = map[string][]string{ }, "browser": {"browser", "web_search", "web_fetch", "http_request", "read_file", "write_file", "todo", "report_finding", "add_intel", "methodology_status"}, "security": { - "read_file", "write_file", "list_files", "glob", "grep", "terminal", + "read_file", "write_file", "list_files", "glob", "grep", "terminal", "process", "web_search", "web_fetch", "http_request", "browser", "todo", "skill", "report_finding", "triage_finding", "add_intel", "methodology_status", "delegate_task", "task", "list_roles", "diagnostics", "ask_user", "vps_run", @@ -226,10 +232,10 @@ var Toolsets = map[string][]string{ }, "reverse": { "re_info", "re_strings", "re_analyze", "re_decompile", "check_dependencies", - "read_file", "list_files", "glob", "grep", "terminal", "todo", "report_finding", + "read_file", "list_files", "glob", "grep", "terminal", "process", "todo", "report_finding", }, "vibecoder": { - "web_fetch", "http_request", "browser", "web_search", "terminal", + "web_fetch", "http_request", "browser", "web_search", "terminal", "process", "read_file", "write_file", "list_files", "glob", "grep", "check_dependencies", "todo", "report_finding", "add_intel", "methodology_status", "skill", "rag_search", }, @@ -239,7 +245,7 @@ var Toolsets = map[string][]string{ }, "default": { "read_file", "read_document", "write_file", "edit_file", "list_files", "glob", "grep", - "terminal", "web_search", "web_fetch", "http_request", "browser", "todo", "board", "project_info", "set_soul", "memory", "list_proxies", "vps_run", + "terminal", "process", "web_search", "web_fetch", "http_request", "browser", "todo", "board", "project_info", "set_soul", "memory", "list_proxies", "vps_run", "session_search", "rag_search", "rag_index", "skill", "delegate_task", "task", "list_roles", "image_generate", "view_image", "speak", "transcribe", "computer", "diagnostics", "ask_user", "schedule", "osint_dns", "osint_dorks", "osint_whois", "osint_ip", "osint_username", "osint_github", "osint_email", "osint_email_full", "osint_breach", "osint_shodan", "osint_reputation", "osint_crypto", "osint_domain", "osint_phone", "osint_scrape", "osint_paste", "osint_footprint", "osint_pivot", "osint_google", "osint_dorks_live", "check_dependencies", "re_info", "re_strings", "re_analyze", "re_decompile", "solve_captcha", "intercept", }, diff --git a/internal/tools/shell.go b/internal/tools/shell.go index 2d457858..8392d900 100644 --- a/internal/tools/shell.go +++ b/internal/tools/shell.go @@ -22,6 +22,7 @@ import ( type ShellManager struct { mu sync.Mutex sessions map[string]*shellSession + jobs map[string]*backgroundProcess cfg config.Terminal // sandboxOnce keeps a confinement warning from repeating on every shell. sandboxOnce sync.Once @@ -39,7 +40,7 @@ type httpShimEnv struct { // NewShellManager builds a manager for the given terminal config. func NewShellManager(cfg config.Terminal) *ShellManager { - return &ShellManager{sessions: map[string]*shellSession{}, cfg: cfg} + return &ShellManager{sessions: map[string]*shellSession{}, jobs: map[string]*backgroundProcess{}, cfg: cfg} } // EnableHTTPShim makes new shells route curl/wget through the fingerprinted @@ -218,10 +219,18 @@ func (m *ShellManager) Close(id string) { m.mu.Lock() s, ok := m.sessions[id] delete(m.sessions, id) + var jobs []*backgroundProcess + for jobID, job := range m.jobs { + if job.sessionID == id { + jobs = append(jobs, job) + delete(m.jobs, jobID) + } + } m.mu.Unlock() if ok { s.terminate() } + stopProcesses(jobs, processCancelled) } // CloseAll terminates every live shell. @@ -232,10 +241,16 @@ func (m *ShellManager) CloseAll() { all = append(all, s) } m.sessions = map[string]*shellSession{} + jobs := make([]*backgroundProcess, 0, len(m.jobs)) + for _, job := range m.jobs { + jobs = append(jobs, job) + } + m.jobs = map[string]*backgroundProcess{} m.mu.Unlock() for _, s := range all { s.terminate() } + stopProcesses(jobs, processCancelled) } // Active reports how many shells are currently running. @@ -349,20 +364,22 @@ type terminalTool struct{} func (terminalTool) Name() string { return "terminal" } func (terminalTool) Description() string { - return "Run a shell command in a persistent session. Working directory, environment variables, and shell state persist across calls." + return "Run a shell command. Foreground commands use a persistent shell; set background=true for work of unknown duration and monitor the returned process_id with the process tool instead of shell sleep." } func (terminalTool) RequiresApproval() bool { return true } func (terminalTool) Schema() map[string]any { return schema(map[string]any{ - "command": prop("string", "Shell command to execute."), - "timeout": propDefault("integer", "Seconds before the command is aborted.", 300), + "command": prop("string", "Shell command to execute."), + "timeout": propDefault("integer", "Foreground: seconds before aborting. Background: optional total runtime limit; zero means no runtime limit.", 0), + "background": propDefault("boolean", "Start as a managed background process and return immediately with a process_id. Use for builds, analysis, imports, servers, and other commands whose duration is unknown.", false), }, "command") } func (terminalTool) Execute(ctx context.Context, in Input) Result { var args struct { - Command string `json:"command"` - Timeout int `json:"timeout"` + Command string `json:"command"` + Timeout int `json:"timeout"` + Background bool `json:"background"` } if err := in.Bind(&args); err != nil { return Errorf("%v", err) @@ -380,6 +397,16 @@ func (terminalTool) Execute(ctx context.Context, in Input) Result { return Errorf("command blocked by policy (matched %q). Adjust terminal.blocked_commands to allow it.", blocked) } } + if args.Background { + if args.Timeout < 0 { + return Errorf("timeout must be non-negative") + } + job, err := in.Deps.Shell.startBackground(in.SessionID, in.Workspace, cmdText, time.Duration(args.Timeout)*time.Second) + if err != nil { + return Errorf("cannot start background process: %v", err) + } + return processJSON(job.view(0, false)) + } timeout := time.Duration(args.Timeout) * time.Second if timeout <= 0 {