summaryrefslogtreecommitdiff
path: root/internal
diff options
context:
space:
mode:
Diffstat (limited to 'internal')
-rw-r--r--internal/executor/container.go8
-rw-r--r--internal/executor/container_test.go16
-rw-r--r--internal/executor/executor.go309
3 files changed, 166 insertions, 167 deletions
diff --git a/internal/executor/container.go b/internal/executor/container.go
index 39f737a..a00448c 100644
--- a/internal/executor/container.go
+++ b/internal/executor/container.go
@@ -437,7 +437,7 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto
} else {
// No commits pushed — check whether the agent left uncommitted work behind.
// If so, fail loudly: the work would be silently lost when the sandbox is deleted.
- if err := detectUncommittedChanges(workspace); err != nil {
+ if err := r.detectUncommittedChanges(ctx, workspace); err != nil {
return err
}
r.Logger.Info("no new commits to push", "taskID", t.ID)
@@ -610,9 +610,9 @@ func isScaffold(path string) bool {
// untracked source files that the agent forgot to commit. Scaffold files written by
// the harness (.claudomator-env, .claudomator-instructions.txt, .agent-home/) are
// excluded from the check.
-func detectUncommittedChanges(workspace string) error {
+func (r *ContainerRunner) detectUncommittedChanges(ctx context.Context, workspace string) error {
// Modified or staged tracked files
- diffOut, err := exec.Command("git", "-c", "safe.directory=*", "-C", workspace,
+ diffOut, err := r.command(ctx, "git", "-c", "safe.directory=*", "-C", workspace,
"diff", "--name-only", "HEAD").CombinedOutput()
if err == nil {
for _, line := range strings.Split(strings.TrimSpace(string(diffOut)), "\n") {
@@ -623,7 +623,7 @@ func detectUncommittedChanges(workspace string) error {
}
// Untracked new source files (excludes gitignored files)
- lsOut, err := exec.Command("git", "-c", "safe.directory=*", "-C", workspace,
+ lsOut, err := r.command(ctx, "git", "-c", "safe.directory=*", "-C", workspace,
"ls-files", "--others", "--exclude-standard").CombinedOutput()
if err == nil {
var dirty []string
diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go
index dec666e..8c00a0f 100644
--- a/internal/executor/container_test.go
+++ b/internal/executor/container_test.go
@@ -312,7 +312,9 @@ func TestDetectUncommittedChanges_ModifiedFile(t *testing.T) {
if err := os.WriteFile(dir+"/main.go", []byte("package main\n// changed"), 0644); err != nil {
t.Fatal(err)
}
- err := detectUncommittedChanges(dir)
+ runner := &ContainerRunner{}
+ err := runner.detectUncommittedChanges(context.Background(), dir)
+
if err == nil {
t.Fatal("expected error for modified uncommitted file, got nil")
}
@@ -338,7 +340,9 @@ func TestDetectUncommittedChanges_NewUntrackedSourceFile(t *testing.T) {
if err := os.WriteFile(dir+"/newfile.go", []byte("package main"), 0644); err != nil {
t.Fatal(err)
}
- err := detectUncommittedChanges(dir)
+ runner := &ContainerRunner{}
+ err := runner.detectUncommittedChanges(context.Background(), dir)
+
if err == nil {
t.Fatal("expected error for new untracked source file, got nil")
}
@@ -361,7 +365,9 @@ func TestDetectUncommittedChanges_ScaffoldFilesIgnored(t *testing.T) {
_ = os.WriteFile(dir+"/.claudomator-env", []byte("KEY=val"), 0600)
_ = os.WriteFile(dir+"/.claudomator-instructions.txt", []byte("do stuff"), 0644)
_ = os.MkdirAll(dir+"/.agent-home/.claude", 0755)
- err := detectUncommittedChanges(dir)
+ runner := &ContainerRunner{}
+ err := runner.detectUncommittedChanges(context.Background(), dir)
+
if err != nil {
t.Errorf("scaffold files should not trigger uncommitted error, got: %v", err)
}
@@ -385,7 +391,9 @@ func TestDetectUncommittedChanges_CleanRepo(t *testing.T) {
run("git", "add", "main.go")
run("git", "commit", "-m", "init")
// No modifications — should pass
- err := detectUncommittedChanges(dir)
+ runner := &ContainerRunner{}
+ err := runner.detectUncommittedChanges(context.Background(), dir)
+
if err != nil {
t.Errorf("clean repo should not error, got: %v", err)
}
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index fbe29a8..507c65b 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -306,47 +306,40 @@ func (p *Pool) decActiveAgent(agentType string, cleaned *bool) {
p.mu.Unlock()
}
-func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
- defer p.releaseDispatchSlot()
-
- agentType := t.Agent.Type
- if agentType == "" {
- agentType = "claude"
- }
-
+// withWorkerSlot manages the lifecycle of a worker: per-agent
+// slots, context timeout/cancellation, and cancellation registration. It
+// provides a finish callback to settle per-agent accounting early (e.g. before
+// reporting results).
+func (p *Pool) withWorkerSlot(ctx context.Context, t *task.Task, agentType string, enforceCapacity bool, fn func(ctx context.Context, finish func()) error) {
p.mu.Lock()
+ if enforceCapacity && p.activePerAgent[agentType] >= p.maxPerAgent {
+ p.mu.Unlock()
+ time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
+ return
+ }
+ if deadline, ok := p.rateLimited[agentType]; ok && time.Now().After(deadline) {
+ delete(p.rateLimited, agentType)
+ agentName := agentType
+ go func() {
+ ev := storage.AgentEvent{
+ ID: uuid.New().String(),
+ Agent: agentName,
+ Event: "available",
+ Timestamp: time.Now(),
+ }
+ if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
+ p.logger.Warn("failed to record agent available event", "error", recErr)
+ }
+ }()
+ }
p.activePerAgent[agentType]++
p.mu.Unlock()
var cleaned bool
- defer p.decActiveAgent(agentType, &cleaned)
-
- runner, err := p.getRunner(t)
- if err != nil {
- p.logger.Error("failed to get runner for resume", "error", err, "taskID", t.ID)
+ finish := func() {
p.decActiveAgent(agentType, &cleaned)
- p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
- return
- }
-
- // Pre-populate log paths.
- if lp, ok := runner.(LogPather); ok {
- if logDir := lp.ExecLogDir(exec.ID); logDir != "" {
- exec.StdoutPath = filepath.Join(logDir, "stdout.log")
- exec.StderrPath = filepath.Join(logDir, "stderr.log")
- exec.ArtifactDir = logDir
- }
- }
- exec.StartTime = time.Now().UTC()
- exec.Status = "RUNNING"
-
- if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
- p.logger.Error("failed to create resume execution record", "error", err)
- }
- select {
- case p.startedCh <- t.ID:
- default:
}
+ defer finish()
var cancel context.CancelFunc
if t.Timeout.Duration > 0 {
@@ -364,28 +357,68 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex
p.mu.Unlock()
}()
- // Populate RepositoryURL from Project registry if missing (ADR-007).
- if t.RepositoryURL == "" && t.Project != "" {
- if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
- t.RepositoryURL = proj.RemoteURL
- }
+ _ = fn(ctx, finish)
+}
+
+func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) {
+ agentType := t.Agent.Type
+ if agentType == "" {
+ agentType = "claude"
}
- // Populate BranchName from Story if missing (ADR-007).
- if t.BranchName == "" && t.StoryID != "" {
- if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" {
- t.BranchName = story.BranchName
+
+ p.withWorkerSlot(ctx, t, agentType, false, func(ctx context.Context, finish func()) error {
+ runner, err := p.getRunner(t)
+ if err != nil {
+ p.logger.Error("failed to get runner for resume", "error", err, "taskID", t.ID)
+ finish()
+ p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
+ return nil
}
- }
- sc := newStoreChannel(p.store, t.ID)
- err = runner.Run(ctx, t, exec, sc)
- exec.EndTime = time.Now().UTC()
- if sum, ok := sc.ReportedSummary(); ok {
- exec.Summary = sum
- }
+ // Pre-populate log paths.
+ if lp, ok := runner.(LogPather); ok {
+ if logDir := lp.ExecLogDir(exec.ID); logDir != "" {
+ exec.StdoutPath = filepath.Join(logDir, "stdout.log")
+ exec.StderrPath = filepath.Join(logDir, "stderr.log")
+ exec.ArtifactDir = logDir
+ }
+ }
+ exec.StartTime = time.Now().UTC()
+ exec.Status = "RUNNING"
+
+ if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
+ p.logger.Error("failed to create resume execution record", "error", err)
+ }
+ select {
+ case p.startedCh <- t.ID:
+ default:
+ }
+
+ // Populate RepositoryURL from Project registry if missing (ADR-007).
+ if t.RepositoryURL == "" && t.Project != "" {
+ if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
+ t.RepositoryURL = proj.RemoteURL
+ }
+ }
+ // Populate BranchName from Story if missing (ADR-007).
+ if t.BranchName == "" && t.StoryID != "" {
+ if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" {
+ t.BranchName = story.BranchName
+ }
+ }
+
+ sc := newStoreChannel(p.store, t.ID)
+ err = runner.Run(ctx, t, exec, sc)
+ exec.EndTime = time.Now().UTC()
+ if sum, ok := sc.ReportedSummary(); ok {
+ exec.Summary = sum
+ }
+
+ finish()
+ p.handleRunResult(ctx, t, exec, err, agentType)
+ return nil
+ })
- p.decActiveAgent(agentType, &cleaned)
- p.handleRunResult(ctx, t, exec, err, agentType)
}
// handleRunResult applies the shared post-run error-classification and
@@ -1042,129 +1075,87 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
return
}
}
- p.mu.Lock()
-
- if p.activePerAgent[agentType] >= p.maxPerAgent {
- p.mu.Unlock()
- time.AfterFunc(p.requeueDelay, func() { p.workCh <- workItem{ctx: ctx, task: t} })
- return
- }
- if deadline, ok := p.rateLimited[agentType]; ok && time.Now().After(deadline) {
- delete(p.rateLimited, agentType)
- agentName := agentType
- go func() {
- ev := storage.AgentEvent{
+ p.withWorkerSlot(ctx, t, agentType, true, func(ctx context.Context, finish func()) error {
+ runner, err := p.getRunner(t)
+ if err != nil {
+ p.logger.Error("failed to get runner", "error", err, "taskID", t.ID)
+ now := time.Now().UTC()
+ exec := &storage.Execution{
ID: uuid.New().String(),
- Agent: agentName,
- Event: "available",
- Timestamp: time.Now(),
+ TaskID: t.ID,
+ StartTime: now,
+ EndTime: now,
+ Status: "FAILED",
+ ErrorMsg: err.Error(),
}
- if recErr := p.store.RecordAgentEvent(ev); recErr != nil {
- p.logger.Warn("failed to record agent available event", "error", recErr)
+ if createErr := p.store.CreateExecution(exec); createErr != nil {
+ p.logger.Error("failed to create execution record", "error", createErr)
}
- }()
- }
- p.activePerAgent[agentType]++
- p.mu.Unlock()
-
- var cleaned bool
- defer p.decActiveAgent(agentType, &cleaned)
+ if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
+ p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
+ }
+ finish()
+ p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
+ return nil
+ }
- runner, err := p.getRunner(t)
- if err != nil {
- p.logger.Error("failed to get runner", "error", err, "taskID", t.ID)
- now := time.Now().UTC()
+ execID := uuid.New().String()
exec := &storage.Execution{
- ID: uuid.New().String(),
+ ID: execID,
TaskID: t.ID,
- StartTime: now,
- EndTime: now,
- Status: "FAILED",
- ErrorMsg: err.Error(),
- }
- if createErr := p.store.CreateExecution(exec); createErr != nil {
- p.logger.Error("failed to create execution record", "error", createErr)
+ StartTime: time.Now().UTC(),
+ Status: "RUNNING",
+ Agent: agentType,
}
- if err := p.store.UpdateTaskState(t.ID, task.StateFailed); err != nil {
- p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateFailed, "error", err)
- }
- p.decActiveAgent(agentType, &cleaned)
- p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: err}
- return
- }
-
- execID := uuid.New().String()
- exec := &storage.Execution{
- ID: execID,
- TaskID: t.ID,
- StartTime: time.Now().UTC(),
- Status: "RUNNING",
- Agent: agentType,
- }
- // Pre-populate log paths so they're available in the DB immediately —
- // before the subprocess starts — enabling live tailing and debugging.
- if lp, ok := runner.(LogPather); ok {
- if logDir := lp.ExecLogDir(execID); logDir != "" {
- exec.StdoutPath = filepath.Join(logDir, "stdout.log")
- exec.StderrPath = filepath.Join(logDir, "stderr.log")
- exec.ArtifactDir = logDir
+ // Pre-populate log paths so they're available in the DB immediately —
+ // before the subprocess starts — enabling live tailing and debugging.
+ if lp, ok := runner.(LogPather); ok {
+ if logDir := lp.ExecLogDir(execID); logDir != "" {
+ exec.StdoutPath = filepath.Join(logDir, "stdout.log")
+ exec.StderrPath = filepath.Join(logDir, "stderr.log")
+ exec.ArtifactDir = logDir
+ }
}
- }
-
- // Record execution start atomically with the RUNNING state transition.
- if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
- p.logger.Error("failed to create execution record", "error", err)
- }
- select {
- case p.startedCh <- t.ID:
- default:
- }
- // Apply task timeout and register cancel so callers can stop this task.
- var cancel context.CancelFunc
- if t.Timeout.Duration > 0 {
- ctx, cancel = context.WithTimeout(ctx, t.Timeout.Duration)
- } else {
- ctx, cancel = context.WithCancel(ctx)
- }
- p.mu.Lock()
- p.cancels[t.ID] = cancel
- p.mu.Unlock()
- defer func() {
- cancel()
- p.mu.Lock()
- delete(p.cancels, t.ID)
- p.mu.Unlock()
- }()
+ // Record execution start atomically with the RUNNING state transition.
+ if err := p.store.CreateExecutionAndSetRunning(exec); err != nil {
+ p.logger.Error("failed to create execution record", "error", err)
+ }
+ select {
+ case p.startedCh <- t.ID:
+ default:
+ }
- // Inject prior failure history so the agent knows what went wrong before.
- priorExecs, priorErr := p.store.ListExecutions(t.ID)
- t = withFailureHistory(t, priorExecs, priorErr)
+ // Inject prior failure history so the agent knows what went wrong before.
+ priorExecs, priorErr := p.store.ListExecutions(t.ID)
+ t = withFailureHistory(t, priorExecs, priorErr)
- // Populate RepositoryURL from Project registry if missing (ADR-007).
- if t.RepositoryURL == "" && t.Project != "" {
- if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
- t.RepositoryURL = proj.RemoteURL
+ // Populate RepositoryURL from Project registry if missing (ADR-007).
+ if t.RepositoryURL == "" && t.Project != "" {
+ if proj, err := p.store.GetProject(t.Project); err == nil && proj.RemoteURL != "" {
+ t.RepositoryURL = proj.RemoteURL
+ }
}
- }
- // Populate BranchName from Story if missing (ADR-007).
- if t.BranchName == "" && t.StoryID != "" {
- if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" {
- t.BranchName = story.BranchName
+ // Populate BranchName from Story if missing (ADR-007).
+ if t.BranchName == "" && t.StoryID != "" {
+ if story, err := p.store.GetStory(t.StoryID); err == nil && story.BranchName != "" {
+ t.BranchName = story.BranchName
+ }
}
- }
- // Run the task.
- sc := newStoreChannel(p.store, t.ID)
- err = runner.Run(ctx, t, exec, sc)
- exec.EndTime = time.Now().UTC()
- if sum, ok := sc.ReportedSummary(); ok {
- exec.Summary = sum
- }
+ // Run the task.
+ sc := newStoreChannel(p.store, t.ID)
+ err = runner.Run(ctx, t, exec, sc)
+ exec.EndTime = time.Now().UTC()
+ if sum, ok := sc.ReportedSummary(); ok {
+ exec.Summary = sum
+ }
- p.decActiveAgent(agentType, &cleaned)
- p.handleRunResult(ctx, t, exec, err, agentType)
+ finish()
+ p.handleRunResult(ctx, t, exec, err, agentType)
+ return nil
+ })
}
// RecoverStaleRunning marks any tasks stuck in RUNNING state (from a previous
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 10:31:43 +0000