fix(envd): distinguish client-cancel from errors in process stream logging

packages/envd/internal/logs/interceptor.go

@@ -2,6 +2,7 @@ package logs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -66,7 +67,7 @@ func NewUnaryLogInterceptor(logger *zerolog.Logger) connect.UnaryInterceptorFunc
 				Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string))
 
 			if err != nil {
-				l = l.Int("error_code", int(connect.CodeOf(err)))
+				l = l.Int("error_code", int(codeOf(err)))
 			}
 
 			if req != nil {
@@ -116,7 +117,7 @@ func LogServerStreamWithoutEvents[T any, R any](
 		Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string))
 
 	if err != nil {
-		logEvent = logEvent.Int("error_code", int(connect.CodeOf(err)))
+		logEvent = logEvent.Int("error_code", int(codeOf(err)))
 	} else {
 		logEvent = logEvent.Interface("response", nil)
 	}
@@ -146,7 +147,7 @@ func LogClientStreamWithoutEvents[T any, R any](
 		Str(string(OperationIDKey), ctx.Value(OperationIDKey).(string))
 
 	if err != nil {
-		logEvent = logEvent.Int("error_code", int(connect.CodeOf(err)))
+		logEvent = logEvent.Int("error_code", int(codeOf(err)))
 	}
 
 	if res != nil && err == nil {
@@ -162,11 +163,39 @@ func LogClientStreamWithoutEvents[T any, R any](
 	return res, err
 }
 
-// Return logger with error level if err is not nil, otherwise return logger with debug level
+// getErrDebugLogEvent picks the log level by the nature of err, so a routine
+// client-side cancellation of a streaming call (the client stopped reading the
+// stream — it timed out, was cancelled by a sibling task, or the caller went
+// away) is not logged at ERROR next to genuine failures.
+//
+//   - nil                      -> Debug (normal completion)
+//   - context.Canceled         -> Info  (client went away; expected, not a fault)
+//   - context.DeadlineExceeded -> Warn  (the call outran its deadline)
+//   - anything else            -> Error (a real failure)
 func getErrDebugLogEvent(logger *zerolog.Logger, err error) *zerolog.Event {
-	if err != nil {
+	switch {
+	case err == nil:
+		return logger.Debug() //nolint:zerologlint // this builds an event, it is not expected to return it
+	case errors.Is(err, context.Canceled):
+		return logger.Info().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it
+	case errors.Is(err, context.DeadlineExceeded):
+		return logger.Warn().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it
+	default:
 		return logger.Error().Err(err) //nolint:zerologlint // this builds an event, it is not expected to return it
 	}
+}
 
-	return logger.Debug() //nolint:zerologlint // this builds an event, it is not expected to return it
+// codeOf maps an error to a connect code, recognizing the standard context
+// sentinels that connect.CodeOf otherwise reports as CodeUnknown. This keeps
+// error_code meaningful for downstream aggregation: a client cancellation is
+// CodeCanceled and a deadline is CodeDeadlineExceeded, not Unknown.
+func codeOf(err error) connect.Code {
+	switch {
+	case errors.Is(err, context.Canceled):
+		return connect.CodeCanceled
+	case errors.Is(err, context.DeadlineExceeded):
+		return connect.CodeDeadlineExceeded
+	default:
+		return connect.CodeOf(err)
+	}
 }

packages/envd/internal/logs/interceptor_test.go

@@ -0,0 +1,89 @@
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"testing"
+
+	"connectrpc.com/connect"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// levelOf runs getErrDebugLogEvent against err and returns the zerolog level
+// the emitted event carried.
+func levelOf(t *testing.T, err error) string {
+	t.Helper()
+
+	var buf bytesBuffer
+	logger := zerolog.New(&buf)
+	getErrDebugLogEvent(&logger, err).Msg("test")
+
+	var parsed map[string]any
+	require.NoError(t, json.Unmarshal(buf.b, &parsed), "log line: %s", string(buf.b))
+
+	lvl, _ := parsed["level"].(string)
+
+	return lvl
+}
+
+// bytesBuffer is a tiny io.Writer so the test needs no extra deps beyond what
+// the package already uses.
+type bytesBuffer struct{ b []byte }
+
+func (w *bytesBuffer) Write(p []byte) (int, error) {
+	w.b = append(w.b, p...)
+
+	return len(p), nil
+}
+
+func TestGetErrDebugLogEvent_Level(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"nil is debug", nil, "debug"},
+		{"context canceled is info", context.Canceled, "info"},
+		{"wrapped canceled is info", fmt.Errorf("stream canceled before start event: %w", context.Canceled), "info"},
+		{"deadline is warn", context.DeadlineExceeded, "warn"},
+		{"wrapped deadline is warn", fmt.Errorf("x: %w", context.DeadlineExceeded), "warn"},
+		{"real failure is error", errors.New("boom"), "error"},
+		{"connect internal is error", connect.NewError(connect.CodeInternal, errors.New("boom")), "error"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, levelOf(t, tc.err))
+		})
+	}
+}
+
+func TestCodeOf(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		err  error
+		want connect.Code
+	}{
+		{"canceled -> Canceled (not Unknown)", context.Canceled, connect.CodeCanceled},
+		{"wrapped canceled -> Canceled", fmt.Errorf("stream canceled: %w", context.Canceled), connect.CodeCanceled},
+		{"deadline -> DeadlineExceeded", context.DeadlineExceeded, connect.CodeDeadlineExceeded},
+		{"plain error -> Unknown", errors.New("boom"), connect.CodeUnknown},
+		{"connect code preserved", connect.NewError(connect.CodeInvalidArgument, errors.New("x")), connect.CodeInvalidArgument},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, codeOf(tc.err))
+		})
+	}
+}

packages/envd/internal/services/process/start.go

@@ -86,7 +86,11 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star
 
 		select {
 		case <-ctx.Done():
-			cancel(ctx.Err())
+			// Client went away before the start event was sent (it stopped
+			// reading the stream). Record the cause so the "server stream end"
+			// log line can tell a client cancel from a deadline; the process
+			// itself keeps running on its own context (see procCtx above).
+			cancel(fmt.Errorf("stream canceled before start event: %w", context.Cause(ctx)))
 
 			return
 		case event := <-start:
@@ -122,7 +126,9 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star
 					return
 				}
 			case <-ctx.Done():
-				cancel(ctx.Err())
+				// Client stopped reading mid-stream (timeout / sibling cancel /
+				// caller exit). The process is unaffected; only the stream ends.
+				cancel(fmt.Errorf("stream canceled while streaming output: %w", context.Cause(ctx)))
 
 				return
 			case event, ok := <-data:
@@ -147,7 +153,8 @@ func (s *Service) handleStart(ctx context.Context, req *connect.Request[rpc.Star
 
 		select {
 		case <-ctx.Done():
-			cancel(ctx.Err())
+			// Client went away while waiting for the terminal exit event.
+			cancel(fmt.Errorf("stream canceled before end event: %w", context.Cause(ctx)))
 
 			return
 		case event, ok := <-end: