coder/coderd/workspaceagents.go

package coderd

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"
	"github.com/sqlc-dev/pqtype"
	"golang.org/x/exp/maps"
	"golang.org/x/sync/errgroup"
	"golang.org/x/xerrors"
	"tailscale.com/tailcfg"

	"cdr.dev/slog/v3"
	"github.com/coder/coder/v2/coderd/agentapi"
	"github.com/coder/coder/v2/coderd/agentapi/metadatabatcher"
	"github.com/coder/coder/v2/coderd/database"
	"github.com/coder/coder/v2/coderd/database/db2sdk"
	"github.com/coder/coder/v2/coderd/database/dbauthz"
	"github.com/coder/coder/v2/coderd/database/dbtime"
	"github.com/coder/coder/v2/coderd/externalauth"
	"github.com/coder/coder/v2/coderd/httpapi"
	"github.com/coder/coder/v2/coderd/httpmw"
	"github.com/coder/coder/v2/coderd/httpmw/loggermw"
	"github.com/coder/coder/v2/coderd/jwtutils"
	"github.com/coder/coder/v2/coderd/prebuilds"
	"github.com/coder/coder/v2/coderd/provisionerdserver"
	"github.com/coder/coder/v2/coderd/rbac"
	"github.com/coder/coder/v2/coderd/rbac/policy"
	"github.com/coder/coder/v2/coderd/telemetry"
	maputil "github.com/coder/coder/v2/coderd/util/maps"
	"github.com/coder/coder/v2/coderd/wspubsub"
	"github.com/coder/coder/v2/coderd/x/gitsync"
	"github.com/coder/coder/v2/codersdk"
	"github.com/coder/coder/v2/codersdk/agentsdk"
	"github.com/coder/coder/v2/codersdk/workspacesdk"
	"github.com/coder/coder/v2/codersdk/wsjson"
	"github.com/coder/coder/v2/tailnet"
	"github.com/coder/coder/v2/tailnet/proto"
	"github.com/coder/websocket"
)

// @Summary Get workspace agent by ID
// @ID get-workspace-agent-by-id
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Success 200 {object} codersdk.WorkspaceAgent
// @Router /api/v2/workspaceagents/{workspaceagent} [get]
func (api *API) workspaceAgent(rw http.ResponseWriter, r *http.Request) {
	var (
		ctx        = r.Context()
		waws       = httpmw.WorkspaceAgentAndWorkspaceParam(r)
		dbApps     []database.WorkspaceApp
		scripts    []database.GetWorkspaceAgentScriptsByAgentIDsRow
		logSources []database.WorkspaceAgentLogSource
	)

	var eg errgroup.Group
	eg.Go(func() (err error) {
		dbApps, err = api.Database.GetWorkspaceAppsByAgentID(ctx, waws.WorkspaceAgent.ID)
		return err
	})
	eg.Go(func() (err error) {
		//nolint:gocritic // TODO: can we make this not require system restricted?
		scripts, err = api.Database.GetWorkspaceAgentScriptsByAgentIDs(dbauthz.AsSystemRestricted(ctx), []uuid.UUID{waws.WorkspaceAgent.ID})
		return err
	})
	eg.Go(func() (err error) {
		//nolint:gocritic // TODO: can we make this not require system restricted?
		logSources, err = api.Database.GetWorkspaceAgentLogSourcesByAgentIDs(dbauthz.AsSystemRestricted(ctx), []uuid.UUID{waws.WorkspaceAgent.ID})
		return err
	})
	err := eg.Wait()
	if httpapi.Is404Error(err) {
		httpapi.ResourceNotFound(rw)
		return
	}
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching workspace agent.",
			Detail:  err.Error(),
		})
		return
	}

	appIDs := []uuid.UUID{}
	for _, app := range dbApps {
		appIDs = append(appIDs, app.ID)
	}
	// nolint:gocritic // This is a system restricted operation.
	statuses, err := api.Database.GetWorkspaceAppStatusesByAppIDs(dbauthz.AsSystemRestricted(ctx), appIDs)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching workspace app statuses.",
			Detail:  err.Error(),
		})
		return
	}

	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(), *api.TailnetCoordinator.Load(), waws.WorkspaceAgent, db2sdk.Apps(dbApps, statuses, waws.WorkspaceAgent, waws.OwnerUsername, waws.WorkspaceTable), convertScripts(scripts), convertLogSources(logSources), api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}

	httpapi.Write(ctx, rw, http.StatusOK, apiAgent)
}

const AgentAPIVersionREST = "1.0"

// @Summary Patch workspace agent logs
// @ID patch-workspace-agent-logs
// @Security CoderSessionToken
// @Accept json
// @Produce json
// @Tags Agents
// @Param request body agentsdk.PatchLogs true "logs"
// @Success 200 {object} codersdk.Response
// @Failure 413 {object} codersdk.Response "Agent log storage limit exceeded"
// @Router /api/v2/workspaceagents/me/logs [patch]
func (api *API) patchWorkspaceAgentLogs(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	workspaceAgent := httpmw.WorkspaceAgent(r)
	var req agentsdk.PatchLogs
	if !httpapi.Read(ctx, rw, r, &req) {
		return
	}
	if len(req.Logs) == 0 {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "No logs provided.",
		})
		return
	}
	// This is to support the legacy API where the log source ID was
	// not provided in the request body. We default to the external
	// log source in this case.
	if req.LogSourceID == uuid.Nil {
		// Use the external log source
		externalSources, err := api.Database.InsertWorkspaceAgentLogSources(ctx, database.InsertWorkspaceAgentLogSourcesParams{
			WorkspaceAgentID: workspaceAgent.ID,
			CreatedAt:        dbtime.Now(),
			ID:               []uuid.UUID{agentsdk.ExternalLogSourceID},
			DisplayName:      []string{"External"},
			Icon:             []string{"/emojis/1f310.png"},
		})
		if database.IsUniqueViolation(err, database.UniqueWorkspaceAgentLogSourcesPkey) {
			err = nil
			req.LogSourceID = agentsdk.ExternalLogSourceID
		}
		if err != nil {
			httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to create external log source.",
				Detail:  err.Error(),
			})
			return
		}
		if len(externalSources) == 1 {
			req.LogSourceID = externalSources[0].ID
		}
	}
	output := make([]string, 0)
	level := make([]database.LogLevel, 0)
	outputLength := 0
	for _, logEntry := range req.Logs {
		sanitizedOutput := agentsdk.SanitizeLogOutput(logEntry.Output)
		output = append(output, sanitizedOutput)
		outputLength += len(sanitizedOutput)
		if logEntry.Level == "" {
			// Default to "info" to support older agents that didn't have the level field.
			logEntry.Level = codersdk.LogLevelInfo
		}
		parsedLevel := database.LogLevel(logEntry.Level)
		if !parsedLevel.Valid() {
			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
				Message: "Invalid log level provided.",
				Detail:  fmt.Sprintf("invalid log level: %q", logEntry.Level),
			})
			return
		}
		level = append(level, parsedLevel)
	}

	logs, err := api.Database.InsertWorkspaceAgentLogs(ctx, database.InsertWorkspaceAgentLogsParams{
		AgentID:     workspaceAgent.ID,
		CreatedAt:   dbtime.Now(),
		Output:      output,
		Level:       level,
		LogSourceID: req.LogSourceID,
		// #nosec G115 - Log output length is limited and fits in int32
		OutputLength: int32(outputLength),
	})
	if err != nil {
		if !database.IsWorkspaceAgentLogsLimitError(err) {
			httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to upload logs",
				Detail:  err.Error(),
			})
			return
		}
		if workspaceAgent.LogsOverflowed {
			httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{
				Message: "Logs limit exceeded",
				Detail:  err.Error(),
			})
			return
		}
		err := api.Database.UpdateWorkspaceAgentLogOverflowByID(ctx, database.UpdateWorkspaceAgentLogOverflowByIDParams{
			ID:             workspaceAgent.ID,
			LogsOverflowed: true,
		})
		if err != nil {
			// We don't want to return here, because the agent will retry
			// on failure and this isn't a huge deal. The overflow state
			// is just a hint to the user that the logs are incomplete.
			api.Logger.Warn(ctx, "failed to update workspace agent log overflow", slog.Error(err))
		}

		workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)
		if err != nil {
			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
				Message: "Failed to get workspace.",
				Detail:  err.Error(),
			})
			return
		}

		api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{
			Kind:        wspubsub.WorkspaceEventKindAgentLogsOverflow,
			WorkspaceID: workspace.ID,
			AgentID:     &workspaceAgent.ID,
		})

		httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{
			Message: "Logs limit exceeded",
		})
		return
	}

	lowestLogID := logs[0].ID

	// Publish by the lowest log ID inserted so the
	// log stream will fetch everything from that point.
	api.publishWorkspaceAgentLogsUpdate(ctx, workspaceAgent.ID, agentsdk.LogsNotifyMessage{
		CreatedAfter: lowestLogID - 1,
	})

	if workspaceAgent.LogsLength == 0 {
		// If these are the first logs being appended, we publish a UI update
		// to notify the UI that logs are now available.
		workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)
		if err != nil {
			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
				Message: "Failed to get workspace.",
				Detail:  err.Error(),
			})
			return
		}

		api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{
			Kind:        wspubsub.WorkspaceEventKindAgentFirstLogs,
			WorkspaceID: workspace.ID,
			AgentID:     &workspaceAgent.ID,
		})
	}

	httpapi.Write(ctx, rw, http.StatusOK, nil)
}

// @Summary Patch workspace agent app status
// @ID patch-workspace-agent-app-status
// @Security CoderSessionToken
// @Accept json
// @Produce json
// @Tags Agents
// @Param request body agentsdk.PatchAppStatus true "app status"
// @Success 200 {object} codersdk.Response
// @Router /api/v2/workspaceagents/me/app-status [patch]
// @Deprecated Use UpdateAppStatus on the Agent API instead.
func (api *API) patchWorkspaceAgentAppStatus(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	workspaceAgent := httpmw.WorkspaceAgent(r)

	var req agentsdk.PatchAppStatus
	if !httpapi.Read(ctx, rw, r, &req) {
		return
	}

	workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to get workspace.",
			Detail:  err.Error(),
		})
		return
	}

	// This functionality has been moved to the AppsAPI in the agentapi. We keep this HTTP handler around for back
	// compatibility with older agents. We'll translate the request into the protobuf so there is only one primary
	// implementation.
	cachedWs := &agentapi.CachedWorkspaceFields{}
	cachedWs.UpdateValues(workspace)

	appAPI := &agentapi.AppsAPI{
		AgentID:   workspaceAgent.ID,
		Database:  api.Database,
		Log:       api.Logger,
		Workspace: cachedWs,
		AgentFn: func(ctx context.Context) (database.WorkspaceAgent, error) {
			return api.Database.GetWorkspaceAgentByID(ctx, workspaceAgent.ID)
		},
		PublishWorkspaceUpdateFn: func(ctx context.Context, agentID uuid.UUID, kind wspubsub.WorkspaceEventKind) error {
			api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{
				Kind:        kind,
				WorkspaceID: workspace.ID,
				AgentID:     &agentID,
			})
			return nil
		},
		NotificationsEnqueuer: api.NotificationsEnqueuer,
		Clock:                 api.Clock,
	}
	protoReq, err := agentsdk.ProtoFromPatchAppStatus(req)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to parse request.",
			Detail:  err.Error(),
		})
		return
	}
	_, err = appAPI.UpdateAppStatus(r.Context(), protoReq)
	if err != nil {
		sdkErr := new(codersdk.Error)
		if xerrors.As(err, &sdkErr) {
			httpapi.Write(ctx, rw, sdkErr.StatusCode(), sdkErr.Response)
			return
		}
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to update app status.",
			Detail:  err.Error(),
		})
		return
	}
	httpapi.Write(ctx, rw, http.StatusOK, nil)
}

// workspaceAgentLogs returns the logs associated with a workspace agent
//
// @Summary Get logs by workspace agent
// @ID get-logs-by-workspace-agent
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Param before query int false "Before log id"
// @Param after query int false "After log id"
// @Param follow query bool false "Follow log stream"
// @Param no_compression query bool false "Disable compression for WebSocket connection"
// @Param format query string false "Log output format. Accepted: 'json' (default), 'text' (plain text with RFC3339 timestamps and ANSI colors). Not supported with follow=true." Enums(json,text)
// @Success 200 {array} codersdk.WorkspaceAgentLog
// @Router /api/v2/workspaceagents/{workspaceagent}/logs [get]
func (api *API) workspaceAgentLogs(rw http.ResponseWriter, r *http.Request) {
	// This mostly copies how provisioner job logs are streamed!
	var (
		ctx           = r.Context()
		waws          = httpmw.WorkspaceAgentAndWorkspaceParam(r)
		logger        = api.Logger.With(slog.F("workspace_agent_id", waws.WorkspaceAgent.ID))
		follow        = r.URL.Query().Has("follow")
		afterRaw      = r.URL.Query().Get("after")
		noCompression = r.URL.Query().Has("no_compression")
		format        = r.URL.Query().Get("format")
	)

	// Validate format parameter.
	if format == "" {
		format = "json"
	}
	if format != "json" && format != "text" {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Invalid format parameter.",
			Detail:  "Allowed values are \"json\" and \"text\".",
		})
		return
	}

	// Text format is not supported with streaming.
	if format == "text" && follow {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Text format is not supported with follow mode.",
			Detail:  "Use format=json or omit the follow parameter.",
		})
		return
	}

	var after int64
	// Only fetch logs created after the time provided.
	if afterRaw != "" {
		var err error
		after, err = strconv.ParseInt(afterRaw, 10, 64)
		if err != nil || after < 0 {
			httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
				Message: "Query param \"after\" must be an integer greater than or equal to zero.",
				Validations: []codersdk.ValidationError{
					{Field: "after", Detail: "Must be an integer greater than or equal to zero"},
				},
			})
			return
		}
	}

	logs, err := api.Database.GetWorkspaceAgentLogsAfter(ctx, database.GetWorkspaceAgentLogsAfterParams{
		AgentID:      waws.WorkspaceAgent.ID,
		CreatedAfter: after,
	})
	if errors.Is(err, sql.ErrNoRows) {
		err = nil
	}
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching provisioner logs.",
			Detail:  err.Error(),
		})
		return
	}
	if logs == nil {
		logs = []database.WorkspaceAgentLog{}
	}

	if !follow {
		if format == "text" {
			sids, err := api.Database.GetWorkspaceAgentLogSourcesByAgentIDs(ctx, []uuid.UUID{waws.WorkspaceAgent.ID})
			if err != nil {
				httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
					Message: "Internal error fetching workspace agent log sources.",
					Detail:  err.Error(),
				})
				return
			}

			lsids := make(map[uuid.UUID]string, len(sids))
			for _, sid := range sids {
				lsids[sid.ID] = sid.DisplayName
			}
			rw.Header().Set("Content-Type", "text/plain; charset=utf-8")
			rw.WriteHeader(http.StatusOK)
			for _, log := range logs {
				_, _ = rw.Write([]byte(db2sdk.WorkspaceAgentLog(log).Text(waws.WorkspaceAgent.Name, lsids[log.LogSourceID])))
				_, _ = rw.Write([]byte("\n"))
			}
			return
		}
		httpapi.Write(ctx, rw, http.StatusOK, convertWorkspaceAgentLogs(logs))
		return
	}

	api.WebsocketWaitMutex.Lock()
	api.WebsocketWaitGroup.Add(1)
	api.WebsocketWaitMutex.Unlock()
	defer api.WebsocketWaitGroup.Done()

	opts := &websocket.AcceptOptions{}

	// Allow client to request no compression. This is useful for buggy
	// clients or if there's a client/server incompatibility. This is
	// needed with e.g. coder/websocket and Safari (confirmed in 16.5).
	//
	// See:
	// * https://github.com/nhooyr/websocket/issues/218
	// * https://github.com/gobwas/ws/issues/169
	if noCompression {
		opts.CompressionMode = websocket.CompressionDisabled
	}

	conn, err := websocket.Accept(rw, r, opts)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to accept websocket.",
			Detail:  err.Error(),
		})
		return
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	ctx = api.wsWatcher.Watch(ctx, api.Logger, conn)

	encoder := wsjson.NewEncoder[[]codersdk.WorkspaceAgentLog](conn, websocket.MessageText)
	defer encoder.Close(websocket.StatusNormalClosure)

	err = encoder.Encode(convertWorkspaceAgentLogs(logs))
	if err != nil {
		return
	}

	lastSentLogID := after
	if len(logs) > 0 {
		lastSentLogID = logs[len(logs)-1].ID
	}

	workspaceNotifyCh := make(chan struct{}, 1)
	notifyCh := make(chan struct{}, 1)
	// Allow us to immediately check if we missed any logs
	// between initial fetch and subscribe.
	notifyCh <- struct{}{}

	// Subscribe to workspace to detect new builds.
	closeSubscribeWorkspace, err := api.Pubsub.SubscribeWithErr(wspubsub.WorkspaceEventChannel(waws.WorkspaceTable.OwnerID),
		wspubsub.HandleWorkspaceEvent(
			func(_ context.Context, e wspubsub.WorkspaceEvent, err error) {
				if err != nil {
					return
				}
				if e.Kind == wspubsub.WorkspaceEventKindStateChange && e.WorkspaceID == waws.WorkspaceTable.ID {
					select {
					case workspaceNotifyCh <- struct{}{}:
					default:
					}
				}
			}))
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to subscribe to workspace for log streaming.",
			Detail:  err.Error(),
		})
		return
	}
	defer closeSubscribeWorkspace()
	// Subscribe early to prevent missing log events.
	closeSubscribe, err := api.Pubsub.Subscribe(agentsdk.LogsNotifyChannel(waws.WorkspaceAgent.ID), func(_ context.Context, _ []byte) {
		// The message is not important, we're tracking lastSentLogID manually.
		select {
		case notifyCh <- struct{}{}:
		default:
		}
	})
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to subscribe to agent for log streaming.",
			Detail:  err.Error(),
		})
		return
	}
	defer closeSubscribe()

	// Buffer size controls the log prefetch capacity.
	bufferedLogs := make(chan []database.WorkspaceAgentLog, 8)
	// Check at least once per minute in case we didn't receive a pubsub message.
	recheckInterval := time.Minute
	t := time.NewTicker(recheckInterval)
	defer t.Stop()

	// Log the request immediately instead of after it completes.
	if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {
		rl.WriteLog(ctx, http.StatusAccepted)
	}

	go func() {
		defer func() {
			logger.Debug(ctx, "end log streaming loop")
			close(bufferedLogs)
		}()
		logger.Debug(ctx, "start log streaming loop", slog.F("last_sent_log_id", lastSentLogID))

		keepGoing := true
		for keepGoing {
			var (
				debugTriggeredBy     string
				onlyCheckLatestBuild bool
			)
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				debugTriggeredBy = "timer"
			case <-workspaceNotifyCh:
				debugTriggeredBy = "workspace"
				onlyCheckLatestBuild = true
			case <-notifyCh:
				debugTriggeredBy = "log"
				t.Reset(recheckInterval)
			}

			agents, err := api.Database.GetWorkspaceAgentsInLatestBuildByWorkspaceID(ctx, waws.WorkspaceTable.ID)
			if err != nil && !xerrors.Is(err, sql.ErrNoRows) {
				if xerrors.Is(err, context.Canceled) {
					return
				}
				logger.Warn(ctx, "failed to get workspace agents in latest build", slog.Error(err))
				continue
			}
			// If the agent is no longer in the latest build, we can stop after
			// checking once.
			keepGoing = slices.ContainsFunc(agents, func(agent database.WorkspaceAgent) bool { return agent.ID == waws.WorkspaceAgent.ID })

			logger.Debug(
				ctx,
				"checking for new logs",
				slog.F("triggered_by", debugTriggeredBy),
				slog.F("only_check_latest_build", onlyCheckLatestBuild),
				slog.F("keep_going", keepGoing),
				slog.F("last_sent_log_id", lastSentLogID),
				slog.F("workspace_has_agents", len(agents) > 0),
			)

			if onlyCheckLatestBuild && keepGoing {
				continue
			}

			logs, err := api.Database.GetWorkspaceAgentLogsAfter(ctx, database.GetWorkspaceAgentLogsAfterParams{
				AgentID:      waws.WorkspaceAgent.ID,
				CreatedAfter: lastSentLogID,
			})
			if err != nil {
				if xerrors.Is(err, context.Canceled) {
					return
				}
				logger.Warn(ctx, "failed to get workspace agent logs after", slog.Error(err))
				continue
			}
			if len(logs) == 0 {
				// Just keep listening - more logs might come in the future!
				continue
			}

			select {
			case <-ctx.Done():
				return
			case bufferedLogs <- logs:
				lastSentLogID = logs[len(logs)-1].ID
			}
		}
	}()
	defer func() {
		// Ensure that we don't return until the goroutine has exited.
		//nolint:revive // Consume channel to wait until it's closed.
		for range bufferedLogs {
		}
	}()

	for {
		select {
		case <-ctx.Done():
			logger.Debug(ctx, "job logs context canceled")
			return
		case logs, ok := <-bufferedLogs:
			if !ok {
				select {
				case <-ctx.Done():
					logger.Debug(ctx, "job logs context canceled")
				default:
					logger.Debug(ctx, "reached the end of published logs")
				}
				return
			}
			err = encoder.Encode(convertWorkspaceAgentLogs(logs))
			if err != nil {
				return
			}
		}
	}
}

// @Summary Get listening ports for workspace agent
// @ID get-listening-ports-for-workspace-agent
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Success 200 {object} codersdk.WorkspaceAgentListeningPortsResponse
// @Router /api/v2/workspaceagents/{workspaceagent}/listening-ports [get]
func (api *API) workspaceAgentListeningPorts(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

	// If the agent is unreachable, the request will hang. Assume that if we
	// don't get a response after 30s that the agent is unreachable.
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()

	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(), *api.TailnetCoordinator.Load(), waws.WorkspaceAgent, nil, nil, nil, api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	if apiAgent.Status != codersdk.WorkspaceAgentConnected {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),
		})
		return
	}

	agentConn, release, err := api.agentProvider.AgentConn(ctx, waws.WorkspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error dialing workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	defer release()

	portsResponse, err := agentConn.ListeningPorts(ctx)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching listening ports.",
			Detail:  err.Error(),
		})
		return
	}

	// Get a list of ports that are in-use by applications.
	apps, err := api.Database.GetWorkspaceAppsByAgentID(ctx, waws.WorkspaceAgent.ID)
	if xerrors.Is(err, sql.ErrNoRows) {
		apps = []database.WorkspaceApp{}
		err = nil
	}
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching workspace apps.",
			Detail:  err.Error(),
		})
		return
	}
	appPorts := make(map[uint16]struct{}, len(apps))
	for _, app := range apps {
		if !app.Url.Valid || app.Url.String == "" {
			continue
		}
		u, err := url.Parse(app.Url.String)
		if err != nil {
			continue
		}
		port := u.Port()
		if port == "" {
			continue
		}
		portNum, err := strconv.ParseUint(port, 10, 16)
		if err != nil {
			continue
		}
		if portNum < 1 || portNum > 65535 {
			continue
		}
		appPorts[uint16(portNum)] = struct{}{}
	}

	// Filter out ports that are globally blocked, in-use by applications, or
	// common non-HTTP ports such as databases, FTP, SSH, etc.
	filteredPorts := make([]codersdk.WorkspaceAgentListeningPort, 0, len(portsResponse.Ports))
	for _, port := range portsResponse.Ports {
		if port.Port < workspacesdk.AgentMinimumListeningPort {
			continue
		}
		if _, ok := appPorts[port.Port]; ok {
			continue
		}
		if _, ok := workspacesdk.AgentIgnoredListeningPorts[port.Port]; ok {
			continue
		}
		filteredPorts = append(filteredPorts, port)
	}

	portsResponse.Ports = filteredPorts
	httpapi.Write(ctx, rw, http.StatusOK, portsResponse)
}

// @Summary Watch workspace agent for container updates.
// @ID watch-workspace-agent-for-container-updates
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Success 200 {object} codersdk.WorkspaceAgentListContainersResponse
// @Router /api/v2/workspaceagents/{workspaceagent}/containers/watch [get]
func (api *API) watchWorkspaceAgentContainers(rw http.ResponseWriter, r *http.Request) {
	var (
		ctx    = r.Context()
		waws   = httpmw.WorkspaceAgentAndWorkspaceParam(r)
		logger = api.Logger.Named("agent_container_watcher").With(slog.F("agent_id", waws.WorkspaceAgent.ID))
	)

	// If the agent is unreachable, the request will hang. Assume that if we
	// don't get a response after 30s that the agent is unreachable.
	dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second)
	defer dialCancel()
	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(),
		*api.TailnetCoordinator.Load(),
		waws.WorkspaceAgent,
		nil,
		nil,
		nil,
		api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	if apiAgent.Status != codersdk.WorkspaceAgentConnected {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),
		})
		return
	}

	agentConn, release, err := api.agentProvider.AgentConn(dialCtx, waws.WorkspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error dialing workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	defer release()

	containersCh, closer, err := agentConn.WatchContainers(ctx, logger)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error watching agent's containers.",
			Detail:  err.Error(),
		})
		return
	}
	defer closer.Close()

	conn, err := websocket.Accept(rw, r, nil)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to upgrade connection to websocket.",
			Detail:  err.Error(),
		})
		return
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Here we close the websocket for reading, so that the websocket library will handle pings and
	// close frames.
	_ = conn.CloseRead(context.Background())

	ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageText)
	defer wsNetConn.Close()

	ctx = api.wsWatcher.Watch(ctx, logger, conn)

	encoder := json.NewEncoder(wsNetConn)

	for {
		select {
		case <-api.ctx.Done():
			return

		case <-ctx.Done():
			return

		case containers, ok := <-containersCh:
			if !ok {
				return
			}

			if err := encoder.Encode(containers); err != nil {
				api.Logger.Error(ctx, "encode containers", slog.Error(err))
				return
			}
		}
	}
}

// @Summary Get running containers for workspace agent
// @ID get-running-containers-for-workspace-agent
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Param label query string true "Labels" format(key=value)
// @Success 200 {object} codersdk.WorkspaceAgentListContainersResponse
// @Router /api/v2/workspaceagents/{workspaceagent}/containers [get]
func (api *API) workspaceAgentListContainers(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

	labelParam, ok := r.URL.Query()["label"]
	if !ok {
		labelParam = []string{}
	}
	labels := make(map[string]string, len(labelParam)/2)
	for _, label := range labelParam {
		kvs := strings.Split(label, "=")
		if len(kvs) != 2 {
			httpapi.Write(r.Context(), rw, http.StatusBadRequest, codersdk.Response{
				Message: "Invalid label format",
				Detail:  "Labels must be in the format key=value",
			})
			return
		}
		labels[kvs[0]] = kvs[1]
	}

	// If the agent is unreachable, the request will hang. Assume that if we
	// don't get a response after 30s that the agent is unreachable.
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(),
		*api.TailnetCoordinator.Load(),
		waws.WorkspaceAgent,
		nil,
		nil,
		nil,
		api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	if apiAgent.Status != codersdk.WorkspaceAgentConnected {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),
		})
		return
	}

	agentConn, release, err := api.agentProvider.AgentConn(ctx, waws.WorkspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error dialing workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	defer release()

	// Get a list of containers that the agent is able to detect
	cts, err := agentConn.ListContainers(ctx)
	if err != nil {
		if errors.Is(err, context.Canceled) {
			httpapi.Write(ctx, rw, http.StatusRequestTimeout, codersdk.Response{
				Message: "Failed to fetch containers from agent.",
				Detail:  "Request timed out.",
			})
			return
		}
		// If the agent returns a codersdk.Error, we can return that directly.
		if cerr, ok := codersdk.AsError(err); ok {
			httpapi.Write(ctx, rw, cerr.StatusCode(), cerr.Response)
			return
		}
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error fetching containers.",
			Detail:  err.Error(),
		})
		return
	}

	// Filter in-place by labels
	cts.Containers = slices.DeleteFunc(cts.Containers, func(ct codersdk.WorkspaceAgentContainer) bool {
		return !maputil.Subset(labels, ct.Labels)
	})

	httpapi.Write(ctx, rw, http.StatusOK, cts)
}

// @Summary Delete devcontainer for workspace agent
// @ID delete-devcontainer-for-workspace-agent
// @Security CoderSessionToken
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Param devcontainer path string true "Devcontainer ID"
// @Success 204
// @Router /api/v2/workspaceagents/{workspaceagent}/containers/devcontainers/{devcontainer} [delete]
func (api *API) workspaceAgentDeleteDevcontainer(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

	if !api.Authorize(r, policy.ActionUpdate, waws.WorkspaceTable) {
		httpapi.Forbidden(rw)
		return
	}

	devcontainer := chi.URLParam(r, "devcontainer")
	if devcontainer == "" {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Devcontainer ID is required.",
			Validations: []codersdk.ValidationError{
				{Field: "devcontainer", Detail: "Devcontainer ID is required."},
			},
		})
		return
	}

	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(),
		*api.TailnetCoordinator.Load(),
		waws.WorkspaceAgent,
		nil,
		nil,
		nil,
		api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	if apiAgent.Status != codersdk.WorkspaceAgentConnected {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),
		})
		return
	}

	// If the agent is unreachable, the request will hang. Assume that if we
	// don't get a response after 30s that the agent is unreachable.
	dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second)
	defer dialCancel()
	agentConn, release, err := api.agentProvider.AgentConn(dialCtx, waws.WorkspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error dialing workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	defer release()

	if err = agentConn.DeleteDevcontainer(ctx, devcontainer); err != nil {
		if errors.Is(err, context.Canceled) {
			httpapi.Write(ctx, rw, http.StatusRequestTimeout, codersdk.Response{
				Message: "Failed to delete devcontainer from agent.",
				Detail:  "Request timed out.",
			})
			return
		}
		// If the agent returns a codersdk.Error, we can return that directly.
		if cerr, ok := codersdk.AsError(err); ok {
			httpapi.Write(ctx, rw, cerr.StatusCode(), cerr.Response)
			return
		}
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error deleting devcontainer.",
			Detail:  err.Error(),
		})
		return
	}

	httpapi.Write(ctx, rw, http.StatusNoContent, nil)
}

// @Summary Recreate devcontainer for workspace agent
// @ID recreate-devcontainer-for-workspace-agent
// @Security CoderSessionToken
// @Tags Agents
// @Produce json
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Param devcontainer path string true "Devcontainer ID"
// @Success 202 {object} codersdk.Response
// @Router /api/v2/workspaceagents/{workspaceagent}/containers/devcontainers/{devcontainer}/recreate [post]
func (api *API) workspaceAgentRecreateDevcontainer(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

	if !api.Authorize(r, policy.ActionUpdate, waws.WorkspaceTable) {
		httpapi.Forbidden(rw)
		return
	}

	devcontainer := chi.URLParam(r, "devcontainer")
	if devcontainer == "" {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Devcontainer ID is required.",
			Validations: []codersdk.ValidationError{
				{Field: "devcontainer", Detail: "Devcontainer ID is required."},
			},
		})
		return
	}

	apiAgent, err := db2sdk.WorkspaceAgent(
		api.DERPMap(),
		*api.TailnetCoordinator.Load(),
		waws.WorkspaceAgent,
		nil,
		nil,
		nil,
		api.AgentInactiveDisconnectTimeout,
		api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),
	)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error reading workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	if apiAgent.Status != codersdk.WorkspaceAgentConnected {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),
		})
		return
	}

	// If the agent is unreachable, the request will hang. Assume that if we
	// don't get a response after 30s that the agent is unreachable.
	dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second)
	defer dialCancel()
	agentConn, release, err := api.agentProvider.AgentConn(dialCtx, waws.WorkspaceAgent.ID)
	if err != nil {
		httpapi.Write(dialCtx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error dialing workspace agent.",
			Detail:  err.Error(),
		})
		return
	}
	defer release()

	m, err := agentConn.RecreateDevcontainer(ctx, devcontainer)
	if err != nil {
		if errors.Is(err, context.Canceled) {
			httpapi.Write(ctx, rw, http.StatusRequestTimeout, codersdk.Response{
				Message: "Failed to recreate devcontainer from agent.",
				Detail:  "Request timed out.",
			})
			return
		}
		// If the agent returns a codersdk.Error, we can return that directly.
		if cerr, ok := codersdk.AsError(err); ok {
			httpapi.Write(ctx, rw, cerr.StatusCode(), cerr.Response)
			return
		}
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error recreating devcontainer.",
			Detail:  err.Error(),
		})
		return
	}

	httpapi.Write(ctx, rw, http.StatusAccepted, m)
}

// @Summary Get connection info for workspace agent
// @ID get-connection-info-for-workspace-agent
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Success 200 {object} workspacesdk.AgentConnectionInfo
// @Router /api/v2/workspaceagents/{workspaceagent}/connection [get]
func (api *API) workspaceAgentConnection(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	httpapi.Write(ctx, rw, http.StatusOK, workspacesdk.AgentConnectionInfo{
		DERPMap:                  api.DERPMap(),
		DERPForceWebSockets:      api.DeploymentValues.DERP.Config.ForceWebSockets.Value(),
		DisableDirectConnections: api.DeploymentValues.DERP.Config.BlockDirect.Value(),
		HostnameSuffix:           api.DeploymentValues.WorkspaceHostnameSuffix.Value(),
	})
}

// workspaceAgentConnectionGeneric is the same as workspaceAgentConnection but
// without the workspaceagent path parameter.
//
// @Summary Get connection info for workspace agent generic
// @ID get-connection-info-for-workspace-agent-generic
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Success 200 {object} workspacesdk.AgentConnectionInfo
// @Router /api/v2/workspaceagents/connection [get]
// @x-apidocgen {"skip": true}
func (api *API) workspaceAgentConnectionGeneric(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	httpapi.Write(ctx, rw, http.StatusOK, workspacesdk.AgentConnectionInfo{
		DERPMap:                  api.DERPMap(),
		DERPForceWebSockets:      api.DeploymentValues.DERP.Config.ForceWebSockets.Value(),
		DisableDirectConnections: api.DeploymentValues.DERP.Config.BlockDirect.Value(),
		HostnameSuffix:           api.DeploymentValues.WorkspaceHostnameSuffix.Value(),
	})
}

// @Summary Get DERP map updates
// @ID get-derp-map-updates
// @Security CoderSessionToken
// @Tags Agents
// @Success 101
// @Router /api/v2/derp-map [get]
func (api *API) derpMapUpdates(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	api.WebsocketWaitMutex.Lock()
	api.WebsocketWaitGroup.Add(1)
	api.WebsocketWaitMutex.Unlock()
	defer api.WebsocketWaitGroup.Done()

	ws, err := websocket.Accept(rw, r, nil)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to accept websocket.",
			Detail:  err.Error(),
		})
		return
	}
	encoder := wsjson.NewEncoder[*tailcfg.DERPMap](ws, websocket.MessageBinary)
	defer encoder.Close(websocket.StatusGoingAway)

	// Log the request immediately instead of after it completes.
	if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {
		rl.WriteLog(ctx, http.StatusAccepted)
	}

	go func(ctx context.Context) {
		// TODO(mafredri): Is this too frequent? Use separate ping disconnect timeout?
		t := time.NewTicker(api.AgentConnectionUpdateFrequency)
		defer t.Stop()

		for {
			select {
			case <-t.C:
			case <-ctx.Done():
				return
			}

			ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
			err := ws.Ping(ctx)
			cancel()
			if err != nil {
				_ = ws.Close(websocket.StatusGoingAway, "ping failed")
				return
			}
		}
	}(ctx)

	ticker := time.NewTicker(api.Options.DERPMapUpdateFrequency)
	defer ticker.Stop()

	var lastDERPMap *tailcfg.DERPMap
	for {
		derpMap := api.DERPMap()
		if lastDERPMap == nil || !tailnet.CompareDERPMaps(lastDERPMap, derpMap) {
			err := encoder.Encode(derpMap)
			if err != nil {
				return
			}
			lastDERPMap = derpMap
		}

		ticker.Reset(api.Options.DERPMapUpdateFrequency)
		select {
		case <-ctx.Done():
			return
		case <-api.ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// workspaceAgentClientCoordinate accepts a WebSocket that reads node network updates.
// After accept a PubSub starts listening for new connection node updates
// which are written to the WebSocket.
//
// @Summary Coordinate workspace agent
// @ID coordinate-workspace-agent
// @Security CoderSessionToken
// @Tags Agents
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Success 101
// @Router /api/v2/workspaceagents/{workspaceagent}/coordinate [get]
func (api *API) workspaceAgentClientCoordinate(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	// Ensure the database is reachable before proceeding.
	_, err := api.Database.Ping(ctx)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: codersdk.DatabaseNotReachable,
			Detail:  err.Error(),
		})
		return
	}

	// This route accepts user API key auth and workspace proxy auth. The moon actor has
	// full permissions so should be able to pass this authz check.
	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)
	if !api.Authorize(r, policy.ActionSSH, waws) {
		httpapi.ResourceNotFound(rw)
		return
	}

	// This is used by Enterprise code to control the functionality of this route.
	// Namely, disabling the route using `CODER_BROWSER_ONLY`.
	override := api.WorkspaceClientCoordinateOverride.Load()
	if override != nil {
		overrideFunc := *override
		if overrideFunc != nil && overrideFunc(rw) {
			return
		}
	}

	version := "1.0"
	qv := r.URL.Query().Get("version")
	if qv != "" {
		version = qv
	}
	if err := proto.CurrentVersion.Validate(version); err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Unknown or unsupported API version",
			Validations: []codersdk.ValidationError{
				{Field: "version", Detail: err.Error()},
			},
		})
		return
	}

	peerID, err := api.handleResumeToken(ctx, rw, r)
	if err != nil {
		// handleResumeToken has already written the response.
		return
	}

	api.WebsocketWaitMutex.Lock()
	api.WebsocketWaitGroup.Add(1)
	api.WebsocketWaitMutex.Unlock()
	defer api.WebsocketWaitGroup.Done()

	conn, err := websocket.Accept(rw, r, nil)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to accept websocket.",
			Detail:  err.Error(),
		})
		return
	}

	api.logTunnelConnection(ctx, r, waws)

	ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageBinary)
	defer wsNetConn.Close()

	ctx = api.wsWatcher.Watch(ctx, api.Logger, conn)

	defer conn.Close(websocket.StatusNormalClosure, "")
	err = api.TailnetClientService.ServeClient(ctx, version, wsNetConn, tailnet.StreamID{
		Name: "client",
		ID:   peerID,
		Auth: tailnet.ClientCoordinateeAuth{
			AgentID: waws.WorkspaceAgent.ID,
		},
	})
	if err != nil && !xerrors.Is(err, io.EOF) && !xerrors.Is(err, context.Canceled) {
		_ = conn.Close(websocket.StatusInternalError, err.Error())
		return
	}
}

// logTunnelConnection records a connection log entry attributing a
// tunnel to the authenticated user who opened it. Agent-reported rows
// cannot identify the user (see coderd/agentapi/connectionlog.go), and
// workspace-proxy-authenticated requests carry no API key and are
// skipped.
func (api *API) logTunnelConnection(ctx context.Context, r *http.Request, waws database.GetWorkspaceAgentAndWorkspaceByIDRow) {
	apiKey, ok := httpmw.APIKeyOptional(r)
	if !ok {
		return
	}
	// Bounded so log backpressure cannot stall tunnel establishment.
	writeCtx, writeCancel := context.WithTimeout(ctx, 3*time.Second)
	defer writeCancel()
	userAgent := r.UserAgent()
	now := dbtime.Now()

	// Clients re-dial automatically, so dedupe reconnects through the
	// same audit session mechanism as workspace apps, keyed on
	// (agent, user, IP, user agent). Status 101 and the empty slug
	// keep tunnel sessions from ever colliding with app or
	// port-forwarding sessions.
	staleInterval := api.Options.WorkspaceAppAuditSessionTimeout
	if staleInterval == 0 {
		staleInterval = time.Hour
	}
	// nolint:gocritic // System context is needed to write audit sessions.
	newSession, err := api.Database.UpsertWorkspaceAppAuditSession(dbauthz.AsSystemRestricted(writeCtx), database.UpsertWorkspaceAppAuditSessionParams{
		// Config.
		StaleIntervalMS: staleInterval.Milliseconds(),

		// Data.
		ID:         uuid.New(),
		AgentID:    waws.WorkspaceAgent.ID,
		AppID:      uuid.Nil, // Tunnels are not associated with an app.
		UserID:     apiKey.UserID,
		Ip:         r.RemoteAddr,
		UserAgent:  userAgent,
		SlugOrPort: "",
		StatusCode: http.StatusSwitchingProtocols,
		StartedAt:  now,
		UpdatedAt:  now,
	})
	if err != nil {
		// Skip logging rather than risk spamming the connection log.
		api.Logger.Error(ctx, "upsert tunnel audit session",
			slog.F("workspace_id", waws.WorkspaceTable.ID),
			slog.F("user_id", apiKey.UserID),
			slog.Error(err),
		)
		return
	}
	if !newSession {
		// Reconnection of an already-logged session.
		return
	}

	connLogger := *api.ConnectionLogger.Load()
	err = connLogger.Upsert(writeCtx, database.UpsertConnectionLogParams{
		ID:               uuid.New(),
		Time:             now,
		OrganizationID:   waws.WorkspaceTable.OrganizationID,
		WorkspaceOwnerID: waws.WorkspaceTable.OwnerID,
		WorkspaceID:      waws.WorkspaceTable.ID,
		WorkspaceName:    waws.WorkspaceTable.Name,
		AgentName:        waws.WorkspaceAgent.Name,
		Type:             database.ConnectionTypeTunnel,
		IP:               database.ParseIP(r.RemoteAddr),
		Code: sql.NullInt32{
			Int32: http.StatusSwitchingProtocols,
			Valid: true,
		},
		UserAgent: sql.NullString{String: userAgent, Valid: userAgent != ""},
		UserID:    uuid.NullUUID{UUID: apiKey.UserID, Valid: true},
		// Left unset so each session gets its own row; reusing peerID
		// would make resume_token reconnects upsert into a stale row.
		ConnectionID:     uuid.NullUUID{},
		ConnectionStatus: database.ConnectionStatusConnected,
		// N/A
		SlugOrPort:       sql.NullString{},
		DisconnectReason: sql.NullString{},
	})
	if err != nil {
		api.Logger.Error(ctx, "upsert tunnel connection log",
			slog.F("workspace_id", waws.WorkspaceTable.ID),
			slog.F("user_id", apiKey.UserID),
			slog.Error(err),
		)
	}
}

// handleResumeToken accepts a resume_token query parameter to use the same peer ID
func (api *API) handleResumeToken(ctx context.Context, rw http.ResponseWriter, r *http.Request) (peerID uuid.UUID, err error) {
	peerID = uuid.New()
	resumeToken := r.URL.Query().Get("resume_token")
	if resumeToken != "" {
		peerID, err = api.Options.CoordinatorResumeTokenProvider.VerifyResumeToken(ctx, resumeToken)
		// If the token is missing the key ID, it's probably an old token in which
		// case we just want to generate a new peer ID.
		switch {
		case xerrors.Is(err, jwtutils.ErrMissingKeyID):
			peerID = uuid.New()
			err = nil
		case err != nil:
			httpapi.Write(ctx, rw, http.StatusUnauthorized, codersdk.Response{
				Message: workspacesdk.CoordinateAPIInvalidResumeToken,
				Detail:  err.Error(),
				Validations: []codersdk.ValidationError{
					{Field: "resume_token", Detail: workspacesdk.CoordinateAPIInvalidResumeToken},
				},
			})
			return peerID, err
		default:
			api.Logger.Debug(ctx, "accepted coordinate resume token for peer",
				slog.F("peer_id", peerID.String()))
		}
	}
	return peerID, err
}

// @Summary Post workspace agent log source
// @ID post-workspace-agent-log-source
// @Security CoderSessionToken
// @Accept json
// @Produce json
// @Tags Agents
// @Param request body agentsdk.PostLogSourceRequest true "Log source request"
// @Success 200 {object} codersdk.WorkspaceAgentLogSource
// @Router /api/v2/workspaceagents/me/log-source [post]
func (api *API) workspaceAgentPostLogSource(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	var req agentsdk.PostLogSourceRequest
	if !httpapi.Read(ctx, rw, r, &req) {
		return
	}

	workspaceAgent := httpmw.WorkspaceAgent(r)

	sources, err := api.Database.InsertWorkspaceAgentLogSources(ctx, database.InsertWorkspaceAgentLogSourcesParams{
		WorkspaceAgentID: workspaceAgent.ID,
		CreatedAt:        dbtime.Now(),
		ID:               []uuid.UUID{req.ID},
		DisplayName:      []string{req.DisplayName},
		Icon:             []string{req.Icon},
	})
	if err != nil {
		if database.IsUniqueViolation(err, "workspace_agent_log_sources_pkey") {
			httpapi.Write(ctx, rw, http.StatusCreated, codersdk.WorkspaceAgentLogSource{
				WorkspaceAgentID: workspaceAgent.ID,
				CreatedAt:        dbtime.Now(),
				ID:               req.ID,
				DisplayName:      req.DisplayName,
				Icon:             req.Icon,
			})
			return
		}
		httpapi.InternalServerError(rw, err)
		return
	}

	if len(sources) != 1 {
		httpapi.InternalServerError(rw, xerrors.Errorf("database should've returned 1 row, got %d", len(sources)))
		return
	}

	apiSource := convertLogSources(sources)[0]

	httpapi.Write(ctx, rw, http.StatusCreated, apiSource)
}

// @Summary Get workspace agent reinitialization
// @ID get-workspace-agent-reinitialization
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param wait query bool false "Opt in to durable reinit checks"
// @Success 200 {object} agentsdk.ReinitializationEvent
// @Failure 409 {object} codersdk.Response
// @Router /api/v2/workspaceagents/me/reinit [get]
func (api *API) workspaceAgentReinit(rw http.ResponseWriter, r *http.Request) {
	// Allow us to interrupt watch via cancel.
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	r = r.WithContext(ctx) // Rewire context for SSE cancellation.

	workspaceAgent := httpmw.WorkspaceAgent(r)
	log := api.Logger.Named("workspace_agent_reinit_watcher").With(
		slog.F("workspace_agent_id", workspaceAgent.ID),
	)

	workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)
	if err != nil {
		log.Error(ctx, "failed to retrieve workspace from agent token", slog.Error(err))
		httpapi.InternalServerError(rw, xerrors.New("failed to determine workspace from agent token"))
		return
	}
	log = log.With(slog.F("workspace_id", workspace.ID))

	log.Info(ctx, "agent waiting for reinit instruction")

	// Subscribe to claim events BEFORE any durable checks to avoid a
	// TOCTOU race: without this, a claim could fire between the
	// IsPrebuild() check and the subscribe call, and we'd miss the
	// pubsub event entirely. By subscribing first, any event that
	// fires during the checks below is buffered in the channel.
	pubsubCh, cancelSub, err := prebuilds.NewPubsubWorkspaceClaimListener(api.Pubsub, log).ListenForWorkspaceClaims(ctx, workspace.ID)
	if err != nil {
		log.Error(ctx, "subscribe to prebuild claimed channel", slog.Error(err))
		httpapi.InternalServerError(rw, xerrors.New("failed to subscribe to prebuild claimed channel"))
		return
	}
	defer cancelSub()

	reinitEvents := pubsubCh

	// Only perform the durable claim check when the agent opts in via
	// the "wait" query parameter. Older agents don't send the
	// "wait" query parameter and lack the duplicate-reinit guard, so
	// they would enter an infinite reinit loop if we pre-seeded the
	// channel on every connection.
	waitParam, _ := strconv.ParseBool(r.URL.Query().Get("wait"))
	if waitParam && !workspace.IsPrebuild() {
		firstBuild, err := api.Database.GetWorkspaceBuildByWorkspaceIDAndBuildNumber(ctx,
			database.GetWorkspaceBuildByWorkspaceIDAndBuildNumberParams{
				WorkspaceID: workspace.ID,
				BuildNumber: 1,
			})
		if err != nil {
			log.Error(ctx, "failed to get first workspace build", slog.Error(err))
			httpapi.InternalServerError(rw, xerrors.New("failed to get first workspace build"))
			return
		}
		if firstBuild.InitiatorID != database.PrebuildsSystemUserID {
			// Not a claimed prebuild — this is a regular workspace.
			// Return 409 so the agent stops reconnecting to this
			// endpoint.
			httpapi.Write(ctx, rw, http.StatusConflict, codersdk.Response{
				Message: "Workspace is not a prebuilt workspace waiting to be claimed.",
				Detail:  "This endpoint is only for agents running in prebuilt workspaces.",
			})
			return
		}

		// This workspace was a prebuild that got claimed. The seeded
		// reinit below recovers a claim event that was missed while
		// the agent's /reinit connection was down. It only applies
		// while the latest build is the claim build itself, which the
		// build's provisioner job input records, mirroring the check
		// the provisioner server uses when publishing the claim
		// event.
		latestBuild, err := api.Database.GetLatestWorkspaceBuildByWorkspaceID(ctx, workspace.ID)
		if err != nil {
			log.Error(ctx, "failed to get latest workspace build", slog.Error(err))
			httpapi.InternalServerError(rw, xerrors.New("failed to get latest workspace build"))
			return
		}
		job, err := api.Database.GetProvisionerJobByID(ctx, latestBuild.JobID)
		if err != nil {
			log.Error(ctx, "failed to get provisioner job", slog.Error(err))
			httpapi.InternalServerError(rw, xerrors.New("failed to get provisioner job"))
			return
		}
		var jobInput provisionerdserver.WorkspaceProvisionJob
		if err := json.Unmarshal(job.Input, &jobInput); err != nil {
			log.Error(ctx, "failed to unmarshal provisioner job input", slog.Error(err))
			httpapi.InternalServerError(rw, xerrors.New("failed to unmarshal provisioner job input"))
			return
		}

		switch {
		case jobInput.PrebuiltWorkspaceBuildStage.IsPrebuiltWorkspaceClaim():
			if job.CompletedAt.Valid && !job.Error.Valid {
				// Claim build succeeded: cancel the pubsub
				// subscription (no longer needed) and swap in a
				// pre-seeded channel so the transmitter delivers
				// exactly one reinit event.
				cancelSub()
				seeded := make(chan agentsdk.ReinitializationEvent, 1)
				seeded <- agentsdk.ReinitializationEvent{
					WorkspaceID: workspace.ID,
					Reason:      agentsdk.ReinitializeReasonPrebuildClaimed,
					OwnerID:     workspace.OwnerID,
				}
				reinitEvents = seeded
			} else if job.CompletedAt.Valid && job.Error.Valid {
				// Claim build failed permanently. Return 409 so the
				// agent treats this as terminal and stops retrying
				// (WaitForReinitLoop exits on any 409).
				cancelSub()
				log.Warn(ctx, "claim build failed",
					slog.F("job_id", job.ID),
					slog.F("error", job.Error.String))
				httpapi.Write(ctx, rw, http.StatusConflict, codersdk.Response{
					Message: "Claim build failed permanently.",
					Detail:  job.Error.String,
				})
				return
			}
			// Claim build still in progress: proceed to the
			// transmitter below. The pubsub subscription (set up
			// above) will deliver the event when the build completes
			// successfully. Note: FailJob does not publish a claim
			// event, so a failed in-progress build will leave the
			// agent blocking here until it disconnects and
			// reconnects (at which point the durable check above
			// handles it).
		case latestBuild.InitiatorID == database.PrebuildsSystemUserID:
			// The workspace owner has changed but the claim build has
			// not been created yet. Proceed to the transmitter below;
			// the pubsub subscription set up above delivers the claim
			// event once the claim build completes.
		default:
			// The latest build is a user-initiated build other than
			// the claim build, so the claim has already been handled.
			// Re-sending the reinit event here would needlessly
			// restart the agent of a long-claimed workspace on every
			// reconnection. Return 409 so the agent stops polling,
			// the same as a regular workspace.
			log.Debug(ctx, "prebuild claim already handled, stopping reinit polling",
				slog.F("job_id", job.ID),
				slog.F("latest_build_id", latestBuild.ID),
				slog.F("latest_build_number", latestBuild.BuildNumber))
			cancelSub()
			httpapi.Write(ctx, rw, http.StatusConflict, codersdk.Response{
				Message: "Workspace is not a prebuilt workspace waiting to be claimed.",
				Detail:  "The prebuild claim for this workspace has already been handled by an earlier build.",
			})
			return
		}
	}

	transmitter := agentsdk.NewSSEAgentReinitTransmitter(log, rw, r)

	err = transmitter.Transmit(ctx, reinitEvents)
	switch {
	case errors.Is(err, agentsdk.ErrTransmissionSourceClosed):
		log.Info(ctx, "agent reinitialization subscription closed", slog.F("workspace_agent_id", workspaceAgent.ID))
	case errors.Is(err, agentsdk.ErrTransmissionTargetClosed):
		log.Info(ctx, "agent connection closed", slog.F("workspace_agent_id", workspaceAgent.ID))
	case errors.Is(err, context.Canceled):
		log.Info(ctx, "agent reinitialization", slog.Error(err))
	case err != nil:
		log.Error(ctx, "failed to stream agent reinit events", slog.Error(err))
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error streaming agent reinitialization events.",
			Detail:  err.Error(),
		})
	}
}

// convertProvisionedApps converts applications that are in the middle of provisioning process.
// It means that they may not have an agent or workspace assigned (dry-run job).
func convertProvisionedApps(dbApps []database.WorkspaceApp) []codersdk.WorkspaceApp {
	return db2sdk.Apps(dbApps, []database.WorkspaceAppStatus{}, database.WorkspaceAgent{}, "", database.WorkspaceTable{})
}

func convertLogSources(dbLogSources []database.WorkspaceAgentLogSource) []codersdk.WorkspaceAgentLogSource {
	logSources := make([]codersdk.WorkspaceAgentLogSource, 0)
	for _, dbLogSource := range dbLogSources {
		logSources = append(logSources, codersdk.WorkspaceAgentLogSource{
			ID:               dbLogSource.ID,
			DisplayName:      dbLogSource.DisplayName,
			WorkspaceAgentID: dbLogSource.WorkspaceAgentID,
			CreatedAt:        dbLogSource.CreatedAt,
			Icon:             dbLogSource.Icon,
		})
	}
	return logSources
}

func convertScripts(dbScripts []database.GetWorkspaceAgentScriptsByAgentIDsRow) []codersdk.WorkspaceAgentScript {
	scripts := make([]codersdk.WorkspaceAgentScript, 0)
	for _, dbScript := range dbScripts {
		scripts = append(scripts, db2sdk.WorkspaceAgentScript(dbScript))
	}
	return scripts
}

// @Summary Watch for workspace agent metadata updates
// @ID watch-for-workspace-agent-metadata-updates
// @Security CoderSessionToken
// @Tags Agents
// @Success 200 "Success"
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Router /api/v2/workspaceagents/{workspaceagent}/watch-metadata [get]
// @x-apidocgen {"skip": true}
// @Deprecated Use /workspaceagents/{workspaceagent}/watch-metadata-ws instead
func (api *API) watchWorkspaceAgentMetadataSSE(rw http.ResponseWriter, r *http.Request) {
	api.watchWorkspaceAgentMetadata(rw, r, httpapi.ServerSentEventSender)
}

// @Summary Watch for workspace agent metadata updates via WebSockets
// @ID watch-for-workspace-agent-metadata-updates-via-websockets
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Success 200 {object} codersdk.ServerSentEvent
// @Param workspaceagent path string true "Workspace agent ID" format(uuid)
// @Router /api/v2/workspaceagents/{workspaceagent}/watch-metadata-ws [get]
// @x-apidocgen {"skip": true}
func (api *API) watchWorkspaceAgentMetadataWS(rw http.ResponseWriter, r *http.Request) {
	api.watchWorkspaceAgentMetadata(rw, r, httpapi.OneWayWebSocketEventSender(api.Logger, api.wsWatcher))
}

func (api *API) watchWorkspaceAgentMetadata(
	rw http.ResponseWriter,
	r *http.Request,
	connect httpapi.EventSender,
) {
	// Allow us to interrupt watch via cancel.
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	r = r.WithContext(ctx) // Rewire context for SSE cancellation.

	waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)
	agentIDEncoded := make([]byte, metadatabatcher.UUIDBase64Size)
	err := metadatabatcher.EncodeAgentID(waws.WorkspaceAgent.ID, agentIDEncoded)
	if err != nil {
		httpapi.InternalServerError(rw, err)
		return
	}
	log := api.Logger.Named("workspace_metadata_watcher").With(
		slog.F("workspace_agent_id", waws.WorkspaceAgent.ID),
		slog.F("workspace_id", waws.WorkspaceTable.ID),
	)

	// Send metadata on updates, we must ensure subscription before sending
	// initial metadata to guarantee that events in-between are not missed.
	// The channel carries no data - it's just a signal to fetch all metadata.
	update := make(chan struct{}, 1)

	// Subscribe to the global batched metadata channel.
	// The batcher publishes only to this channel to achieve O(1) NOTIFY scaling.
	cancelBatchSub, err := api.Pubsub.Subscribe(metadatabatcher.MetadataBatchPubsubChannel, func(_ context.Context, byt []byte) {
		if ctx.Err() != nil {
			return
		}

		if len(byt)%metadatabatcher.UUIDBase64Size != 0 {
			log.Error(ctx, "invalid batched pubsub message, pubsub message length was not a multiple of encoded agent UUID length", slog.Error(err))
			return
		}

		// Compare each encoded agentID to our encoded agent ID.
		for i := 0; i < len(byt); i += metadatabatcher.UUIDBase64Size {
			if !bytes.Equal(byt[i:i+metadatabatcher.UUIDBase64Size], agentIDEncoded) {
				continue
			}

			log.Debug(ctx, "received metadata update from batch channel",
				slog.F("agent_id", waws.WorkspaceAgent.ID),
				slog.F("batch_size", len(byt)/metadatabatcher.UUIDBase64Size),
			)

			// Signal to re-fetch all metadata for this agent.
			// Batch notifications don't include which keys changed, so we
			// always fetch all keys for this agent.
			// Attempt to read from the channel first so that we do not block on the write.
			select {
			case <-update:
			default:
			}
			update <- struct{}{}
			break
		}
	})
	if err != nil {
		httpapi.InternalServerError(rw, err)
		return
	}
	defer cancelBatchSub()

	// We always use the original Request context because it contains
	// the RBAC actor.
	initialMD, err := api.Database.GetWorkspaceAgentMetadata(ctx, database.GetWorkspaceAgentMetadataParams{
		WorkspaceAgentID: waws.WorkspaceAgent.ID,
		Keys:             nil,
	})
	if err != nil {
		// If we can't successfully pull the initial metadata, pubsub
		// updates will be no-op so we may as well terminate the
		// connection early.
		httpapi.InternalServerError(rw, err)
		return
	}

	log.Debug(ctx, "got initial metadata", slog.F("num", len(initialMD)))

	metadataMap := make(map[string]database.WorkspaceAgentMetadatum, len(initialMD))
	for _, datum := range initialMD {
		metadataMap[datum.Key] = datum
	}
	//nolint:ineffassign // Release memory.
	initialMD = nil

	sendEvent, senderClosed, err := connect(rw, r)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error setting up server-sent events.",
			Detail:  err.Error(),
		})
		return
	}
	// Prevent handler from returning until the sender is closed.
	defer func() {
		cancel()
		<-senderClosed
	}()
	// Synchronize cancellation from SSE -> context, this lets us simplify the
	// cancellation logic.
	go func() {
		select {
		case <-ctx.Done():
		case <-senderClosed:
			cancel()
		}
	}()

	var lastSend time.Time
	sendMetadata := func() {
		lastSend = time.Now()
		values := maps.Values(metadataMap)

		log.Debug(ctx, "sending metadata", slog.F("num", len(values)))

		_ = sendEvent(codersdk.ServerSentEvent{
			Type: codersdk.ServerSentEventTypeData,
			Data: convertWorkspaceAgentMetadata(values),
		})
	}

	// We send updates exactly every second.
	const sendInterval = time.Second * 1
	sendTicker := time.NewTicker(sendInterval)
	defer sendTicker.Stop()

	// Log the request immediately instead of after it completes.
	if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {
		rl.WriteLog(ctx, http.StatusAccepted)
	}

	// Send initial metadata.
	sendMetadata()

	// Fetch updated metadata keys as they come in.
	fetchedMetadata := make(chan []database.WorkspaceAgentMetadatum)
	go func() {
		defer close(fetchedMetadata)
		defer cancel()

		for {
			select {
			case <-ctx.Done():
				return
			case <-update:
				// Batch notification received - fetch all metadata for this agent.
				md, err := api.Database.GetWorkspaceAgentMetadata(ctx, database.GetWorkspaceAgentMetadataParams{
					WorkspaceAgentID: waws.WorkspaceAgent.ID,
					Keys:             nil, // nil means fetch all keys
				})
				if err != nil {
					if !database.IsQueryCanceledError(err) {
						log.Error(ctx, "failed to get metadata", slog.Error(err))
						_ = sendEvent(codersdk.ServerSentEvent{
							Type: codersdk.ServerSentEventTypeError,
							Data: codersdk.Response{
								Message: "Failed to get metadata.",
								Detail:  err.Error(),
							},
						})
					}
					return
				}
				select {
				case <-ctx.Done():
					return
				// We want to block here to avoid constantly pinging the
				// database when the metadata isn't being processed.
				case fetchedMetadata <- md:
					log.Debug(ctx, "fetched all metadata after batch update", slog.F("num", len(md)))
				}
			}
		}
	}()
	defer func() {
		<-fetchedMetadata
	}()

	pendingChanges := true
	for {
		select {
		case <-ctx.Done():
			return
		case md, ok := <-fetchedMetadata:
			if !ok {
				return
			}
			for _, datum := range md {
				metadataMap[datum.Key] = datum
			}
			pendingChanges = true
			continue
		case <-sendTicker.C:
			// We send an update even if there's no change every 5 seconds
			// to ensure that the frontend always has an accurate "Result.Age".
			if !pendingChanges && time.Since(lastSend) < 5*time.Second {
				continue
			}
			pendingChanges = false
		}

		sendMetadata()
	}
}

func convertWorkspaceAgentMetadata(db []database.WorkspaceAgentMetadatum) []codersdk.WorkspaceAgentMetadata {
	// Sort the input database slice by DisplayOrder and then by Key before processing
	sort.Slice(db, func(i, j int) bool {
		if db[i].DisplayOrder == db[j].DisplayOrder {
			return db[i].Key < db[j].Key
		}
		return db[i].DisplayOrder < db[j].DisplayOrder
	})

	// An empty array is easier for clients to handle than a null.
	result := make([]codersdk.WorkspaceAgentMetadata, len(db))
	for i, datum := range db {
		result[i] = codersdk.WorkspaceAgentMetadata{
			Result: codersdk.WorkspaceAgentMetadataResult{
				Value:       datum.Value,
				Error:       datum.Error,
				CollectedAt: datum.CollectedAt.UTC(),
				Age:         int64(time.Since(datum.CollectedAt).Seconds()),
			},
			Description: codersdk.WorkspaceAgentMetadataDescription{
				DisplayName: datum.DisplayName,
				Key:         datum.Key,
				Script:      datum.Script,
				Interval:    datum.Interval,
				Timeout:     datum.Timeout,
			},
		}
	}
	return result
}

// workspaceAgentsExternalAuth returns an access token for a given URL
// or finds a provider by ID.
//
// @Summary Get workspace agent external auth
// @ID get-workspace-agent-external-auth
// @Security CoderSessionToken
// @Produce json
// @Tags Agents
// @Param match query string true "Match"
// @Param id query string true "Provider ID"
// @Param listen query bool false "Wait for a new token to be issued"
// @Success 200 {object} agentsdk.ExternalAuthResponse
// @Router /api/v2/workspaceagents/me/external-auth [get]
func (api *API) workspaceAgentsExternalAuth(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	query := r.URL.Query()
	gitRef := chatGitRef{
		Branch:       strings.TrimSpace(query.Get("git_branch")),
		RemoteOrigin: strings.TrimSpace(query.Get("git_remote_origin")),
	}
	if raw := strings.TrimSpace(query.Get("chat_id")); raw != "" {
		if parsed, err := uuid.Parse(raw); err == nil {
			gitRef.ChatID = parsed
		}
	}
	// Either match or configID must be provided!
	match := query.Get("match")
	if match == "" {
		// Support legacy agents!
		match = query.Get("url")
	}
	id := query.Get("id")
	if match == "" && id == "" {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "'url' or 'id' must be provided!",
		})
		return
	}
	if match != "" && id != "" {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "'url' and 'id' cannot be provided together!",
		})
		return
	}

	// listen determines if the request will wait for a
	// new token to be issued!
	listen := query.Has("listen")

	// Resolve the calling agent's own workspace/build before selecting a
	// provider below, so the match-only (GIT_ASKPASS) path can be scoped to
	// this workspace's own template-declared providers instead of scanning
	// every provider configured on the deployment.
	workspaceAgent := httpmw.WorkspaceAgent(r)
	// We must get the workspace to get the owner ID!
	resource, err := api.Database.GetWorkspaceResourceByID(ctx, workspaceAgent.ResourceID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to get workspace resource.",
			Detail:  err.Error(),
		})
		return
	}
	build, err := api.Database.GetWorkspaceBuildByJobID(ctx, resource.JobID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to get build.",
			Detail:  err.Error(),
		})
		return
	}
	workspace, err := api.Database.GetWorkspaceByID(ctx, build.WorkspaceID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to get workspace.",
			Detail:  err.Error(),
		})
		return
	}

	var externalAuthConfig *externalauth.Config
	if id != "" {
		// Explicit ID path: exact match only. Deterministic, and deliberately
		// unaffected by the template-scoped narrowing below.
		for _, extAuth := range api.ExternalAuthConfigs {
			if extAuth.ID == id {
				externalAuthConfig = extAuth
				break
			}
		}
	} else {
		// match-only path (GIT_ASKPASS supplies a hostname, never an ID):
		// narrow to the workspace's own template-declared providers first.
		// Only fall back to the full deployment-wide scan when every declared
		// provider is configured and none of them match this hostname (e.g. a
		// host the template never declared). Report an error rather than guess
		// when several declared providers match, or when the declaration set is
		// stale.
		//
		// Both errors use 404 so `coder gitaskpass` warns and defers to git's
		// own credential behavior instead of failing the git operation.
		declared, err := api.workspaceAgentsExternalAuthDeclaredCandidates(ctx, build, match)
		if err != nil {
			httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to resolve the workspace's template-declared external auth providers.",
				Detail:  err.Error(),
			})
			return
		}
		switch {
		case len(declared.matchedIDs) > 1:
			httpapi.Write(ctx, rw, http.StatusNotFound, codersdk.Response{
				Message: fmt.Sprintf("Multiple external auth providers declared by this workspace's template match %q: %s.", match, strings.Join(declared.matchedIDs, ", ")),
				Detail:  "Coder cannot tell which of them to use. Request a token with an explicit provider ID instead, using `coder external-auth access-token <id>`.",
			})
			return
		case declared.config != nil:
			externalAuthConfig = declared.config
		case len(declared.missingIDs) > 0:
			// A declared provider that the deployment no longer configures
			// leaves no way to tell whether it was the one meant to serve this
			// hostname, so another provider's token could silently stand in for
			// it. Refuse instead, which keeps the template scoping above intact
			// even while a template and the deployment config disagree.
			httpapi.Write(ctx, rw, http.StatusNotFound, codersdk.Response{
				Message: fmt.Sprintf("This workspace's template declares external auth provider(s) that this deployment no longer configures: %s.", strings.Join(declared.missingIDs, ", ")),
				Detail:  "Coder will not substitute a different provider's token. Restore that provider's configuration, or update the template to declare a configured provider.",
			})
			return
		default:
			for _, extAuth := range api.ExternalAuthConfigs {
				if extAuth.Regex == nil || !extAuth.Regex.MatchString(match) {
					continue
				}
				externalAuthConfig = extAuth
			}
		}
	}
	if externalAuthConfig == nil {
		detail := "External auth provider not found."
		if len(api.ExternalAuthConfigs) > 0 {
			regexURLs := make([]string, 0, len(api.ExternalAuthConfigs))
			for _, extAuth := range api.ExternalAuthConfigs {
				if extAuth.Regex == nil {
					continue
				}
				regexURLs = append(regexURLs, fmt.Sprintf("%s=%q", extAuth.ID, extAuth.Regex.String()))
			}
			detail = fmt.Sprintf("The configured external auth provider have regex filters that do not match the url. Provider url regex: %s", strings.Join(regexURLs, ","))
		}
		httpapi.Write(ctx, rw, http.StatusNotFound, codersdk.Response{
			Message: fmt.Sprintf("No matching external auth provider found in Coder for the url %q.", match),
			Detail:  detail,
		})
		return
	}

	// Pre-check if the caller can read the external auth links for the owner of the
	// workspace. Do this up front because a sql.ErrNoRows is expected if the user is
	// in the flow of authenticating. If no row is present, the auth check is delayed
	// until the user authenticates. It is preferred to reject early.
	if !api.Authorize(r, policy.ActionReadPersonal, rbac.ResourceUserObject(workspace.OwnerID)) {
		httpapi.Forbidden(rw)
		return
	}

	// MarkStale will trigger a refresh by coderd/gitsync. This allows us to
	// persist git refs as soon as the agent requests external auth so branch
	// context is retained even if the flow requires an out-of-band login.
	if gitRef.Branch != "" && gitRef.RemoteOrigin != "" {
		//nolint:gocritic // Chat processor context required for cross-user chat lookup
		api.gitSyncWorker.MarkStale(dbauthz.AsChatd(ctx), gitsync.MarkStaleParams{
			WorkspaceID: workspace.ID,
			Branch:      gitRef.Branch,
			Origin:      gitRef.RemoteOrigin,
			ChatID:      gitRef.ChatID,
		})
	}

	var previousToken *database.ExternalAuthLink
	// handleRetrying will attempt to continually check for a new token
	// if listen is true. This is useful if an error is encountered in the
	// original single flow.
	//
	// By default, if no errors are encountered, then the single flow response
	// is returned.
	handleRetrying := func(code int, response any) {
		if !listen {
			httpapi.Write(ctx, rw, code, response)
			return
		}

		api.workspaceAgentsExternalAuthListen(ctx, rw, previousToken, externalAuthConfig, workspace, gitRef)
	}

	// This is the URL that will redirect the user with a state token.
	redirectURL, err := api.AccessURL.Parse(fmt.Sprintf("/external-auth/%s", externalAuthConfig.ID))
	if err != nil {
		handleRetrying(http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to parse access URL.",
			Detail:  err.Error(),
		})
		return
	}

	externalAuthLink, err := api.Database.GetExternalAuthLink(ctx, database.GetExternalAuthLinkParams{
		ProviderID: externalAuthConfig.ID,
		UserID:     workspace.OwnerID,
	})
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			handleRetrying(http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to get external auth link.",
				Detail:  err.Error(),
			})
			return
		}

		handleRetrying(http.StatusOK, agentsdk.ExternalAuthResponse{
			URL: redirectURL.String(),
		})
		return
	}

	refreshedLink, err := externalAuthConfig.RefreshToken(ctx, api.Database, externalAuthLink)
	if err != nil && !externalauth.IsInvalidTokenError(err) {
		handleRetrying(http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to refresh external auth token.",
			Detail:  err.Error(),
		})
		return
	}
	if err != nil {
		// Set the previous token so the retry logic will skip validating the
		// same token again. This should only be set if the token is invalid and there
		// was no error. If it is invalid because of an error, then we should recheck.
		previousToken = &refreshedLink
		handleRetrying(http.StatusOK, agentsdk.ExternalAuthResponse{
			URL: redirectURL.String(),
		})
		return
	}
	resp, err := createExternalAuthResponse(externalAuthConfig.Type, refreshedLink.OAuthAccessToken, refreshedLink.OAuthExtra, refreshedLink.OAuthExpiry)
	if err != nil {
		handleRetrying(http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to create external auth response.",
			Detail:  err.Error(),
		})
		return
	}
	httpapi.Write(ctx, rw, http.StatusOK, resp)
}

// declaredExternalAuthCandidates describes how a workspace template's declared
// external auth providers relate to one hostname-only (GIT_ASKPASS) request.
type declaredExternalAuthCandidates struct {
	// config is the provider to use, set only when exactly one declared and
	// configured provider matches the hostname.
	config *externalauth.Config
	// matchedIDs holds the ID of every declared and configured provider whose
	// regex matches the hostname, in declaration order.
	matchedIDs []string
	// missingIDs holds the ID of every declared provider that the deployment
	// does not configure.
	missingIDs []string
}

// workspaceAgentsExternalAuthDeclaredCandidates resolves which configured
// external auth provider should service a hostname-only (GIT_ASKPASS) request,
// scoped to the given build's template version's declared providers. The zero
// value means the template declares no providers at all.
//
// config is set only when exactly one declared provider matched, so callers
// should distinguish four outcomes:
//   - several matchedIDs: report the collision rather than guess between them.
//   - config set: use it.
//   - no match but some missingIDs: the declaration set is stale, so report
//     that instead of substituting a provider the template never declared.
//   - no match and nothing missing: the template's providers simply do not
//     serve this hostname, so fall back to a deployment-wide scan.
func (api *API) workspaceAgentsExternalAuthDeclaredCandidates(ctx context.Context, build database.WorkspaceBuild, match string) (declaredExternalAuthCandidates, error) {
	// Template reads authorize through the template's ACL, which the owner may
	// no longer have. The version ID is server-derived from the agent's token.
	//nolint:gocritic // Agent needs system access to read its own template version's declared providers.
	sysCtx := dbauthz.AsSystemRestricted(ctx)
	templateVersion, err := api.Database.GetTemplateVersionByID(sysCtx, build.TemplateVersionID)
	if err != nil {
		return declaredExternalAuthCandidates{}, xerrors.Errorf("get template version: %w", err)
	}

	var declared []database.ExternalAuthProvider
	if err := json.Unmarshal(templateVersion.ExternalAuthProviders, &declared); err != nil {
		return declaredExternalAuthCandidates{}, xerrors.Errorf("unmarshal template version external auth providers: %w", err)
	}

	var (
		candidates []*externalauth.Config
		out        declaredExternalAuthCandidates
	)
	for _, provider := range declared {
		idx := slices.IndexFunc(api.ExternalAuthConfigs, func(extAuth *externalauth.Config) bool {
			return extAuth.ID == provider.ID
		})
		if idx < 0 {
			out.missingIDs = append(out.missingIDs, provider.ID)
			continue
		}
		extAuth := api.ExternalAuthConfigs[idx]
		if extAuth.Regex == nil || !extAuth.Regex.MatchString(match) {
			continue
		}
		candidates = append(candidates, extAuth)
	}

	for _, candidate := range candidates {
		out.matchedIDs = append(out.matchedIDs, candidate.ID)
	}
	// Only a single match is actionable. Leave config nil when several match so
	// the caller reports the collision instead of picking one arbitrarily.
	if len(candidates) == 1 {
		out.config = candidates[0]
	}
	return out, nil
}

func (api *API) workspaceAgentsExternalAuthListen(ctx context.Context, rw http.ResponseWriter, previous *database.ExternalAuthLink, externalAuthConfig *externalauth.Config, workspace database.Workspace, gitRef chatGitRef) {
	// Since we're ticking frequently and this sign-in operation is rare,
	// we are OK with polling to avoid the complexity of pubsub.
	ticker, done := api.NewTicker(time.Second)
	defer done()
	// If we have a previous token that is invalid, we should not check this again.
	// This serves to prevent doing excessive unauthorized requests to the external
	// auth provider. For github, this limit is 60 per hour, so saving a call
	// per invalid token can be significant.
	var previousToken database.ExternalAuthLink
	if previous != nil {
		previousToken = *previous
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker:
		}
		externalAuthLink, err := api.Database.GetExternalAuthLink(ctx, database.GetExternalAuthLinkParams{
			ProviderID: externalAuthConfig.ID,
			UserID:     workspace.OwnerID,
		})
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				continue
			}
			httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to get external auth link.",
				Detail:  err.Error(),
			})
			return
		}

		// Expiry may be unset if the application doesn't configure tokens
		// to expire.
		// See
		// https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app.
		if externalAuthLink.OAuthExpiry.Before(dbtime.Now()) && !externalAuthLink.OAuthExpiry.IsZero() {
			continue
		}

		// Only attempt to revalidate an oauth token if it has actually changed.
		// No point in trying to validate the same token over and over again.
		if previousToken.OAuthAccessToken == externalAuthLink.OAuthAccessToken &&
			previousToken.OAuthRefreshToken == externalAuthLink.OAuthRefreshToken &&
			previousToken.OAuthExpiry.Equal(externalAuthLink.OAuthExpiry) {
			continue
		}

		valid, _, err := externalAuthConfig.ValidateToken(ctx, externalAuthLink.OAuthToken())
		if err != nil {
			api.Logger.Warn(ctx, "failed to validate external auth token",
				slog.F("workspace_owner_id", workspace.OwnerID.String()),
				slog.F("validate_url", externalAuthConfig.ValidateURL),
				slog.Error(err),
			)
		}
		previousToken = externalAuthLink
		if !valid {
			continue
		}
		resp, err := createExternalAuthResponse(externalAuthConfig.Type, externalAuthLink.OAuthAccessToken, externalAuthLink.OAuthExtra, externalAuthLink.OAuthExpiry)
		if err != nil {
			httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
				Message: "Failed to create external auth response.",
				Detail:  err.Error(),
			})
			return
		}
		// MarkStale will trigger a refresh by coderd/gitsync.
		//nolint:gocritic // Chat processor context required for cross-user chat lookup
		api.gitSyncWorker.MarkStale(dbauthz.AsChatd(ctx), gitsync.MarkStaleParams{
			WorkspaceID: workspace.ID,
			Branch:      gitRef.Branch,
			Origin:      gitRef.RemoteOrigin,
			ChatID:      gitRef.ChatID,
		})
		httpapi.Write(ctx, rw, http.StatusOK, resp)
		return
	}
}

// @Summary User-scoped tailnet RPC connection
// @ID user-scoped-tailnet-rpc-connection
// @Security CoderSessionToken
// @Tags Agents
// @Success 101
// @Router /api/v2/tailnet [get]
func (api *API) tailnetRPCConn(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	// This is used by Enterprise code to control the functionality of this route.
	// Namely, disabling the route using `CODER_BROWSER_ONLY`.
	override := api.WorkspaceClientCoordinateOverride.Load()
	if override != nil {
		overrideFunc := *override
		if overrideFunc != nil && overrideFunc(rw) {
			return
		}
	}

	version := "2.0"
	qv := r.URL.Query().Get("version")
	if qv != "" {
		version = qv
	}
	if err := proto.CurrentVersion.Validate(version); err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Unknown or unsupported API version",
			Validations: []codersdk.ValidationError{
				{Field: "version", Detail: err.Error()},
			},
		})
		return
	}

	peerID, err := api.handleResumeToken(ctx, rw, r)
	if err != nil {
		// handleResumeToken has already written the response.
		return
	}

	// Used to authorize tunnel request
	sshPrep, err := api.HTTPAuth.AuthorizeSQLFilter(r, policy.ActionSSH, rbac.ResourceWorkspace.Type)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Internal error preparing sql filter.",
			Detail:  err.Error(),
		})
		return
	}

	api.WebsocketWaitMutex.Lock()
	api.WebsocketWaitGroup.Add(1)
	api.WebsocketWaitMutex.Unlock()
	defer api.WebsocketWaitGroup.Done()

	conn, err := websocket.Accept(rw, r, nil)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{
			Message: "Failed to accept websocket.",
			Detail:  err.Error(),
		})
		return
	}
	ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageBinary)
	defer wsNetConn.Close()
	defer conn.Close(websocket.StatusNormalClosure, "")

	// Get user ID for telemetry
	apiKey := httpmw.APIKey(r)
	userID := apiKey.UserID.String()

	// Store connection telemetry event
	now := dbtime.Now()
	connectionTelemetryEvent := telemetry.UserTailnetConnection{
		ConnectedAt:         now,
		DisconnectedAt:      nil,
		UserID:              userID,
		PeerID:              peerID.String(),
		DeviceID:            nil,
		DeviceOS:            nil,
		CoderDesktopVersion: nil,
	}

	fillCoderDesktopTelemetry(r, &connectionTelemetryEvent, api.Logger)
	api.Telemetry.Report(&telemetry.Snapshot{
		UserTailnetConnections: []telemetry.UserTailnetConnection{connectionTelemetryEvent},
	})
	defer func() {
		// Update telemetry event with disconnection time
		disconnectTime := dbtime.Now()
		connectionTelemetryEvent.DisconnectedAt = &disconnectTime
		api.Telemetry.Report(&telemetry.Snapshot{
			UserTailnetConnections: []telemetry.UserTailnetConnection{connectionTelemetryEvent},
		})
	}()

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	ctx = api.wsWatcher.Watch(ctx, api.Logger, conn)
	err = api.TailnetClientService.ServeClient(ctx, version, wsNetConn, tailnet.StreamID{
		Name: "client",
		ID:   peerID,
		Auth: tailnet.ClientUserCoordinateeAuth{
			Auth: &rbacAuthorizer{
				sshPrep: sshPrep,
				db:      api.Database,
			},
		},
	})
	if err != nil && !xerrors.Is(err, io.EOF) && !xerrors.Is(err, context.Canceled) {
		_ = conn.Close(websocket.StatusInternalError, err.Error())
		return
	}
}

// fillCoderDesktopTelemetry fills out the provided event based on a Coder Desktop telemetry header on the request, if
// present.
func fillCoderDesktopTelemetry(r *http.Request, event *telemetry.UserTailnetConnection, logger slog.Logger) {
	// Parse desktop telemetry from header if it exists
	desktopTelemetryHeader := r.Header.Get(codersdk.CoderDesktopTelemetryHeader)
	if desktopTelemetryHeader != "" {
		var telemetryData codersdk.CoderDesktopTelemetry
		if err := telemetryData.FromHeader(desktopTelemetryHeader); err == nil {
			// Only set fields if they aren't empty
			if telemetryData.DeviceID != "" {
				event.DeviceID = &telemetryData.DeviceID
			}
			if telemetryData.DeviceOS != "" {
				event.DeviceOS = &telemetryData.DeviceOS
			}
			if telemetryData.CoderDesktopVersion != "" {
				event.CoderDesktopVersion = &telemetryData.CoderDesktopVersion
			}
			logger.Debug(r.Context(), "received desktop telemetry",
				slog.F("device_id", telemetryData.DeviceID),
				slog.F("device_os", telemetryData.DeviceOS),
				slog.F("desktop_version", telemetryData.CoderDesktopVersion))
		} else {
			logger.Warn(r.Context(), "failed to parse desktop telemetry header", slog.Error(err))
		}
	}
}

// createExternalAuthResponse creates an ExternalAuthResponse based on the
// provider type. This is to support legacy `/workspaceagents/me/gitauth`
// which uses `Username` and `Password`.
func createExternalAuthResponse(typ, token string, extra pqtype.NullRawMessage, expiry time.Time) (agentsdk.ExternalAuthResponse, error) {
	var resp agentsdk.ExternalAuthResponse
	switch typ {
	case string(codersdk.EnhancedExternalAuthProviderGitLab):
		// https://stackoverflow.com/questions/25409700/using-gitlab-token-to-clone-without-authentication
		resp = agentsdk.ExternalAuthResponse{
			Username: "oauth2",
			Password: token,
		}
	case string(codersdk.EnhancedExternalAuthProviderBitBucketCloud), string(codersdk.EnhancedExternalAuthProviderBitBucketServer):
		// The string "bitbucket" was a legacy parameter that needs to still be supported.
		// https://support.atlassian.com/bitbucket-cloud/docs/use-oauth-on-bitbucket-cloud/#Cloning-a-repository-with-an-access-token
		resp = agentsdk.ExternalAuthResponse{
			Username: "x-token-auth",
			Password: token,
		}
	default:
		resp = agentsdk.ExternalAuthResponse{
			Username: token,
		}
	}
	resp.AccessToken = token
	resp.Type = typ
	// Normalize to UTC so JSON encoding always uses the "Z" suffix and
	// preserves the full timestamp without losing sub-minute precision from
	// historical timezone offsets (e.g. LMT).
	resp.ExpiresAt = expiry.UTC()

	var err error
	if extra.Valid {
		err = json.Unmarshal(extra.RawMessage, &resp.TokenExtra)
	}
	return resp, err
}

func convertWorkspaceAgentLogs(logs []database.WorkspaceAgentLog) []codersdk.WorkspaceAgentLog {
	sdk := make([]codersdk.WorkspaceAgentLog, 0, len(logs))
	for _, logEntry := range logs {
		sdk = append(sdk, db2sdk.WorkspaceAgentLog(logEntry))
	}
	return sdk
}

// workspaceAgentRefreshChatContext re-pins every drifted chat bound to the
// calling agent to the agent's latest context snapshot, clearing their
// drift markers. It backs the in-workspace `coder exp chat context refresh`
// (no chat argument), which uses the agent token rather than a user
// session.
//
// @x-apidocgen {"skip": true}
func (api *API) workspaceAgentRefreshChatContext(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	workspaceAgent := httpmw.WorkspaceAgent(r)

	// Chats are processed by the chat daemon; without it there is
	// nothing to refresh.
	if api.chatDaemon == nil {
		httpapi.Write(ctx, rw, http.StatusOK, agentsdk.RefreshChatContextResponse{})
		return
	}

	// Use system context for chat operations since the workspace agent
	// scope does not include chat resources.
	//nolint:gocritic // Agent needs system access to read/write chat resources.
	sysCtx := dbauthz.AsSystemRestricted(ctx)
	workspace, err := api.Database.GetWorkspaceByAgentID(sysCtx, workspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to determine workspace from agent token.",
			Detail:  err.Error(),
		})
		return
	}

	chats, err := api.Database.GetActiveChatsByAgentID(sysCtx, workspaceAgent.ID)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to list chats for agent.",
			Detail:  err.Error(),
		})
		return
	}

	refreshed := 0
	for _, chat := range chats {
		// Only re-pin chats owned by this workspace's owner that have
		// drifted from the agent's latest snapshot.
		if chat.OwnerID != workspace.OwnerID || !chat.ContextDirtySince.Valid {
			continue
		}
		if _, err := api.chatDaemon.RefreshChatContext(sysCtx, chat); err != nil {
			api.Logger.Warn(ctx, "failed to refresh chat context for agent",
				slog.F("chat_id", chat.ID),
				slog.F("agent_id", workspaceAgent.ID),
				slog.Error(err),
			)
			continue
		}
		refreshed++
	}

	httpapi.Write(ctx, rw, http.StatusOK, agentsdk.RefreshChatContextResponse{
		Refreshed: refreshed,
	})
}