mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
feat(agent): add process execution API and rewrite execute tool (#22416)
## Summary Adds a new agent-side process management HTTP API and rewrites the chat execute tool to use it instead of SSH sessions. ## What changed ### New agent/agentproc/ package - **headtail.go** — Thread-safe io.Writer with bounded memory (16KB head + 16KB tail ring buffer). Provides LLM-ready output with truncation metadata and long-line truncation at 2048 bytes. - **headtail_test.go** — 16 tests including race detector coverage for concurrent writes. - **process.go** — Manager + Process types for lifecycle management using agentexec.Execer for proper OOM/nice scores. - **api.go** — HTTP API following the agentfiles chi router pattern. 4 endpoints: start, list, output, signal. ### Agent wiring (agent/agent.go, agent/api.go) Mounts the process API at /api/v0/processes, mirroring how agentfiles is mounted. ### SDK (codersdk/workspacesdk/agentconn.go) 4 new AgentConn interface methods + 7 request/response types: - StartProcess, ListProcesses, ProcessOutput, SignalProcess ### Execute tool rewrite (coderd/chatd/chattool/execute.go) - SSH to Agent API: conn.StartProcess() + conn.ProcessOutput() polling - New parameters: workdir, run_in_background - Structured response: success, exit_code, wall_duration_ms, error, truncated, note, background_process_id - Non-interactive env vars: GIT_EDITOR=true, TERM=dumb, NO_COLOR=1, PAGER=cat, etc. - Output truncation: HeadTailBuffer caps at 32KB for LLM consumption - File-dump detection with advisory notes suggesting read_file - Default timeout: 60s to 10s - Foreground polling: 200ms intervals until exit or timeout ## Architecture State lives on the agent, surviving coderd failover and instance changes. Any coderd replica can query any agent via HTTP over tailnet.
This commit is contained in:
@@ -2071,6 +2071,15 @@ func (p *Server) runChat(
|
||||
chattool.Execute(chattool.ExecuteOptions{
|
||||
GetWorkspaceConn: getWorkspaceConn,
|
||||
}),
|
||||
chattool.ProcessOutput(chattool.ProcessToolOptions{
|
||||
GetWorkspaceConn: getWorkspaceConn,
|
||||
}),
|
||||
chattool.ProcessList(chattool.ProcessToolOptions{
|
||||
GetWorkspaceConn: getWorkspaceConn,
|
||||
}),
|
||||
chattool.ProcessSignal(chattool.ProcessToolOptions{
|
||||
GetWorkspaceConn: getWorkspaceConn,
|
||||
}),
|
||||
}
|
||||
tools = append(tools, p.subagentTools(func() database.Chat {
|
||||
return chat
|
||||
|
||||
@@ -2,32 +2,88 @@ package chattool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"charm.land/fantasy"
|
||||
"golang.org/x/crypto/ssh"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/coder/coder/v2/codersdk/workspacesdk"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultExecuteTimeout = 60 * time.Second
|
||||
chatAgentEnvVar = "CODER_CHAT_AGENT"
|
||||
gitAuthRequiredPrefix = "CODER_GITAUTH_REQUIRED:"
|
||||
authRequiredResultReason = "authentication_required"
|
||||
// defaultTimeout is the default timeout for command
|
||||
// execution.
|
||||
defaultTimeout = 10 * time.Second
|
||||
|
||||
// maxOutputToModel is the maximum output sent to the LLM.
|
||||
maxOutputToModel = 32 << 10 // 32KB
|
||||
|
||||
// pollInterval is how often we check for process completion
|
||||
// in foreground mode.
|
||||
pollInterval = 200 * time.Millisecond
|
||||
)
|
||||
|
||||
// nonInteractiveEnvVars are set on every process to prevent
|
||||
// interactive prompts that would hang a headless execution.
|
||||
var nonInteractiveEnvVars = map[string]string{
|
||||
"GIT_EDITOR": "true",
|
||||
"GIT_SEQUENCE_EDITOR": "true",
|
||||
"EDITOR": "true",
|
||||
"VISUAL": "true",
|
||||
"GIT_TERMINAL_PROMPT": "0",
|
||||
"NO_COLOR": "1",
|
||||
"TERM": "dumb",
|
||||
"PAGER": "cat",
|
||||
"GIT_PAGER": "cat",
|
||||
}
|
||||
|
||||
// fileDumpPatterns detects commands that dump entire files.
|
||||
// When matched, a note is added suggesting read_file instead.
|
||||
var fileDumpPatterns = []*regexp.Regexp{
|
||||
regexp.MustCompile(`^cat\s+`),
|
||||
regexp.MustCompile(`^(rg|grep)\s+.*--include-all`),
|
||||
regexp.MustCompile(`^(rg|grep)\s+-l\s+`),
|
||||
}
|
||||
|
||||
// ExecuteResult is the structured response from the execute
|
||||
// tool.
|
||||
type ExecuteResult struct {
|
||||
Success bool `json:"success"`
|
||||
Output string `json:"output,omitempty"`
|
||||
ExitCode int `json:"exit_code"`
|
||||
WallDurationMs int64 `json:"wall_duration_ms"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Truncated *workspacesdk.ProcessTruncation `json:"truncated,omitempty"`
|
||||
Note string `json:"note,omitempty"`
|
||||
BackgroundProcessID string `json:"background_process_id,omitempty"`
|
||||
}
|
||||
|
||||
// ExecuteOptions configures the execute tool.
|
||||
type ExecuteOptions struct {
|
||||
GetWorkspaceConn func(context.Context) (workspacesdk.AgentConn, error)
|
||||
DefaultTimeout time.Duration
|
||||
}
|
||||
|
||||
type ExecuteArgs struct {
|
||||
Command string `json:"command"`
|
||||
TimeoutSeconds *int `json:"timeout_seconds,omitempty"`
|
||||
// ProcessToolOptions configures a process management tool
|
||||
// (process_output, process_list, or process_signal). Each of
|
||||
// these tools only needs a workspace connection resolver.
|
||||
type ProcessToolOptions struct {
|
||||
GetWorkspaceConn func(context.Context) (workspacesdk.AgentConn, error)
|
||||
}
|
||||
|
||||
// ExecuteArgs are the parameters accepted by the execute tool.
|
||||
type ExecuteArgs struct {
|
||||
Command string `json:"command"`
|
||||
Timeout *string `json:"timeout,omitempty"`
|
||||
WorkDir *string `json:"workdir,omitempty"`
|
||||
RunInBackground *bool `json:"run_in_background,omitempty"`
|
||||
}
|
||||
|
||||
// Execute returns an AgentTool that runs a shell command in the
|
||||
// workspace via the agent HTTP API.
|
||||
func Execute(options ExecuteOptions) fantasy.AgentTool {
|
||||
return fantasy.NewAgentTool(
|
||||
"execute",
|
||||
@@ -49,85 +105,342 @@ func executeTool(
|
||||
ctx context.Context,
|
||||
conn workspacesdk.AgentConn,
|
||||
args ExecuteArgs,
|
||||
defaultTimeout time.Duration,
|
||||
optTimeout time.Duration,
|
||||
) fantasy.ToolResponse {
|
||||
if args.Command == "" {
|
||||
return fantasy.NewTextErrorResponse("command is required")
|
||||
}
|
||||
|
||||
timeout := defaultTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = defaultExecuteTimeout
|
||||
// Build the environment map for the process request.
|
||||
env := make(map[string]string, len(nonInteractiveEnvVars)+1)
|
||||
env["CODER_CHAT_AGENT"] = "true"
|
||||
for k, v := range nonInteractiveEnvVars {
|
||||
env[k] = v
|
||||
}
|
||||
if args.TimeoutSeconds != nil {
|
||||
timeout = time.Duration(*args.TimeoutSeconds) * time.Second
|
||||
}
|
||||
cmdCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
output, exitCode, err := runCommand(cmdCtx, conn, args.Command)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error())
|
||||
background := args.RunInBackground != nil && *args.RunInBackground
|
||||
|
||||
var workDir string
|
||||
if args.WorkDir != nil {
|
||||
workDir = *args.WorkDir
|
||||
}
|
||||
return toolResponse(map[string]any{
|
||||
"output": output,
|
||||
"exit_code": exitCode,
|
||||
})
|
||||
|
||||
if background {
|
||||
return executeBackground(ctx, conn, args.Command, workDir, env)
|
||||
}
|
||||
return executeForeground(ctx, conn, args, optTimeout, workDir, env)
|
||||
}
|
||||
|
||||
func runCommand(
|
||||
// executeBackground starts a process in the background and
|
||||
// returns immediately with the process ID.
|
||||
func executeBackground(
|
||||
ctx context.Context,
|
||||
conn workspacesdk.AgentConn,
|
||||
command string,
|
||||
) (string, int, error) {
|
||||
sshClient, err := conn.SSHClient(ctx)
|
||||
workDir string,
|
||||
env map[string]string,
|
||||
) fantasy.ToolResponse {
|
||||
resp, err := conn.StartProcess(ctx, workspacesdk.StartProcessRequest{
|
||||
Command: command,
|
||||
WorkDir: workDir,
|
||||
Env: env,
|
||||
Background: true,
|
||||
})
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
return errorResult(fmt.Sprintf("start background process: %v", err))
|
||||
}
|
||||
defer sshClient.Close()
|
||||
|
||||
session, err := sshClient.NewSession()
|
||||
result := ExecuteResult{
|
||||
Success: true,
|
||||
BackgroundProcessID: resp.ID,
|
||||
}
|
||||
data, err := json.Marshal(result)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
return fantasy.NewTextErrorResponse(err.Error())
|
||||
}
|
||||
defer session.Close()
|
||||
if err := session.Setenv(chatAgentEnvVar, "true"); err != nil {
|
||||
return "", 0, xerrors.Errorf("set %s: %w", chatAgentEnvVar, err)
|
||||
return fantasy.NewTextResponse(string(data))
|
||||
}
|
||||
|
||||
// executeForeground starts a process and polls for its
|
||||
// completion, enforcing the configured timeout.
|
||||
func executeForeground(
|
||||
ctx context.Context,
|
||||
conn workspacesdk.AgentConn,
|
||||
args ExecuteArgs,
|
||||
optTimeout time.Duration,
|
||||
workDir string,
|
||||
env map[string]string,
|
||||
) fantasy.ToolResponse {
|
||||
timeout := optTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = defaultTimeout
|
||||
}
|
||||
|
||||
resultCh := make(chan struct {
|
||||
output string
|
||||
exitCode int
|
||||
err error
|
||||
}, 1)
|
||||
|
||||
go func() {
|
||||
output, err := session.CombinedOutput(command)
|
||||
exitCode := 0
|
||||
if args.Timeout != nil {
|
||||
parsed, err := time.ParseDuration(*args.Timeout)
|
||||
if err != nil {
|
||||
var exitErr *ssh.ExitError
|
||||
if xerrors.As(err, &exitErr) {
|
||||
exitCode = exitErr.ExitStatus()
|
||||
} else {
|
||||
exitCode = 1
|
||||
return fantasy.NewTextErrorResponse(
|
||||
fmt.Sprintf("invalid timeout %q: %v", *args.Timeout, err),
|
||||
)
|
||||
}
|
||||
timeout = parsed
|
||||
}
|
||||
|
||||
cmdCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
start := time.Now()
|
||||
|
||||
resp, err := conn.StartProcess(cmdCtx, workspacesdk.StartProcessRequest{
|
||||
Command: args.Command,
|
||||
WorkDir: workDir,
|
||||
Env: env,
|
||||
Background: false,
|
||||
})
|
||||
if err != nil {
|
||||
return errorResult(fmt.Sprintf("start process: %v", err))
|
||||
}
|
||||
|
||||
result := pollProcess(cmdCtx, conn, resp.ID, timeout)
|
||||
result.WallDurationMs = time.Since(start).Milliseconds()
|
||||
|
||||
// Add an advisory note for file-dump commands.
|
||||
if note := detectFileDump(args.Command); note != "" {
|
||||
result.Note = note
|
||||
}
|
||||
|
||||
data, err := json.Marshal(result)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error())
|
||||
}
|
||||
return fantasy.NewTextResponse(string(data))
|
||||
}
|
||||
|
||||
// truncateOutput safely truncates output to maxOutputToModel,
|
||||
// ensuring the result is valid UTF-8 even if the cut falls in
|
||||
// the middle of a multi-byte character.
|
||||
func truncateOutput(output string) string {
|
||||
if len(output) > maxOutputToModel {
|
||||
output = strings.ToValidUTF8(output[:maxOutputToModel], "")
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
// pollProcess polls for process output until the process exits
|
||||
// or the context times out.
|
||||
func pollProcess(
|
||||
ctx context.Context,
|
||||
conn workspacesdk.AgentConn,
|
||||
processID string,
|
||||
timeout time.Duration,
|
||||
) ExecuteResult {
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// Timeout — get whatever output we have. Use a
|
||||
// fresh context since cmdCtx is already canceled.
|
||||
bgCtx, bgCancel := context.WithTimeout(
|
||||
context.Background(),
|
||||
5*time.Second,
|
||||
)
|
||||
outputResp, _ := conn.ProcessOutput(bgCtx, processID)
|
||||
bgCancel()
|
||||
output := truncateOutput(outputResp.Output)
|
||||
return ExecuteResult{
|
||||
Success: false,
|
||||
Output: output,
|
||||
ExitCode: -1,
|
||||
Error: fmt.Sprintf("command timed out after %s", timeout),
|
||||
Truncated: outputResp.Truncated,
|
||||
}
|
||||
case <-ticker.C:
|
||||
outputResp, err := conn.ProcessOutput(ctx, processID)
|
||||
if err != nil {
|
||||
return ExecuteResult{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("get process output: %v", err),
|
||||
}
|
||||
}
|
||||
if !outputResp.Running {
|
||||
exitCode := 0
|
||||
if outputResp.ExitCode != nil {
|
||||
exitCode = *outputResp.ExitCode
|
||||
}
|
||||
output := truncateOutput(outputResp.Output)
|
||||
return ExecuteResult{
|
||||
Success: exitCode == 0,
|
||||
Output: output,
|
||||
ExitCode: exitCode,
|
||||
Truncated: outputResp.Truncated,
|
||||
}
|
||||
}
|
||||
}
|
||||
resultCh <- struct {
|
||||
output string
|
||||
exitCode int
|
||||
err error
|
||||
}{
|
||||
output: string(output),
|
||||
exitCode: exitCode,
|
||||
err: err,
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
_ = session.Close()
|
||||
return "", 0, ctx.Err()
|
||||
case result := <-resultCh:
|
||||
return result.output, result.exitCode, result.err
|
||||
}
|
||||
}
|
||||
|
||||
// errorResult builds a ToolResponse from an ExecuteResult with
|
||||
// an error message.
|
||||
func errorResult(msg string) fantasy.ToolResponse {
|
||||
data, err := json.Marshal(ExecuteResult{
|
||||
Success: false,
|
||||
Error: msg,
|
||||
})
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(msg)
|
||||
}
|
||||
return fantasy.NewTextResponse(string(data))
|
||||
}
|
||||
|
||||
// detectFileDump checks whether the command matches a file-dump
|
||||
// pattern and returns an advisory note, or empty string if no
|
||||
// match.
|
||||
func detectFileDump(command string) string {
|
||||
for _, pat := range fileDumpPatterns {
|
||||
if pat.MatchString(command) {
|
||||
return "Consider using read_file instead of " +
|
||||
"dumping file contents with shell commands."
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ProcessOutputArgs are the parameters accepted by the
|
||||
// process_output tool.
|
||||
type ProcessOutputArgs struct {
|
||||
ProcessID string `json:"process_id"`
|
||||
}
|
||||
|
||||
// ProcessOutput returns an AgentTool that retrieves the output
|
||||
// of a background process by its ID.
|
||||
func ProcessOutput(options ProcessToolOptions) fantasy.AgentTool {
|
||||
return fantasy.NewAgentTool(
|
||||
"process_output",
|
||||
"Retrieve output from a background process. "+
|
||||
"Use the process_id returned by execute with "+
|
||||
"run_in_background=true. Returns the current output, "+
|
||||
"whether the process is still running, and the exit "+
|
||||
"code if it has finished.",
|
||||
func(ctx context.Context, args ProcessOutputArgs, _ fantasy.ToolCall) (fantasy.ToolResponse, error) {
|
||||
if options.GetWorkspaceConn == nil {
|
||||
return fantasy.NewTextErrorResponse("workspace connection resolver is not configured"), nil
|
||||
}
|
||||
if args.ProcessID == "" {
|
||||
return fantasy.NewTextErrorResponse("process_id is required"), nil
|
||||
}
|
||||
conn, err := options.GetWorkspaceConn(ctx)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
resp, err := conn.ProcessOutput(ctx, args.ProcessID)
|
||||
if err != nil {
|
||||
return errorResult(fmt.Sprintf("get process output: %v", err)), nil
|
||||
}
|
||||
output := truncateOutput(resp.Output)
|
||||
exitCode := 0
|
||||
if resp.ExitCode != nil {
|
||||
exitCode = *resp.ExitCode
|
||||
}
|
||||
result := ExecuteResult{
|
||||
Success: !resp.Running && exitCode == 0,
|
||||
Output: output,
|
||||
ExitCode: exitCode,
|
||||
Truncated: resp.Truncated,
|
||||
}
|
||||
if resp.Running {
|
||||
// Process is still running — success is not
|
||||
// yet determined.
|
||||
result.Success = true
|
||||
result.Note = "process is still running"
|
||||
}
|
||||
data, err := json.Marshal(result)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
return fantasy.NewTextResponse(string(data)), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// ProcessList returns an AgentTool that lists all tracked
|
||||
// processes on the workspace agent.
|
||||
func ProcessList(options ProcessToolOptions) fantasy.AgentTool {
|
||||
return fantasy.NewAgentTool(
|
||||
"process_list",
|
||||
"List all tracked processes in the workspace. "+
|
||||
"Returns process IDs, commands, status (running or "+
|
||||
"exited), and exit codes. Use this to discover "+
|
||||
"background processes or check which processes are "+
|
||||
"still running.",
|
||||
func(ctx context.Context, _ struct{}, _ fantasy.ToolCall) (fantasy.ToolResponse, error) {
|
||||
if options.GetWorkspaceConn == nil {
|
||||
return fantasy.NewTextErrorResponse("workspace connection resolver is not configured"), nil
|
||||
}
|
||||
conn, err := options.GetWorkspaceConn(ctx)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
resp, err := conn.ListProcesses(ctx)
|
||||
if err != nil {
|
||||
return errorResult(fmt.Sprintf("list processes: %v", err)), nil
|
||||
}
|
||||
data, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
return fantasy.NewTextResponse(string(data)), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
// ProcessSignalArgs are the parameters accepted by the
|
||||
// process_signal tool.
|
||||
type ProcessSignalArgs struct {
|
||||
ProcessID string `json:"process_id"`
|
||||
Signal string `json:"signal"`
|
||||
}
|
||||
|
||||
// ProcessSignal returns an AgentTool that sends a signal to a
|
||||
// tracked process on the workspace agent.
|
||||
func ProcessSignal(options ProcessToolOptions) fantasy.AgentTool {
|
||||
return fantasy.NewAgentTool(
|
||||
"process_signal",
|
||||
"Send a signal to a background process. "+
|
||||
"Use \"terminate\" (SIGTERM) for graceful shutdown "+
|
||||
"or \"kill\" (SIGKILL) to force stop. Use the "+
|
||||
"process_id returned by execute with "+
|
||||
"run_in_background=true or from process_list.",
|
||||
func(ctx context.Context, args ProcessSignalArgs, _ fantasy.ToolCall) (fantasy.ToolResponse, error) {
|
||||
if options.GetWorkspaceConn == nil {
|
||||
return fantasy.NewTextErrorResponse("workspace connection resolver is not configured"), nil
|
||||
}
|
||||
if args.ProcessID == "" {
|
||||
return fantasy.NewTextErrorResponse("process_id is required"), nil
|
||||
}
|
||||
if args.Signal != "terminate" && args.Signal != "kill" {
|
||||
return fantasy.NewTextErrorResponse(
|
||||
"signal must be \"terminate\" or \"kill\"",
|
||||
), nil
|
||||
}
|
||||
conn, err := options.GetWorkspaceConn(ctx)
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
if err := conn.SignalProcess(ctx, args.ProcessID, args.Signal); err != nil {
|
||||
return errorResult(fmt.Sprintf("signal process: %v", err)), nil
|
||||
}
|
||||
data, err := json.Marshal(map[string]any{
|
||||
"success": true,
|
||||
"message": fmt.Sprintf(
|
||||
"signal %q sent to process %s",
|
||||
args.Signal, args.ProcessID,
|
||||
),
|
||||
})
|
||||
if err != nil {
|
||||
return fantasy.NewTextErrorResponse(err.Error()), nil
|
||||
}
|
||||
return fantasy.NewTextResponse(string(data)), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user