fix(agent): avoid oversized env spawn failures (#1344)
This commit is contained in:
@@ -43,7 +43,6 @@ WORKDIR /workspace
|
||||
# Environment variables (passed at runtime)
|
||||
# TASK_ID - The task identifier
|
||||
# TASK_TITLE - Task title/name
|
||||
# TASK_DESCRIPTION - Full task description
|
||||
#
|
||||
# For Auggie:
|
||||
# AUGMENT_SESSION_AUTH - Session auth JSON for Augment (from ~/.augment/session.json)
|
||||
@@ -62,4 +61,3 @@ EXPOSE 39429
|
||||
|
||||
# Run agentctl HTTP server
|
||||
ENTRYPOINT ["/agent/entrypoint.sh"]
|
||||
|
||||
|
||||
@@ -661,9 +661,7 @@ func (m *Manager) RestartAgentProcess(ctx context.Context, executionID string) e
|
||||
|
||||
// 5. Reconfigure and start new agent subprocess
|
||||
approvalPolicy, _ := m.resolveApprovalPolicyAndDisplayName(ctx, execution)
|
||||
taskDescription := getTaskDescriptionFromMetadata(execution)
|
||||
|
||||
if _, err := m.configureAndStartAgent(ctx, execution, taskDescription, approvalPolicy); err != nil {
|
||||
if _, err := m.configureAndStartAgent(ctx, execution, approvalPolicy); err != nil {
|
||||
m.updateExecutionError(executionID, "failed to restart agent: "+err.Error())
|
||||
return fmt.Errorf("failed to restart agent: %w", err)
|
||||
}
|
||||
|
||||
@@ -485,7 +485,10 @@ func (m *Manager) launchBuildExecutorRequest(ctx context.Context, executionID st
|
||||
return nil, nil, nil, fmt.Errorf("no runtime configured: %w", err)
|
||||
}
|
||||
|
||||
env := m.buildEnvForExecution(ctx, executionID, reqWithWorktree, agentConfig, profileInfo)
|
||||
env, err := m.buildEnvForExecution(ctx, executionID, reqWithWorktree, agentConfig, profileInfo)
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("build launch environment: %w", err)
|
||||
}
|
||||
|
||||
acpMcpServers, err := m.resolveMcpServersWithParams(ctx, reqWithWorktree.AgentProfileID, reqWithWorktree.Metadata, agentConfig)
|
||||
if err != nil {
|
||||
@@ -1120,12 +1123,13 @@ func getAttachmentsFromMetadata(execution *AgentExecution) []MessageAttachment {
|
||||
|
||||
// configureAndStartAgent configures the agent command and starts the agent subprocess.
|
||||
// Returns the effective boot command (full command with adapter args, or base command).
|
||||
func (m *Manager) configureAndStartAgent(ctx context.Context, execution *AgentExecution, taskDescription, approvalPolicy string) (string, error) {
|
||||
func (m *Manager) configureAndStartAgent(ctx context.Context, execution *AgentExecution, approvalPolicy string) (string, error) {
|
||||
env := runtimeEnvFromMetadata(execution.Metadata)
|
||||
if taskDescription != "" {
|
||||
env["TASK_DESCRIPTION"] = taskDescription
|
||||
}
|
||||
m.mergeAgentProfileEnvForExecution(ctx, execution, env)
|
||||
if err := spillLargeWakePayloadEnv(env, execution.WorkspacePath, m.logger.Zap()); err != nil {
|
||||
m.updateExecutionError(execution.ID, "failed to prepare agent env: "+err.Error())
|
||||
return "", fmt.Errorf("failed to prepare agent env: %w", err)
|
||||
}
|
||||
|
||||
if err := execution.agentctl.ConfigureAgent(ctx, execution.AgentCommand, env, approvalPolicy, execution.ContinueCommand); err != nil {
|
||||
return "", fmt.Errorf("failed to configure agent: %w", err)
|
||||
|
||||
@@ -2,7 +2,14 @@ package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -148,18 +155,161 @@ func TestBuildEnvForExecution_ResolvesSecretBackedProfileEnv(t *testing.T) {
|
||||
EnvVars: []settingsmodels.ProfileEnvVar{{Key: "FROM_SECRET", SecretID: "sec-1"}},
|
||||
}
|
||||
|
||||
env := mgr.buildEnvForExecution(
|
||||
env, err := mgr.buildEnvForExecution(
|
||||
context.Background(),
|
||||
"exec-1",
|
||||
&LaunchRequest{AgentProfileID: "profile-1"},
|
||||
nil,
|
||||
profileInfo,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("buildEnvForExecution: %v", err)
|
||||
}
|
||||
if env["FROM_SECRET"] != "revealed" {
|
||||
t.Fatalf("FROM_SECRET: got %q want revealed", env["FROM_SECRET"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildEnvForExecution_DoesNotCopyTaskDescriptionToEnv(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
env, err := mgr.buildEnvForExecution(
|
||||
context.Background(),
|
||||
"exec-1",
|
||||
&LaunchRequest{
|
||||
TaskID: "task-1",
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
TaskDescription: strings.Repeat("large prompt\n", 1000),
|
||||
},
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("buildEnvForExecution: %v", err)
|
||||
}
|
||||
|
||||
if _, exists := env["TASK_DESCRIPTION"]; exists {
|
||||
t.Fatalf("TASK_DESCRIPTION must not be copied into subprocess env")
|
||||
}
|
||||
if env["KANDEV_TASK_ID"] != "task-1" {
|
||||
t.Fatalf("KANDEV_TASK_ID = %q, want task-1", env["KANDEV_TASK_ID"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildEnvForExecution_SpillsLargeWakePayloadToWorkspaceFile(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
workspace := t.TempDir()
|
||||
payload := strings.Repeat("x", envWakePayloadInlineMax+1)
|
||||
|
||||
env, err := mgr.buildEnvForExecution(
|
||||
context.Background(),
|
||||
"exec-1",
|
||||
&LaunchRequest{
|
||||
TaskID: "task-1",
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
WorkspacePath: workspace,
|
||||
Env: map[string]string{
|
||||
"KANDEV_RUN_ID": "run-1",
|
||||
"KANDEV_WAKE_PAYLOAD_JSON": payload,
|
||||
},
|
||||
},
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("buildEnvForExecution: %v", err)
|
||||
}
|
||||
if _, exists := env["KANDEV_WAKE_PAYLOAD_JSON"]; exists {
|
||||
t.Fatalf("large wake payload must not remain inline in env")
|
||||
}
|
||||
relPath := env["KANDEV_WAKE_PAYLOAD_PATH"]
|
||||
if relPath == "" {
|
||||
t.Fatalf("KANDEV_WAKE_PAYLOAD_PATH missing from env: %+v", env)
|
||||
}
|
||||
got, err := os.ReadFile(filepath.Join(workspace, filepath.FromSlash(relPath)))
|
||||
if err != nil {
|
||||
t.Fatalf("read spilled payload: %v", err)
|
||||
}
|
||||
if string(got) != payload {
|
||||
t.Fatalf("spilled payload mismatch")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigureAndStartAgent_DoesNotSendTaskDescriptionEnv(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
var configuredEnv map[string]string
|
||||
client := newConfigureCaptureAgentctlClient(t, newTestLogger(), &configuredEnv)
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-1",
|
||||
TaskID: "task-1",
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
AgentCommand: "npx -y @zed-industries/codex-acp",
|
||||
WorkspacePath: t.TempDir(),
|
||||
Metadata: map[string]interface{}{
|
||||
"runtime_env": map[string]string{"KEEP_ME": "yes"},
|
||||
"task_description": strings.Repeat("large prompt\n", 1000),
|
||||
},
|
||||
agentctl: client,
|
||||
}
|
||||
|
||||
bootCommand, err := mgr.configureAndStartAgent(context.Background(), execution, "never")
|
||||
if err != nil {
|
||||
t.Fatalf("configureAndStartAgent() error = %v", err)
|
||||
}
|
||||
if bootCommand != "npx -y @zed-industries/codex-acp" {
|
||||
t.Fatalf("bootCommand = %q, want agent command", bootCommand)
|
||||
}
|
||||
if configuredEnv["KEEP_ME"] != "yes" {
|
||||
t.Fatalf("runtime_env was not preserved: %+v", configuredEnv)
|
||||
}
|
||||
if _, exists := configuredEnv["TASK_DESCRIPTION"]; exists {
|
||||
t.Fatalf("TASK_DESCRIPTION must not be sent to agentctl configure env")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigureAndStartAgent_SpillsLargeWakePayloadEnv(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
var configuredEnv map[string]string
|
||||
workspace := t.TempDir()
|
||||
client := newConfigureCaptureAgentctlClient(t, newTestLogger(), &configuredEnv)
|
||||
payload := strings.Repeat("x", envWakePayloadInlineMax+1)
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-1",
|
||||
TaskID: "task-1",
|
||||
SessionID: "session-1",
|
||||
AgentProfileID: "profile-1",
|
||||
AgentCommand: "npx -y @zed-industries/codex-acp",
|
||||
WorkspacePath: workspace,
|
||||
Metadata: map[string]interface{}{
|
||||
"runtime_env": map[string]string{
|
||||
"KANDEV_RUN_ID": "run-2",
|
||||
"KANDEV_WAKE_PAYLOAD_JSON": payload,
|
||||
},
|
||||
},
|
||||
agentctl: client,
|
||||
}
|
||||
|
||||
if _, err := mgr.configureAndStartAgent(context.Background(), execution, "never"); err != nil {
|
||||
t.Fatalf("configureAndStartAgent() error = %v", err)
|
||||
}
|
||||
if _, exists := configuredEnv["KANDEV_WAKE_PAYLOAD_JSON"]; exists {
|
||||
t.Fatalf("large wake payload must not be sent inline to agentctl configure env")
|
||||
}
|
||||
relPath := configuredEnv["KANDEV_WAKE_PAYLOAD_PATH"]
|
||||
if relPath == "" {
|
||||
t.Fatalf("KANDEV_WAKE_PAYLOAD_PATH missing from env: %+v", configuredEnv)
|
||||
}
|
||||
got, err := os.ReadFile(filepath.Join(workspace, filepath.FromSlash(relPath)))
|
||||
if err != nil {
|
||||
t.Fatalf("read spilled payload: %v", err)
|
||||
}
|
||||
if string(got) != payload {
|
||||
t.Fatalf("spilled payload mismatch")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetExecutionEnv_DoesNotSnapshotProfileEnvVars(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
mgr.profileResolver = &mockPassthroughProfileResolver{
|
||||
@@ -191,6 +341,44 @@ func TestSetExecutionEnv_DoesNotSnapshotProfileEnvVars(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func newConfigureCaptureAgentctlClient(t *testing.T, log *logger.Logger, captured *map[string]string) *agentctl.Client {
|
||||
t.Helper()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api/v1/agent/configure":
|
||||
var req struct {
|
||||
Env map[string]string `json:"env"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
t.Errorf("decode configure request: %v", err)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
*captured = req.Env
|
||||
_, _ = w.Write([]byte(`{"success":true}`))
|
||||
case "/api/v1/start":
|
||||
_, _ = w.Write([]byte(`{"success":true,"command":"npx -y @zed-industries/codex-acp"}`))
|
||||
default:
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
parsed, err := url.Parse(server.URL)
|
||||
if err != nil {
|
||||
t.Fatalf("parse test server URL: %v", err)
|
||||
}
|
||||
host, portString, err := net.SplitHostPort(parsed.Host)
|
||||
if err != nil {
|
||||
t.Fatalf("split test server host: %v", err)
|
||||
}
|
||||
port, err := strconv.Atoi(portString)
|
||||
if err != nil {
|
||||
t.Fatalf("parse test server port: %v", err)
|
||||
}
|
||||
return agentctl.NewClient(host, port, log)
|
||||
}
|
||||
|
||||
// trackingPreparer records whether Prepare was called.
|
||||
type trackingPreparer struct {
|
||||
called bool
|
||||
|
||||
@@ -135,7 +135,7 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
|
||||
zap.String("acp_session_id", execution.ACPSessionID))
|
||||
|
||||
var err error
|
||||
bootCommand, err = m.configureAndStartAgent(ctx, execution, taskDescription, approvalPolicy)
|
||||
bootCommand, err = m.configureAndStartAgent(ctx, execution, approvalPolicy)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -224,7 +224,7 @@ func (m *Manager) finalizeBootMessage(execution *AgentExecution, msg *models.Mes
|
||||
|
||||
// buildEnvForExecution builds environment variables for any runtime.
|
||||
// This is the unified method used by the runtime interface.
|
||||
func (m *Manager) buildEnvForExecution(ctx context.Context, executionID string, req *LaunchRequest, agentConfig agents.Agent, profileInfo *AgentProfileInfo) map[string]string {
|
||||
func (m *Manager) buildEnvForExecution(ctx context.Context, executionID string, req *LaunchRequest, agentConfig agents.Agent, profileInfo *AgentProfileInfo) (map[string]string, error) {
|
||||
env := make(map[string]string)
|
||||
|
||||
// Copy request environment
|
||||
@@ -243,7 +243,6 @@ func (m *Manager) buildEnvForExecution(ctx context.Context, executionID string,
|
||||
env["KANDEV_TASK_ID"] = req.TaskID
|
||||
env["KANDEV_SESSION_ID"] = req.SessionID
|
||||
env["KANDEV_AGENT_PROFILE_ID"] = req.AgentProfileID
|
||||
env["TASK_DESCRIPTION"] = req.TaskDescription
|
||||
|
||||
// Add agent runtime default env vars (e.g., MCP_TIMEOUT for Claude Code)
|
||||
if agentConfig != nil {
|
||||
@@ -265,7 +264,11 @@ func (m *Manager) buildEnvForExecution(ctx context.Context, executionID string,
|
||||
}
|
||||
}
|
||||
|
||||
return env
|
||||
if err := spillLargeWakePayloadEnv(env, req.WorkspacePath, m.logger.Zap()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return env, nil
|
||||
}
|
||||
|
||||
// waitForAgentctlReady waits for the agentctl HTTP server to be ready.
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
"unicode"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const (
|
||||
envWakePayloadJSON = "KANDEV_WAKE_PAYLOAD_JSON"
|
||||
envWakePayloadPath = "KANDEV_WAKE_PAYLOAD_PATH"
|
||||
envWakePayloadInlineMax = 64 * 1024
|
||||
wakePayloadDirRel = ".kandev/wake-payloads"
|
||||
wakePayloadExcludeLine = ".kandev/wake-payloads/"
|
||||
defaultWakePayloadFileID = "payload"
|
||||
)
|
||||
|
||||
func spillLargeWakePayloadEnv(env map[string]string, workspacePath string, log *zap.Logger) error {
|
||||
payload := env[envWakePayloadJSON]
|
||||
if payload == "" || len(payload) <= envWakePayloadInlineMax {
|
||||
return nil
|
||||
}
|
||||
fileID := sanitizeWakePayloadFileID(env["KANDEV_RUN_ID"])
|
||||
if fileID == defaultWakePayloadFileID && log != nil {
|
||||
log.Warn("KANDEV_RUN_ID is missing or empty; spill file may collide across runs",
|
||||
zap.String("payload_path", filepath.Join(wakePayloadDirRel, fileID+".json")))
|
||||
}
|
||||
if workspacePath == "" {
|
||||
return fmt.Errorf("%s is %d bytes, above %d byte inline limit, but workspace path is empty",
|
||||
envWakePayloadJSON, len(payload), envWakePayloadInlineMax)
|
||||
}
|
||||
|
||||
relPath := filepath.ToSlash(filepath.Join(wakePayloadDirRel, fileID+".json"))
|
||||
absPath := filepath.Join(workspacePath, filepath.FromSlash(relPath))
|
||||
if err := os.MkdirAll(filepath.Dir(absPath), 0o700); err != nil {
|
||||
return fmt.Errorf("create wake payload directory: %w", err)
|
||||
}
|
||||
if err := os.WriteFile(absPath, []byte(payload), 0o600); err != nil {
|
||||
_ = os.Remove(absPath)
|
||||
return fmt.Errorf("write wake payload file: %w", err)
|
||||
}
|
||||
|
||||
delete(env, envWakePayloadJSON)
|
||||
env[envWakePayloadPath] = relPath
|
||||
|
||||
if err := ensureWakePayloadGitExclude(workspacePath); err != nil && log != nil {
|
||||
log.Warn("failed to update git exclude for wake payload spill file",
|
||||
zap.String("workspace_path", workspacePath),
|
||||
zap.Error(err))
|
||||
}
|
||||
if log != nil {
|
||||
log.Info("spilled oversized wake payload env to workspace file",
|
||||
zap.Int("payload_bytes", len(payload)),
|
||||
zap.String("payload_path", relPath))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func sanitizeWakePayloadFileID(id string) string {
|
||||
id = strings.TrimSpace(id)
|
||||
if id == "" {
|
||||
return defaultWakePayloadFileID
|
||||
}
|
||||
var b strings.Builder
|
||||
for _, r := range id {
|
||||
if unicode.IsLetter(r) || unicode.IsDigit(r) || r == '-' || r == '_' {
|
||||
b.WriteRune(r)
|
||||
}
|
||||
}
|
||||
if b.Len() == 0 {
|
||||
return defaultWakePayloadFileID
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func ensureWakePayloadGitExclude(workspacePath string) error {
|
||||
infoDir, err := gitInfoDir(workspacePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if infoDir == "" {
|
||||
return nil
|
||||
}
|
||||
if err := os.MkdirAll(infoDir, 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
excludePath := filepath.Join(infoDir, "exclude")
|
||||
data, err := os.ReadFile(excludePath)
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return err
|
||||
}
|
||||
if strings.Contains(string(data), wakePayloadExcludeLine) {
|
||||
return nil
|
||||
}
|
||||
prefix := ""
|
||||
if len(data) > 0 && !strings.HasSuffix(string(data), "\n") {
|
||||
prefix = "\n"
|
||||
}
|
||||
return os.WriteFile(excludePath, append(data, []byte(prefix+wakePayloadExcludeLine+"\n")...), 0o600)
|
||||
}
|
||||
|
||||
func gitInfoDir(workspacePath string) (string, error) {
|
||||
gitPath := filepath.Join(workspacePath, ".git")
|
||||
if infoDir, err := gitDirInfo(gitPath); err != nil || infoDir != "" {
|
||||
return infoDir, err
|
||||
}
|
||||
return gitWorktreeInfo(workspacePath, gitPath)
|
||||
}
|
||||
|
||||
func gitDirInfo(gitPath string) (string, error) {
|
||||
infoPath := filepath.Join(gitPath, "info")
|
||||
st, err := os.Stat(infoPath)
|
||||
if err == nil {
|
||||
if st.IsDir() {
|
||||
return infoPath, nil
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
if os.IsNotExist(err) || errors.Is(err, syscall.ENOTDIR) {
|
||||
if fi, statErr := os.Stat(gitPath); statErr == nil && fi.IsDir() {
|
||||
return infoPath, nil
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
|
||||
func gitWorktreeInfo(workspacePath string, gitPath string) (string, error) {
|
||||
data, err := os.ReadFile(gitPath)
|
||||
if os.IsNotExist(err) {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
const prefix = "gitdir:"
|
||||
line := strings.TrimSpace(string(data))
|
||||
if !strings.HasPrefix(line, prefix) {
|
||||
return "", nil
|
||||
}
|
||||
gitDir := strings.TrimSpace(strings.TrimPrefix(line, prefix))
|
||||
if gitDir == "" {
|
||||
return "", nil
|
||||
}
|
||||
if !filepath.IsAbs(gitDir) {
|
||||
gitDir = filepath.Join(workspacePath, gitDir)
|
||||
}
|
||||
infoDir := filepath.Join(filepath.Clean(gitDir), "info")
|
||||
st, err := os.Stat(infoDir)
|
||||
switch {
|
||||
case os.IsNotExist(err):
|
||||
return infoDir, nil
|
||||
case err != nil:
|
||||
return "", err
|
||||
case !st.IsDir():
|
||||
return "", fmt.Errorf("git info path is not a directory: %s", infoDir)
|
||||
}
|
||||
return infoDir, nil
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSpillLargeWakePayloadEnv_UsesFallbackFileIDWhenRunIDMissing(t *testing.T) {
|
||||
workspace := t.TempDir()
|
||||
env := map[string]string{
|
||||
envWakePayloadJSON: strings.Repeat("x", envWakePayloadInlineMax+1),
|
||||
}
|
||||
if err := spillLargeWakePayloadEnv(env, workspace, nil); err != nil {
|
||||
t.Fatalf("spillLargeWakePayloadEnv: %v", err)
|
||||
}
|
||||
if _, exists := env[envWakePayloadJSON]; exists {
|
||||
t.Fatalf("expected %s removed", envWakePayloadJSON)
|
||||
}
|
||||
relPath, exists := env[envWakePayloadPath]
|
||||
if !exists {
|
||||
t.Fatalf("expected %s set", envWakePayloadPath)
|
||||
}
|
||||
if got, want := filepath.ToSlash(relPath), filepath.ToSlash(filepath.Join(wakePayloadDirRel, defaultWakePayloadFileID+".json")); got != want {
|
||||
t.Fatalf("payload path = %q, want %q", got, want)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(workspace, filepath.FromSlash(relPath))); err != nil {
|
||||
t.Fatalf("payload file missing: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnsureWakePayloadGitExclude_WorktreeGitFile(t *testing.T) {
|
||||
workspace := t.TempDir()
|
||||
gitDir := filepath.Join(t.TempDir(), "worktrees", "task", ".git")
|
||||
if err := os.MkdirAll(filepath.Join(gitDir, "info"), 0o700); err != nil {
|
||||
t.Fatalf("create git info dir: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(workspace, ".git"), []byte("gitdir: "+gitDir+"\n"), 0o600); err != nil {
|
||||
t.Fatalf("write .git file: %v", err)
|
||||
}
|
||||
|
||||
if err := ensureWakePayloadGitExclude(workspace); err != nil {
|
||||
t.Fatalf("ensureWakePayloadGitExclude: %v", err)
|
||||
}
|
||||
|
||||
data, err := os.ReadFile(filepath.Join(gitDir, "info", "exclude"))
|
||||
if err != nil {
|
||||
t.Fatalf("read exclude: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(data), wakePayloadExcludeLine) {
|
||||
t.Fatalf("exclude missing %q: %s", wakePayloadExcludeLine, string(data))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnsureWakePayloadGitExclude_WorktreeInfoPathIsFile(t *testing.T) {
|
||||
workspace := t.TempDir()
|
||||
gitDir := filepath.Join(t.TempDir(), "worktrees", "task", ".git")
|
||||
if err := os.MkdirAll(gitDir, 0o700); err != nil {
|
||||
t.Fatalf("create git dir: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(gitDir, "info"), []byte("not a directory"), 0o600); err != nil {
|
||||
t.Fatalf("write git info file: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(workspace, ".git"), []byte("gitdir: "+gitDir+"\n"), 0o600); err != nil {
|
||||
t.Fatalf("write .git file: %v", err)
|
||||
}
|
||||
|
||||
err := ensureWakePayloadGitExclude(workspace)
|
||||
if err == nil {
|
||||
t.Fatalf("expected error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "git info path is not a directory") {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnsureWakePayloadGitExclude_NormalGitDirWithoutInfo(t *testing.T) {
|
||||
workspace := t.TempDir()
|
||||
gitDir := filepath.Join(workspace, ".git")
|
||||
if err := os.Mkdir(gitDir, 0o700); err != nil {
|
||||
t.Fatalf("create .git dir: %v", err)
|
||||
}
|
||||
|
||||
if err := ensureWakePayloadGitExclude(workspace); err != nil {
|
||||
t.Fatalf("ensureWakePayloadGitExclude: %v", err)
|
||||
}
|
||||
|
||||
excludePath := filepath.Join(gitDir, "info", "exclude")
|
||||
data, err := os.ReadFile(excludePath)
|
||||
if err != nil {
|
||||
t.Fatalf("read exclude: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(data), wakePayloadExcludeLine) {
|
||||
t.Fatalf("exclude missing %q: %s", wakePayloadExcludeLine, string(data))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSanitizeWakePayloadFileID(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in string
|
||||
want string
|
||||
}{
|
||||
{"empty", "", defaultWakePayloadFileID},
|
||||
{"spaces", " ", defaultWakePayloadFileID},
|
||||
{"allowed", "run_01-id", "run_01-id"},
|
||||
{"specials", "a*b c#1", "abc1"},
|
||||
{"only-specials", " /\\*?? ", defaultWakePayloadFileID},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
got := sanitizeWakePayloadFileID(tc.in)
|
||||
if got != tc.want {
|
||||
t.Fatalf("sanitizeWakePayloadFileID(%q) = %q, want %q", tc.in, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpillLargeWakePayloadEnv_NormalizesLargePayload(t *testing.T) {
|
||||
workspace := t.TempDir()
|
||||
env := map[string]string{
|
||||
envWakePayloadJSON: strings.Repeat("x", envWakePayloadInlineMax-1),
|
||||
"KANDEV_RUN_ID": "keep-inline",
|
||||
}
|
||||
if err := spillLargeWakePayloadEnv(env, workspace, nil); err != nil {
|
||||
t.Fatalf("spillLargeWakePayloadEnv: %v", err)
|
||||
}
|
||||
if _, exists := env[envWakePayloadPath]; exists {
|
||||
t.Fatalf("did not expect %s set for under-limit payload", envWakePayloadPath)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(workspace, wakePayloadDirRel)); err == nil {
|
||||
t.Fatalf("expected %s not created for under-limit payload", wakePayloadDirRel)
|
||||
} else if !os.IsNotExist(err) {
|
||||
t.Fatalf("stat wake payload dir: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpillLargeWakePayloadEnv_UsesWorkspaceErrorForMissingPath(t *testing.T) {
|
||||
workspace := ""
|
||||
env := map[string]string{
|
||||
envWakePayloadJSON: strings.Repeat("x", envWakePayloadInlineMax+1),
|
||||
"KANDEV_RUN_ID": "x",
|
||||
}
|
||||
err := spillLargeWakePayloadEnv(env, workspace, nil)
|
||||
if err == nil {
|
||||
t.Fatalf("expected error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "workspace path is empty") {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -4,15 +4,18 @@ package process
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter"
|
||||
@@ -728,7 +731,7 @@ func (m *Manager) Start(ctx context.Context) error {
|
||||
// Start the subprocess now that pipes are connected
|
||||
if err := m.cmd.Start(); err != nil {
|
||||
m.status.Store(StatusError)
|
||||
return fmt.Errorf("failed to start agent: %w", err)
|
||||
return formatAgentStartError(err, m.cfg.AgentEnv)
|
||||
}
|
||||
|
||||
m.stopCh = make(chan struct{})
|
||||
@@ -881,15 +884,76 @@ func (m *Manager) buildFinalCommand() error {
|
||||
// (npx -> sh -> node -> opencode binary).
|
||||
setProcGroup(m.cmd)
|
||||
|
||||
envBytes, largestEnv := summarizeEnvBytes(m.cfg.AgentEnv, 3)
|
||||
m.logger.Info("agent command prepared",
|
||||
zap.Strings("args", m.cfg.AgentArgs),
|
||||
zap.Strings("extra_args", extraArgs),
|
||||
zap.String("workdir", m.cfg.WorkDir),
|
||||
zap.Int("env_count", len(m.cfg.AgentEnv)))
|
||||
zap.Int("env_count", len(m.cfg.AgentEnv)),
|
||||
zap.Int("env_bytes", envBytes),
|
||||
zap.String("largest_env_keys", formatEnvEntrySizes(largestEnv)))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type envEntrySize struct {
|
||||
key string
|
||||
bytes int
|
||||
}
|
||||
|
||||
func formatAgentStartError(err error, env []string) error {
|
||||
if isArgumentListTooLong(err) {
|
||||
total, largest := summarizeEnvBytes(env, 3)
|
||||
return fmt.Errorf(
|
||||
"failed to start agent: environment/arguments too large; env_bytes=%d largest_env_keys=%s: %w",
|
||||
total, formatEnvEntrySizes(largest), err,
|
||||
)
|
||||
}
|
||||
return fmt.Errorf("failed to start agent: %w", err)
|
||||
}
|
||||
|
||||
func isArgumentListTooLong(err error) bool {
|
||||
if errors.Is(err, syscall.E2BIG) {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(strings.ToLower(err.Error()), "argument list too long")
|
||||
}
|
||||
|
||||
func summarizeEnvBytes(env []string, limit int) (int, []envEntrySize) {
|
||||
entries := make([]envEntrySize, 0, len(env))
|
||||
total := 0
|
||||
for _, item := range env {
|
||||
size := len(item) + 1
|
||||
total += size
|
||||
key := item
|
||||
if idx := strings.IndexByte(item, '='); idx >= 0 {
|
||||
key = item[:idx]
|
||||
}
|
||||
entries = append(entries, envEntrySize{key: key, bytes: size})
|
||||
}
|
||||
sort.Slice(entries, func(i, j int) bool {
|
||||
if entries[i].bytes == entries[j].bytes {
|
||||
return entries[i].key < entries[j].key
|
||||
}
|
||||
return entries[i].bytes > entries[j].bytes
|
||||
})
|
||||
if limit > 0 && len(entries) > limit {
|
||||
entries = entries[:limit]
|
||||
}
|
||||
return total, entries
|
||||
}
|
||||
|
||||
func formatEnvEntrySizes(entries []envEntrySize) string {
|
||||
if len(entries) == 0 {
|
||||
return "[]"
|
||||
}
|
||||
parts := make([]string, 0, len(entries))
|
||||
for _, entry := range entries {
|
||||
parts = append(parts, fmt.Sprintf("%s:%d", entry.key, entry.bytes))
|
||||
}
|
||||
return "[" + strings.Join(parts, ",") + "]"
|
||||
}
|
||||
|
||||
func (m *Manager) ensureAgentTempEnv() error {
|
||||
dir := filepath.Join(os.TempDir(), agentTempDirRoot, agentTempDirName(m.cfg.SessionID, m.cfg.Port))
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
|
||||
@@ -2,8 +2,12 @@ package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"syscall"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter"
|
||||
@@ -141,3 +145,25 @@ func TestManager_Start_PipesCreatedBeforeProcessStart(t *testing.T) {
|
||||
_ = m.stdin.Close()
|
||||
_ = m.cmd.Wait()
|
||||
}
|
||||
|
||||
func TestFormatAgentStartError_E2BIGIncludesEnvDiagnostics(t *testing.T) {
|
||||
err := formatAgentStartError(&os.PathError{Op: "fork/exec", Path: "npx", Err: syscall.E2BIG}, []string{
|
||||
"SMALL=1",
|
||||
"KANDEV_WAKE_PAYLOAD_JSON=" + strings.Repeat("x", 100),
|
||||
"PATH=/usr/bin",
|
||||
})
|
||||
|
||||
msg := err.Error()
|
||||
if !strings.Contains(msg, "environment/arguments too large") {
|
||||
t.Fatalf("missing actionable E2BIG message: %s", msg)
|
||||
}
|
||||
if !strings.Contains(msg, "env_bytes=") {
|
||||
t.Fatalf("missing env byte diagnostics: %s", msg)
|
||||
}
|
||||
if !strings.Contains(msg, "KANDEV_WAKE_PAYLOAD_JSON") {
|
||||
t.Fatalf("missing largest env key diagnostics: %s", msg)
|
||||
}
|
||||
if !errors.Is(err, syscall.E2BIG) {
|
||||
t.Fatalf("formatted error must preserve E2BIG wrapping: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ Follow this checklist each time you are woken up.
|
||||
|
||||
## 8-Step Wakeup Procedure
|
||||
|
||||
1. **Read the wake reason** from the environment variable `KANDEV_WAKE_REASON` and the payload from `KANDEV_WAKE_PAYLOAD_JSON`.
|
||||
1. **Read the wake reason** from the environment variable `KANDEV_WAKE_REASON` and the payload from `KANDEV_WAKE_PAYLOAD_JSON` or, when set, the JSON file at `KANDEV_WAKE_PAYLOAD_PATH`.
|
||||
|
||||
2. **If task_assigned**: Triage the task. Determine the domain, find the right delegate from your routing table, create subtasks, and assign them. Post a comment explaining your delegation decision.
|
||||
|
||||
|
||||
@@ -70,9 +70,10 @@ Your session ends. The orchestrator will wake you with reason
|
||||
## On wakeup after resolution
|
||||
|
||||
When you wake with `KANDEV_WAKE_REASON=task_blockers_resolved`, parse
|
||||
`$KANDEV_WAKE_PAYLOAD_JSON` for the resolved blocker titles, then continue
|
||||
your work incorporating the human's decision (visible in the resolved task's
|
||||
comments or description).
|
||||
`$KANDEV_WAKE_PAYLOAD_JSON` for the resolved blocker titles. If it is absent,
|
||||
read the JSON file at `$KANDEV_WAKE_PAYLOAD_PATH` instead. Then continue your
|
||||
work incorporating the human's decision (visible in the resolved task's comments
|
||||
or description).
|
||||
|
||||
## Rules
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@ These are injected into your session automatically. Do not hardcode them.
|
||||
| `KANDEV_WAKE_REASON` | Why you were woken (see wake reasons below) |
|
||||
| `KANDEV_WAKE_COMMENT_ID` | Comment ID that triggered the wake (if applicable) |
|
||||
| `KANDEV_WAKE_PAYLOAD_JSON` | Pre-computed task context -- parse this first |
|
||||
| `KANDEV_WAKE_PAYLOAD_PATH` | Workspace-relative JSON file path when the payload is too large for inline env |
|
||||
|
||||
Note: `KANDEV_API_URL` and `KANDEV_API_KEY` are also set but you do not need
|
||||
to use them directly. The CLI handles authentication and run-ID headers for you.
|
||||
@@ -47,8 +48,10 @@ Check `$KANDEV_WAKE_REASON`. Possible values:
|
||||
|
||||
### Step 2: Parse wake payload
|
||||
|
||||
If `$KANDEV_WAKE_PAYLOAD_JSON` is set, parse it. It contains pre-computed context
|
||||
so you don't need to fetch it from the API (saves tokens):
|
||||
If `$KANDEV_WAKE_PAYLOAD_JSON` is set, parse it. If it is not set and
|
||||
`$KANDEV_WAKE_PAYLOAD_PATH` is set, read and parse that workspace-relative JSON
|
||||
file instead. The payload contains pre-computed context so you don't need to
|
||||
fetch it from the API (saves tokens):
|
||||
|
||||
```json
|
||||
{
|
||||
@@ -227,7 +230,7 @@ fi
|
||||
(missing permissions, external dependency), post a comment explaining why and
|
||||
exit. Do not set the task to done if it is not actually done.
|
||||
|
||||
5. **Parse KANDEV_WAKE_PAYLOAD_JSON first.** It contains pre-computed context.
|
||||
5. **Parse KANDEV_WAKE_PAYLOAD_JSON first.** If it is absent, read `KANDEV_WAKE_PAYLOAD_PATH`. The payload contains pre-computed context.
|
||||
Only call the API for data not in the payload. This saves tokens.
|
||||
|
||||
6. **Keep comments concise but informative.** Other agents and humans read them.
|
||||
|
||||
@@ -221,6 +221,7 @@ Injected before each agent session:
|
||||
| `KANDEV_WAKE_REASON` | Reason string | Why the agent was woken |
|
||||
| `KANDEV_WAKE_COMMENT_ID` | Comment ID (if applicable) | Which comment triggered wake |
|
||||
| `KANDEV_WAKE_PAYLOAD_JSON` | Inline JSON | Pre-computed task context |
|
||||
| `KANDEV_WAKE_PAYLOAD_PATH` | Workspace-relative JSON file path | Pre-computed task context when too large for inline env |
|
||||
| `KANDEV_CLI` | Path to agentctl | CLI binary for API operations |
|
||||
|
||||
`KANDEV_CLI` resolves per executor:
|
||||
@@ -230,7 +231,7 @@ Injected before each agent session:
|
||||
|
||||
### Wake payload
|
||||
|
||||
`KANDEV_WAKE_PAYLOAD_JSON` carries pre-computed context. Fresh session: full task context (`task` object with id, identifier, title, status, priority, project, `blockedBy`, `childTasks`). Resume: only new comments since last run plus a `commentWindow` rollup (`{total, included, fetchMore}`). New comments include author, body, createdAt.
|
||||
`KANDEV_WAKE_PAYLOAD_JSON` carries pre-computed context. Fresh session: full task context (`task` object with id, identifier, title, status, priority, project, `blockedBy`, `childTasks`). Resume: only new comments since last run plus a `commentWindow` rollup (`{total, included, fetchMore}`). New comments include author, body, createdAt. If the serialized payload exceeds 64KB for inline environment delivery, Kandev writes it under the workspace and sets `KANDEV_WAKE_PAYLOAD_PATH` to that workspace-relative file path instead.
|
||||
|
||||
### Instructions delivery
|
||||
|
||||
@@ -297,7 +298,7 @@ When the scheduler processes a wakeup:
|
||||
5. Clean `kandev-*` from the skill dir; write desired skills to `<worktree>/<ProjectSkillDir>/kandev-<slug>/SKILL.md`; ensure `.git/info/exclude` has `kandev-*` patterns.
|
||||
6. Build prompt: read `AGENTS.md` content, append path directive, prepend to user-turn prompt, add wake context. For CEO heartbeat: add workspace status section.
|
||||
7. Set env vars (`KANDEV_API_KEY`, `KANDEV_TASK_ID`, `KANDEV_CLI`, etc.).
|
||||
8. Set `KANDEV_WAKE_PAYLOAD_JSON` with pre-computed task context.
|
||||
8. Set `KANDEV_WAKE_PAYLOAD_JSON` with pre-computed task context, or `KANDEV_WAKE_PAYLOAD_PATH` when the payload is too large for inline env.
|
||||
9. Launch agent via the task starter (prompt + env, CWD = worktree). Skills are cleaned up automatically when the worktree is deleted at session end.
|
||||
|
||||
### Default instruction templates per role
|
||||
|
||||
Reference in New Issue
Block a user