internal/service/scheduler/zombie_detector.go (2)

16-22: Move panicToError into the recover block or enrich it (stack / “panic:” prefix).

Current helper is fine, but it drops stack context; consider attaching a stack (or at least prefixing the message) so panics are actionable in logs.

---

69-73: Recover logging: consider emitting stack trace (and/or recovered value separately).

Right now you only log an error, which makes panics hard to debug in prod. If you already have a standard for stack logging elsewhere, reuse it here.

internal/service/frontend/server.go (2)

70-73: NewServer(ctx, ...) API change is good; consider nil-ctx guard to prevent panics.
Go convention is “context must not be nil”; a defensive fallback avoids surprising crashes in older call sites/tests.

Proposed guard

 func NewServer(ctx context.Context, cfg *config.Config, dr execution.DAGStore, drs execution.DAGRunStore, qs execution.QueueStore, ps execution.ProcStore, drm runtime.Manager, cc coordinator.Client, sr execution.ServiceRegistry, mr *prometheus.Registry, collector *telemetry.Collector, rs *resource.Service) (*Server, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	var remoteNodes []string

---

134-146: Good: builtin OIDC init now respects startup cancellation.
Minor: the “OIDC enabled…” log uses context.Background() (Line 146); if you want consistent structured tags/cancellation semantics, prefer ctx.

internal/integration/queue_shell_test.go (1)

40-58: Verify context cancellation already handles cleanup, but explicit server shutdown is still recommended.

The test's th.Context is reliably cancelled during test teardown (via helper.Cleanup() which calls Cancel()), so the server goroutine will exit. However, cancelling the context alone doesn't guarantee the HTTP listener is fully released. Adding explicit server.Shutdown() follows Go HTTP server best practices and ensures proper resource cleanup without relying on context cancellation timing.

Suggested cleanup

 		server, err := frontend.NewServer(
 			th.Context,
 			th.Config,
 			th.DAGStore,
 			th.DAGRunStore,
 			th.QueueStore,
 			th.ProcStore,
 			th.DAGRunMgr,
 			nil, // no coordinator client for local execution
 			th.ServiceRegistry,
 			nil, // no metrics registry
 			nil, // no telemetry collector
 			nil, // no resource service
 		)
 		require.NoError(t, err, "failed to create server")
+
+		t.Cleanup(func() {
+			_ = server.Shutdown(context.Background())
+		})

 		go func() {
 			_ = server.Serve(th.Context)
 		}()

internal/service/frontend/auth/oidc.go (1)

71-107: Nice improvement: OIDC init + retries are now cancellable.
Two follow-ups worth considering: (1) add defensive nil check for ctx in backoff.Retry to prevent panic, (2) wrap obviously-permanent errors (bad issuer URL format, unsupported scheme) with backoff.PermanentError() to avoid 10 unnecessary retry attempts for errors that will never succeed.

internal/common/stringutil/spinner.go (1)

5-5: Consider making SpinnerFrames immutable to prevent accidental modification.

While exported package-level slices are acceptable, they can be reassigned or mutated by consumers. Since these frames are intended to be a constant set of values, consider either:

1. Using a function that returns a copy: func SpinnerFrames() []string
2. Documenting that the slice should not be modified
3. Accepting the current pattern as idiomatic for shared read-only data

The current approach is functional, but documenting the intended immutability would help prevent misuse.

♻️ Option 1: Use a function to return frames

-// SpinnerFrames contains the animation frames for a braille spinner.
-// Used by progress displays throughout the application.
-var SpinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"}
+var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"}
+
+// SpinnerFrames returns the animation frames for a braille spinner.
+// Used by progress displays throughout the application.
+func SpinnerFrames() []string {
+	return spinnerFrames
+}

♻️ Option 2: Add documentation warning

-// SpinnerFrames contains the animation frames for a braille spinner.
-// Used by progress displays throughout the application.
+// SpinnerFrames contains the animation frames for a braille spinner.
+// Used by progress displays throughout the application.
+// This slice should be treated as read-only and not modified.
 var SpinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"}

internal/service/coordinator/log_handler.go (1)

243-256: Consider renaming function for clarity.

The function name StreamTypeToExtension suggests it returns just an extension (e.g., "log" or "stdout"), but it actually returns a filename suffix including the .log extension (e.g., "stdout.log", "scheduler.log").

Consider renaming to something like StreamTypeToFileSuffix or updating the return values to be true extensions without .log, then appending .log at the call site.

♻️ Alternative: Return true extensions

If you prefer to keep the function name as-is, you could return actual extensions and append .log at line 231:

-func StreamTypeToExtension(streamType coordinatorv1.LogStreamType) string {
+func StreamTypeToFileExtension(streamType coordinatorv1.LogStreamType) string {
 	switch streamType {
 	case coordinatorv1.LogStreamType_LOG_STREAM_TYPE_STDOUT:
-		return "stdout.log"
+		return "stdout"
 	case coordinatorv1.LogStreamType_LOG_STREAM_TYPE_STDERR:
-		return "stderr.log"
+		return "stderr"
 	case coordinatorv1.LogStreamType_LOG_STREAM_TYPE_SCHEDULER:
-		return "scheduler.log"
+		return "scheduler"
 	case coordinatorv1.LogStreamType_LOG_STREAM_TYPE_UNSPECIFIED:
 		return "log"
 	}
 	return "log"
 }

Then update line 231:

-		filename = fmt.Sprintf("%s.%s", fileutil.SafeName(chunk.StepName), ext)
+		filename = fmt.Sprintf("%s.%s.log", fileutil.SafeName(chunk.StepName), ext)

Note: This would require updating test expectations as well.

internal/persistence/filedagrun/store.go (1)

435-457: Consider adding locking for sub-attempt creation.

The CreateAttempt method (lines 259-310) acquires a lock before creating a new DAG run or attempt to prevent race conditions. Since CreateSubAttempt also performs directory creation and file operations, it should similarly acquire a lock to ensure thread-safety and prevent concurrent creation conflicts.

🔒 Proposed fix to add locking

 func (store *Store) CreateSubAttempt(ctx context.Context, rootRef execution.DAGRunRef, subDAGRunID string) (execution.DAGRunAttempt, error) {
 	if rootRef.ID == "" {
 		return nil, ErrDAGRunIDEmpty
 	}
 
 	root := NewDataRoot(store.baseDir, rootRef.Name)
 	dagRun, err := root.FindByDAGRunID(ctx, rootRef.ID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find root dag-run: %w", err)
 	}
 
+	lockCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	if err := root.Lock(lockCtx); err != nil {
+		return nil, fmt.Errorf("failed to acquire lock for sub dag-run %s: %w", subDAGRunID, err)
+	}
+	defer func() {
+		if err := root.Unlock(); err != nil {
+			logger.Error(ctx, "Failed to unlock sub dag-run", tag.RunID(subDAGRunID), tag.Error(err))
+		}
+	}()
+
 	// Create the sub-DAG run directory
 	subDAGRun, err := dagRun.CreateSubDAGRun(ctx, subDAGRunID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create sub dag-run directory: %w", err)
 	}
 
 	// Create an attempt within the sub-DAG run
 	return subDAGRun.CreateAttempt(ctx, execution.NewUTC(time.Now()), store.cache)
 }

internal/runtime/executor/dag_runner.go (1)

535-577: Consider using a timeout context for cancellation requests.

Using context.Background() at line 535 means cancellation requests have no timeout. If the coordinator or database is unresponsive, this could cause the Kill method to hang indefinitely. Consider using a timeout context:

⏱️ Proposed fix to add timeout context

 func (e *SubDAGExecutor) Kill(sig os.Signal) error {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
 	var errs []error
-	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
 
 	// Cancel distributed runs
 	for runID := range e.distributedRuns {

internal/cmd/retry.go (1)

180-237: Consider reusing waitForDAGCompletionWithProgress and handleDistributedCancellation.

This function shares significant logic with dispatchToCoordinatorAndWait in start.go:

• Signal-aware context setup (lines 183-185)
• Progress display initialization (lines 188-202)
• Status polling with progress updates (line 229)
• Cancellation handling (lines 232-234)

The only differences are the task creation (using OPERATION_RETRY vs OPERATION_START and including WithPreviousStatus).

♻️ Suggested refactor to reduce duplication

Consider extracting the common orchestration logic into a shared helper:

// dispatchAndWait is a shared helper for dispatching tasks and waiting for completion
func dispatchAndWait(ctx *Context, dag *core.DAG, dagRunID string, coordinatorCli coordinator.Client, task *coordinatorv1.Task) error {
    // Signal-aware context, progress display, dispatch, wait, and cancellation handling
    // ... shared logic here
}

Then both dispatchToCoordinatorAndWait and dispatchRetryToCoordinatorAndWait can delegate to this helper with their respective tasks.

internal/runtime/remote/log_streamer.go (1)

94-169: Consider adding explicit stream closure on error paths.

When stream.Send(chunk) fails on line 143-145 or 162-163, the function returns immediately without calling CloseAndRecv(). While gRPC will eventually clean up the stream, explicitly closing it ensures proper resource release and server-side notification.

♻️ Suggested improvement

 func (s *LogStreamer) StreamSchedulerLog(ctx context.Context, logFilePath string) error {
 	// ... existing code ...
 
 	// Create a stream to the coordinator
 	stream, err := s.client.StreamLogs(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to create log stream: %w", err)
 	}
+	defer func() {
+		// Ensure stream is closed on all paths
+		_, _ = stream.CloseAndRecv()
+	}()
 
 	// Split into chunks if necessary (scheduler logs can be large)
 	var sequence uint64 = 0
 	for len(data) > 0 {
 		// ... existing chunking code ...
 
 		if err := stream.Send(chunk); err != nil {
-			return fmt.Errorf("failed to send scheduler log chunk: %w", err)
+			return fmt.Errorf("failed to send scheduler log chunk: %w", err)
 		}
 	}
 
 	// Send final marker
 	// ... existing code ...
 
-	// Close and get response
-	_, err = stream.CloseAndRecv()
-	return err
+	return nil  // defer handles CloseAndRecv
 }

internal/cmd/start.go (1)

569-626: LGTM with one observation.

The polling logic is robust with:

• Consecutive error tracking with a reasonable threshold (10 errors = ~10 seconds)
• Periodic logging when progress display is disabled
• Proper status checking for completion

The error message on line 622 could be more informative by including the actual error from the status if available (resp.Status.Error).

♻️ Minor improvement for error message

 			if !status.IsActive() {
 				if status.IsSuccess() {
 					logger.Info(ctx, "DAG completed successfully", tag.RunID(dagRunID))
 					return nil
 				}
-				return fmt.Errorf("DAG run failed with status: %s", status)
+				errMsg := resp.Status.Error
+				if errMsg == "" {
+					errMsg = "no error details available"
+				}
+				return fmt.Errorf("DAG run failed with status %s: %s", status, errMsg)
 			}

internal/service/coordinator/handler.go (2)

448-544: Log-path rewrite: handle empty step names (esp. handler nodes) to avoid “.stdout.log”.

transformLogPaths() uses node.Step.Name for filename generation; if that’s empty (common for synthetic/handler nodes depending on how they’re populated), you’ll end up with odd filenames like .stdout.log. Consider providing explicit names for handler nodes (onInit/onExit/…) or falling back when stepName == "".

---

52-60: Open-attempt + per-run mutex maps can grow unbounded in long-lived coordinators.

You’re intentionally keeping attempts open until coordinator shutdown; combined with runMutexes never being removed, this can become a slow FD/memory leak under many runs. If execution.DAGRunAttempt.Open() holds OS resources, this becomes a production reliability issue. Consider evicting/closing on terminal statuses (with “already-closed” tolerance) and deleting the per-run mutex entry once a run is terminal.

Also applies to: 109-123, 451-477

internal/runtime/agent/agent.go (1)

649-657: Avoid concrete *remote.LogStreamer type-assertion + guard against blocking shutdown.

This couples the agent to one concrete implementation and may silently skip scheduler-log streaming for wrappers/mocks. Also, streaming at the end can delay completion indefinitely if the coordinator is slow/unreachable. Consider asserting on a small interface and applying a timeout (optionally with context.WithoutCancel so it still attempts to flush on cancellation).

Proposed refactor

-	// Stream scheduler log to coordinator if using remote logging (shared-nothing mode)
-	if a.logWriterFactory != nil {
-		if streamer, ok := a.logWriterFactory.(*remote.LogStreamer); ok {
-			if err := streamer.StreamSchedulerLog(ctx, a.logFile); err != nil {
-				logger.Warn(ctx, "Failed to stream scheduler log", tag.Error(err))
-			}
-		}
-	}
+	// Stream scheduler log to coordinator if supported by the configured log writer factory.
+	type schedulerLogStreamer interface {
+		StreamSchedulerLog(context.Context, string) error
+	}
+	if streamer, ok := a.logWriterFactory.(schedulerLogStreamer); ok {
+		streamCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+		defer cancel()
+		if err := streamer.StreamSchedulerLog(streamCtx, a.logFile); err != nil {
+			logger.Warn(ctx, "Failed to stream scheduler log", tag.Error(err))
+		}
+	}

internal/cmd/progress_remote.go (2)

167-179: Non-TTY mode never prints a “finished” line.

For CI logs / redirected stderr, printHeader() prints “Started: …” once, but printFinal() returns early when !isTTY, so you don’t get an explicit completion line. Consider printing a final one-liner in non-TTY too.

Also applies to: 201-222

---

182-199: Cap progress percent (and handle completed > total) to avoid >100%.

If the coordinator status ever includes extra nodes (or total is underestimated), percent can exceed 100 and the (completed/total) can look wrong. Consider clamping completedShown := min(p.completed, p.total) and percent := min(percent, 100).

Also applies to: 201-222

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: CR
Repo: dagu-org/dagu PR: 0
File: AGENTS.md:0-0
Timestamp: 2025-12-04T10:34:17.062Z
Learning: Applies to **/*.go : Backend entrypoint in `cmd/` orchestrates the scheduler and CLI; runtime, persistence, and service layers sit under `internal/*` (for example `internal/runtime`, `internal/persistence`)

Learnt from: CR
Repo: dagu-org/dagu PR: 0
File: AGENTS.md:0-0
Timestamp: 2025-12-04T10:34:17.062Z
Learning: `make run` starts the Go scheduler and serves the compiled UI (fails fast if `ui/dist` is missing)