From 41c452ad84a99a3fda75bb0e8dcbd17a2d3bb122 Mon Sep 17 00:00:00 2001 From: Piyush Jagadish Bag Date: Tue, 30 Jun 2026 00:55:22 -0700 Subject: [PATCH] feat: stream remote agent-send and materialize stderr live Remote wait-mode agent-send buffered the full HTTP response before printing. Route default remote wait through SSH with live stdout/stderr, matching local docker exec behavior. Keep the HTTP agent-send path for --session-id and --new-session. Stream materialize stderr during runs. Closes #156 --- go/cmd/amika/sandbox/sandbox_agent.go | 27 +++++++++++--- go/cmd/amika/sandbox/sandbox_agent_test.go | 21 +++++++---- go/internal/materialize/materialize.go | 22 +---------- go/internal/ssh/ssh.go | 30 +++++++++++++++ go/internal/ssh/ssh_test.go | 43 ++++++++++++++++++++++ 5 files changed, 111 insertions(+), 32 deletions(-) create mode 100644 go/internal/ssh/ssh_test.go diff --git a/go/cmd/amika/sandbox/sandbox_agent.go b/go/cmd/amika/sandbox/sandbox_agent.go index d32f1cb8..a281605e 100644 --- a/go/cmd/amika/sandbox/sandbox_agent.go +++ b/go/cmd/amika/sandbox/sandbox_agent.go @@ -115,8 +115,8 @@ func buildAgentShellCmd(message string, noWait bool, workdir string, agent agent return cmd } -func buildRemoteAgentShellCmd(message string, noWait bool, workdir string, agent agentConfig, opts agentRunOpts) string { - agentStr := strings.Join(agentCmdPartsWithOpts(agent, fmt.Sprintf("%q", message), opts, !noWait), " ") +func buildRemoteAgentShellCmd(message string, noWait bool, workdir string, agent agentConfig, opts agentRunOpts, jsonOutput bool) string { + agentStr := strings.Join(agentCmdPartsWithOpts(agent, fmt.Sprintf("%q", message), opts, jsonOutput), " ") cmd := fmt.Sprintf("cd %s && %s", workdir, agentStr) if noWait { sessionName := fmt.Sprintf("amika-agent-send-%d", time.Now().UnixNano()) @@ -125,12 +125,29 @@ func buildRemoteAgentShellCmd(message string, noWait bool, workdir string, agent return cmd } -func runRemoteAgentSend(client *apiclient.Client, name, message string, noWait bool, workdir string, agent agentConfig, opts agentRunOpts, stdout io.Writer) error { +func runRemoteAgentSend(client *apiclient.Client, name, message string, noWait bool, workdir string, agent agentConfig, opts agentRunOpts, stdout, stderr io.Writer) error { if noWait { - shellCmd := buildRemoteAgentShellCmd(message, noWait, workdir, agent, opts) + shellCmd := buildRemoteAgentShellCmd(message, noWait, workdir, agent, opts, false) return ssh.ExecSSH(client, name, false, []string{shellCmd}) } + // Server-managed sessions use the synchronous agent-send API so session IDs + // and structured responses stay on the control plane. + if opts.NewSession || opts.SessionID != "" { + return runRemoteAgentSendHTTP(client, name, message, agent, opts, stdout) + } + + shellCmd := buildRemoteAgentShellCmd(message, false, workdir, agent, opts, false) + if err := ssh.RunSSH(client, name, []string{shellCmd}, nil, stdout, stderr); err != nil { + if exitErr, ok := err.(*exec.ExitError); ok && exitErr.ExitCode() == 127 { + return fmt.Errorf("%s CLI not found in sandbox %q; was it created with the right preset?", agent.Binary, name) + } + return fmt.Errorf("agent-send failed for sandbox %q: %w", name, err) + } + return nil +} + +func runRemoteAgentSendHTTP(client *apiclient.Client, name, message string, agent agentConfig, opts agentRunOpts, stdout io.Writer) error { req := apiclient.AgentSendRequest{ Message: message, NewSession: opts.NewSession, @@ -244,7 +261,7 @@ Use --no-wait to send the message and return immediately.`, newSession, _ := cmd.Flags().GetBool("new-session") opts := agentRunOpts{SessionID: sessionID, NewSession: newSession} - if err := runRemoteAgentSend(client, name, message, noWait, workdir, agent, opts, os.Stdout); err != nil { + if err := runRemoteAgentSend(client, name, message, noWait, workdir, agent, opts, os.Stdout, os.Stderr); err != nil { return err } if noWait { diff --git a/go/cmd/amika/sandbox/sandbox_agent_test.go b/go/cmd/amika/sandbox/sandbox_agent_test.go index 7bcd34b1..dd1104b1 100644 --- a/go/cmd/amika/sandbox/sandbox_agent_test.go +++ b/go/cmd/amika/sandbox/sandbox_agent_test.go @@ -187,15 +187,22 @@ func TestAgentCmdPartsWithOpts(t *testing.T) { func TestBuildRemoteAgentShellCmd(t *testing.T) { claude := knownAgents["claude"] + t.Run("wait mode streaming omits json output", func(t *testing.T) { + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{}, false) + if strings.Contains(got, "--output-format") { + t.Fatalf("cmd = %q, should not contain --output-format when streaming", got) + } + }) + t.Run("wait mode includes json output", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{}) + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{}, true) if !strings.Contains(got, "--output-format json") { t.Fatalf("cmd = %q, want --output-format json", got) } }) t.Run("no-wait mode has no json and wraps in tmux", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", true, "/home/amika", claude, agentRunOpts{}) + got := buildRemoteAgentShellCmd("hello", true, "/home/amika", claude, agentRunOpts{}, false) if strings.Contains(got, "--output-format") { t.Fatalf("cmd = %q, should not contain --output-format in no-wait mode", got) } @@ -205,14 +212,14 @@ func TestBuildRemoteAgentShellCmd(t *testing.T) { }) t.Run("session id maps to --resume", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{SessionID: "sess-42"}) + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{SessionID: "sess-42"}, true) if !strings.Contains(got, "--resume sess-42") { t.Fatalf("cmd = %q, want --resume sess-42", got) } }) t.Run("new session passes no session flag to claude", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{NewSession: true}) + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", claude, agentRunOpts{NewSession: true}, true) if strings.Contains(got, "--new-session") { t.Fatalf("cmd = %q, should not contain --new-session", got) } @@ -227,7 +234,7 @@ func TestBuildRemoteAgentShellCmd(t *testing.T) { codex := knownAgents["codex"] t.Run("codex wait mode includes --json", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", false, "/home/amika", codex, agentRunOpts{}) + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", codex, agentRunOpts{}, true) if !strings.Contains(got, "--json") { t.Fatalf("cmd = %q, want --json", got) } @@ -237,7 +244,7 @@ func TestBuildRemoteAgentShellCmd(t *testing.T) { }) t.Run("codex no-wait wraps in tmux without json", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", true, "/home/amika", codex, agentRunOpts{}) + got := buildRemoteAgentShellCmd("hello", true, "/home/amika", codex, agentRunOpts{}, false) if strings.Contains(got, "--json") { t.Fatalf("cmd = %q, should not contain --json in no-wait mode", got) } @@ -247,7 +254,7 @@ func TestBuildRemoteAgentShellCmd(t *testing.T) { }) t.Run("codex session id uses resume subcommand", func(t *testing.T) { - got := buildRemoteAgentShellCmd("hello", false, "/home/amika", codex, agentRunOpts{SessionID: "sess-42"}) + got := buildRemoteAgentShellCmd("hello", false, "/home/amika", codex, agentRunOpts{SessionID: "sess-42"}, true) if !strings.Contains(got, "codex exec resume") { t.Fatalf("cmd = %q, want 'codex exec resume'", got) } diff --git a/go/internal/materialize/materialize.go b/go/internal/materialize/materialize.go index d38b6e62..e8d5d25c 100644 --- a/go/internal/materialize/materialize.go +++ b/go/internal/materialize/materialize.go @@ -2,7 +2,6 @@ package materialize import ( - "bytes" "fmt" "os" "os/exec" @@ -69,9 +68,7 @@ func Run(opts Options) error { cmd.Env = append(os.Environ(), opts.Env...) } cmd.Stdout = os.Stdout - - var stderrBuf bytes.Buffer - cmd.Stderr = &stderrBuf + cmd.Stderr = os.Stderr // Print header before execution if hasScript { @@ -86,25 +83,10 @@ func Run(opts Options) error { if hasCmd { label = "Command" } - captured := strings.TrimRight(stderrBuf.String(), "\n") - if captured != "" { - lines := strings.Split(captured, "\n") - quoted := make([]string, len(lines)) - for i, line := range lines { - quoted[i] = "> " + line - } - fmt.Fprintf(os.Stderr, "%s failed to run:\n\n%s\n\n", label, strings.Join(quoted, "\n")) - } else { - fmt.Fprintf(os.Stderr, "%s failed to run.\n", label) - } + fmt.Fprintf(os.Stderr, "%s failed to run.\n", label) return fmt.Errorf("execution failed: %w", err) } - // On success, write captured stderr through so it's still visible - if stderrBuf.Len() > 0 { - stderrBuf.WriteTo(os.Stderr) - } - // Copy outdir contents to destdir using rsync rsyncCmd := exec.Command("rsync", "-a", opts.Outdir+"/", opts.Destdir+"/") rsyncCmd.Stdout = os.Stdout diff --git a/go/internal/ssh/ssh.go b/go/internal/ssh/ssh.go index 5d1b0f87..09cbfbf5 100644 --- a/go/internal/ssh/ssh.go +++ b/go/internal/ssh/ssh.go @@ -5,6 +5,7 @@ package ssh import ( "fmt" + "io" "os" "os/exec" "strings" @@ -42,3 +43,32 @@ func ExecSSH(client *apiclient.Client, name string, forcePTY bool, extraArgs []s } return syscall.Exec(sshBin, append([]string{"ssh"}, sshArgs...), os.Environ()) } + +// RunSSH runs a remote command over SSH, streaming stdin/stdout/stderr to the +// provided writers. Unlike ExecSSH, it does not replace the current process, +// so callers can observe output as it arrives and inspect the exit status. +func RunSSH(client *apiclient.Client, name string, extraArgs []string, stdin io.Reader, stdout, stderr io.Writer) error { + info, err := client.GetSSH(name) + if err != nil { + return err + } + if info.SSHDestination == "" { + return fmt.Errorf("server returned empty SSH destination") + } + + sshArgs := strings.Fields(info.SSHDestination) + if len(extraArgs) > 0 { + sshArgs = append(sshArgs, extraArgs...) + } + + sshBin, err := exec.LookPath("ssh") + if err != nil { + return fmt.Errorf("ssh not found: %w", err) + } + + cmd := exec.Command(sshBin, append([]string{"ssh"}, sshArgs...)...) + cmd.Stdin = stdin + cmd.Stdout = stdout + cmd.Stderr = stderr + return cmd.Run() +} diff --git a/go/internal/ssh/ssh_test.go b/go/internal/ssh/ssh_test.go new file mode 100644 index 00000000..3447d623 --- /dev/null +++ b/go/internal/ssh/ssh_test.go @@ -0,0 +1,43 @@ +package ssh + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/gofixpoint/amika/go/internal/apiclient" +) + +func TestRunSSHStreamsOutput(t *testing.T) { + binDir := t.TempDir() + sshPath := filepath.Join(binDir, "ssh") + script := "#!/bin/sh\nfor arg in \"$@\"; do case \"$arg\" in echo*) eval \"$arg\" ;; esac; done\n" + if err := os.WriteFile(sshPath, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", binDir) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/ssh") { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ssh_destination":"example.com","token":"tok","expires_at":"2099-01-01T00:00:00Z"}`)) + })) + t.Cleanup(server.Close) + + client := apiclient.NewClient(server.URL, "test-token") + var stdout bytes.Buffer + if err := RunSSH(client, "sb-1", []string{"echo hello"}, nil, &stdout, io.Discard); err != nil { + t.Fatalf("RunSSH() error: %v", err) + } + if got := strings.TrimSpace(stdout.String()); got != "hello" { + t.Fatalf("stdout = %q, want %q", got, "hello") + } +}