waveterm/pkg/jobcontroller/jobcontroller.go

// Copyright 2025, Command Line Inc.
// SPDX-License-Identifier: Apache-2.0

package jobcontroller

import (
	"context"
	"encoding/base64"
	"fmt"
	"io"
	"io/fs"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/wavetermdev/waveterm/pkg/blocklogger"
	"github.com/wavetermdev/waveterm/pkg/filestore"
	"github.com/wavetermdev/waveterm/pkg/panichandler"
	"github.com/wavetermdev/waveterm/pkg/remote/conncontroller"
	"github.com/wavetermdev/waveterm/pkg/streamclient"
	"github.com/wavetermdev/waveterm/pkg/telemetry"
	"github.com/wavetermdev/waveterm/pkg/telemetry/telemetrydata"
	"github.com/wavetermdev/waveterm/pkg/util/ds"
	"github.com/wavetermdev/waveterm/pkg/util/envutil"
	"github.com/wavetermdev/waveterm/pkg/util/shellutil"
	"github.com/wavetermdev/waveterm/pkg/util/utilfn"
	"github.com/wavetermdev/waveterm/pkg/utilds"
	"github.com/wavetermdev/waveterm/pkg/wavebase"
	"github.com/wavetermdev/waveterm/pkg/wavejwt"
	"github.com/wavetermdev/waveterm/pkg/waveobj"
	"github.com/wavetermdev/waveterm/pkg/wconfig"
	"github.com/wavetermdev/waveterm/pkg/wcore"
	"github.com/wavetermdev/waveterm/pkg/wps"
	"github.com/wavetermdev/waveterm/pkg/wshrpc"
	"github.com/wavetermdev/waveterm/pkg/wshrpc/wshclient"
	"github.com/wavetermdev/waveterm/pkg/wshutil"
	"github.com/wavetermdev/waveterm/pkg/wstore"
	"golang.org/x/sync/singleflight"
)

const DefaultTimeout = 2 * time.Second

const (
	JobManagerStatus_Init    = "init"
	JobManagerStatus_Running = "running"
	JobManagerStatus_Done    = "done"
)

const (
	JobDoneReason_StartupError = "startuperror"
	JobDoneReason_Gone         = "gone"
	JobDoneReason_Terminated   = "terminated"
)

const (
	JobConnStatus_Disconnected = "disconnected"
	JobConnStatus_Connecting   = "connecting"
	JobConnStatus_Connected    = "connected"
)

const (
	JobKind_Shell = "shell"
	JobKind_Task  = "task"
)

const DefaultStreamRwnd = 64 * 1024
const MetaKey_TotalGap = "totalgap"
const JobOutputFileName = "term"
const AutoReconnectDelay = 1 * time.Second
const AutoReconnectCooldown = 30 * time.Second

type connState struct {
	actual      bool
	processed   bool
	reconciling bool
}

type connStateManager struct {
	sync.Mutex
	m           map[string]*connState
	reconcileCh chan struct{}
}

type jobState struct {
	stateLock       sync.Mutex
	isConnecting    bool
	connectedStatus string
}

var (
	jobConnStates         = make(map[string]string)
	jobControllerLock     sync.Mutex
	blockJobStatusVersion utilds.VersionTs

	connStates = &connStateManager{
		m:           make(map[string]*connState),
		reconcileCh: make(chan struct{}, 1),
	}

	jobStreamIds = ds.MakeSyncMap[string]()

	jobTerminationMessageWritten = ds.MakeSyncMap[bool]()

	lastAutoReconnectAttempt = ds.MakeSyncMap[int64]()

	reconnectGroup           singleflight.Group
	terminateJobManagerGroup singleflight.Group
)

func InitJobController() {
	go connReconcileWorker()
	go jobPruningWorker()

	rpcClient := wshclient.GetBareRpcClient()
	rpcClient.EventListener.On(wps.Event_RouteUp, handleRouteUpEvent)
	rpcClient.EventListener.On(wps.Event_RouteDown, handleRouteDownEvent)
	rpcClient.EventListener.On(wps.Event_ConnChange, handleConnChangeEvent)
	rpcClient.EventListener.On(wps.Event_BlockClose, handleBlockCloseEvent)
	wshclient.EventSubCommand(rpcClient, wps.SubscriptionRequest{
		Event:     wps.Event_RouteUp,
		AllScopes: true,
	}, nil)
	wshclient.EventSubCommand(rpcClient, wps.SubscriptionRequest{
		Event:     wps.Event_RouteDown,
		AllScopes: true,
	}, nil)
	wshclient.EventSubCommand(rpcClient, wps.SubscriptionRequest{
		Event:     wps.Event_ConnChange,
		AllScopes: true,
	}, nil)
	wshclient.EventSubCommand(rpcClient, wps.SubscriptionRequest{
		Event:     wps.Event_BlockClose,
		AllScopes: true,
	}, nil)
}

func isJobManagerRunning(job *waveobj.Job) bool {
	return job.JobManagerStatus == JobManagerStatus_Running
}

func GetJobManagerStatus(ctx context.Context, jobId string) (string, error) {
	job, err := wstore.DBGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return "", fmt.Errorf("failed to get job: %w", err)
	}
	if job == nil {
		return JobManagerStatus_Done, nil
	}
	return job.JobManagerStatus, nil
}

func GetAllJobManagerStatus(ctx context.Context) ([]*wshrpc.JobManagerStatusUpdate, error) {
	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		return nil, fmt.Errorf("failed to get jobs: %w", err)
	}

	var statuses []*wshrpc.JobManagerStatusUpdate
	for _, job := range allJobs {
		statuses = append(statuses, &wshrpc.JobManagerStatusUpdate{
			JobId:            job.OID,
			JobManagerStatus: job.JobManagerStatus,
		})
	}

	return statuses, nil
}

func GetBlockJobStatus(ctx context.Context, blockId string) (*wshrpc.BlockJobStatusData, error) {
	block, err := wstore.DBGet[*waveobj.Block](ctx, blockId)
	if err != nil {
		return nil, fmt.Errorf("failed to get block: %w", err)
	}
	if block == nil {
		return nil, fmt.Errorf("block not found: %s", blockId)
	}

	data := &wshrpc.BlockJobStatusData{
		BlockId:   blockId,
		VersionTs: blockJobStatusVersion.GetVersionTs(),
	}

	if block.JobId == "" {
		return data, nil
	}

	job, err := wstore.DBGet[*waveobj.Job](ctx, block.JobId)
	if err != nil {
		return nil, fmt.Errorf("failed to get job: %w", err)
	}
	if job == nil {
		return data, nil
	}

	data.JobId = job.OID
	data.DoneReason = job.JobManagerDoneReason
	data.StartupError = job.JobManagerStartupError
	data.CmdExitTs = job.CmdExitTs
	data.CmdExitCode = job.CmdExitCode
	data.CmdExitSignal = job.CmdExitSignal

	if job.JobManagerStatus == JobManagerStatus_Init {
		data.Status = "init"
	} else if job.JobManagerStatus == JobManagerStatus_Done {
		data.Status = "done"
	} else if job.JobManagerStatus == JobManagerStatus_Running {
		connStatus := GetJobConnStatus(job.OID)
		if connStatus == JobConnStatus_Connected {
			data.Status = "connected"
		} else {
			data.Status = "disconnected"
		}
	}

	return data, nil
}

func SendBlockJobStatusEvent(ctx context.Context, blockId string) {
	data, err := GetBlockJobStatus(ctx, blockId)
	if err != nil {
		log.Printf("[block:%s] error getting block job status: %v", blockId, err)
		return
	}
	wps.Broker.Publish(wps.WaveEvent{
		Event:  wps.Event_BlockJobStatus,
		Scopes: []string{fmt.Sprintf("block:%s", blockId)},
		Data:   data,
	})
}

func sendBlockJobStatusEventByJob(ctx context.Context, job *waveobj.Job) {
	if job == nil || job.AttachedBlockId == "" {
		return
	}
	SendBlockJobStatusEvent(ctx, job.AttachedBlockId)
}

func connReconcileWorker() {
	defer func() {
		panichandler.PanicHandler("jobcontroller:connReconcileWorker", recover())
	}()

	for range connStates.reconcileCh {
		reconcileAllConns()
	}
}

func reconcileAllConns() {
	connStates.Lock()
	defer connStates.Unlock()

	for connName, cs := range connStates.m {
		if cs.reconciling || cs.actual == cs.processed {
			continue
		}

		cs.reconciling = true
		actual := cs.actual
		go reconcileConn(connName, actual)
	}
}

func reconcileConn(connName string, targetState bool) {
	defer func() {
		panichandler.PanicHandler("jobcontroller:reconcileConn", recover())
	}()

	if targetState {
		onConnectionUp(connName)
	} else {
		onConnectionDown(connName)
	}

	connStates.Lock()
	defer connStates.Unlock()
	if cs, exists := connStates.m[connName]; exists {
		cs.processed = targetState
		cs.reconciling = false
	}

	select {
	case connStates.reconcileCh <- struct{}{}:
	default:
	}
}

func getMetaInt64(meta wshrpc.FileMeta, key string) int64 {
	val, ok := meta[key]
	if !ok {
		return 0
	}
	if intVal, ok := val.(int64); ok {
		return intVal
	}
	if floatVal, ok := val.(float64); ok {
		return int64(floatVal)
	}
	return 0
}

func jobPruningWorker() {
	defer func() {
		panichandler.PanicHandler("jobcontroller:jobPruningWorker", recover())
	}()

	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	var previousCandidates []string
	for range ticker.C {
		previousCandidates = pruneUnusedJobs(previousCandidates)
	}
}

func pruneUnusedJobs(previousCandidates []string) []string {
	ctx, cancelFn := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancelFn()

	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		log.Printf("[jobpruner] error getting all jobs: %v", err)
		return previousCandidates
	}

	var currentCandidates []string
	for _, job := range allJobs {
		if job.JobManagerStatus == JobManagerStatus_Done && job.AttachedBlockId == "" {
			currentCandidates = append(currentCandidates, job.OID)
		}
	}

	jobsToDelete := utilfn.StrSetIntersection(previousCandidates, currentCandidates)
	if len(previousCandidates) > 0 || len(currentCandidates) > 0 {
		log.Printf("[jobpruner] prev=%d current=%d deleting=%d", len(previousCandidates), len(currentCandidates), len(jobsToDelete))
	}

	for _, jobId := range jobsToDelete {
		err := DeleteJob(ctx, jobId)
		if err != nil {
			log.Printf("[jobpruner] error deleting job %s: %v", jobId, err)
		}
	}

	return currentCandidates
}

func handleRouteUpEvent(event *wps.WaveEvent) {
	handleRouteEvent(event, JobConnStatus_Connected)
}

func handleRouteDownEvent(event *wps.WaveEvent) {
	handleRouteEvent(event, JobConnStatus_Disconnected)
}

func handleRouteEvent(event *wps.WaveEvent, newStatus string) {
	ctx := context.Background()
	for _, scope := range event.Scopes {
		if strings.HasPrefix(scope, "job:") {
			jobId := strings.TrimPrefix(scope, "job:")
			SetJobConnStatus(jobId, newStatus)
			log.Printf("[job:%s] connection status changed to %s", jobId, newStatus)

			job, err := wstore.DBGet[*waveobj.Job](ctx, jobId)
			if err != nil {
				log.Printf("[job:%s] error getting job for status event: %v", jobId, err)
				continue
			}
			sendBlockJobStatusEventByJob(ctx, job)

			if newStatus == JobConnStatus_Disconnected && job != nil && isJobManagerRunning(job) {
				if shouldAttemptAutoReconnect(jobId) {
					go attemptAutoReconnect(jobId, job.Connection)
				}
			}
		}
	}
}

func shouldAttemptAutoReconnect(jobId string) bool {
	now := time.Now().Unix()
	lastAttempt, exists := lastAutoReconnectAttempt.GetEx(jobId)

	if !exists {
		lastAutoReconnectAttempt.Set(jobId, now)
		return true
	}

	timeSinceLastAttempt := time.Duration(now-lastAttempt) * time.Second
	if timeSinceLastAttempt >= AutoReconnectCooldown {
		lastAutoReconnectAttempt.Set(jobId, now)
		return true
	}

	return false
}

func attemptAutoReconnect(jobId string, connName string) {
	defer func() {
		panichandler.PanicHandler("jobcontroller:attemptAutoReconnect", recover())
	}()

	time.Sleep(AutoReconnectDelay)

	isConnected, err := conncontroller.IsConnected(connName)
	if err != nil || !isConnected {
		log.Printf("[job:%s] connection %s is down, skipping auto-reconnect", jobId, connName)
		return
	}

	log.Printf("[job:%s] connection %s still up after route down, attempting auto-reconnect to determine job manager status", jobId, connName)
	ctx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelFn()
	err = ReconnectJob(ctx, jobId, nil)
	if err != nil {
		log.Printf("[job:%s] auto-reconnect failed: %v", jobId, err)
	} else {
		log.Printf("[job:%s] auto-reconnect succeeded", jobId)
	}
}

func handleConnChangeEvent(event *wps.WaveEvent) {
	var connStatus wshrpc.ConnStatus
	err := utilfn.ReUnmarshal(&connStatus, event.Data)
	if err != nil {
		log.Printf("[connchange] error unmarshaling ConnStatus: %v", err)
		return
	}

	var connName string
	for _, scope := range event.Scopes {
		if strings.HasPrefix(scope, "connection:") {
			connName = strings.TrimPrefix(scope, "connection:")
			break
		}
	}
	if connName == "" {
		return
	}

	connStates.Lock()
	cs, exists := connStates.m[connName]
	if !exists {
		cs = &connState{actual: false, processed: false, reconciling: false}
		connStates.m[connName] = cs
	}
	cs.actual = connStatus.Connected
	connStates.Unlock()

	select {
	case connStates.reconcileCh <- struct{}{}:
	default:
	}
}

func handleBlockCloseEvent(event *wps.WaveEvent) {
	ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancelFn()
	blockId, ok := event.Data.(string)
	if !ok {
		log.Printf("[blockclose] invalid event data type")
		return
	}

	jobIds, err := wstore.WithTxRtn(ctx, func(tx *wstore.TxWrap) ([]string, error) {
		query := `SELECT oid FROM db_job WHERE json_extract(data, '$.attachedblockid') = ?`
		jobIds := tx.SelectStrings(query, blockId)
		return jobIds, nil
	})
	if err != nil {
		log.Printf("[block:%s] error looking up jobids: %v", blockId, err)
		return
	}
	if len(jobIds) == 0 {
		return
	}

	for _, jobId := range jobIds {
		TerminateAndDetachJob(ctx, jobId)
	}
}

func onConnectionUp(connName string) {
	log.Printf("[conn:%s] connection became connected, reconnecting jobs", connName)
	ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancelFn()

	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		log.Printf("[conn:%s] failed to get jobs for reconnection: %v", connName, err)
		return
	}

	var jobsToReconnect []*waveobj.Job
	for _, job := range allJobs {
		if job.Connection == connName && isJobManagerRunning(job) {
			jobsToReconnect = append(jobsToReconnect, job)
		}
	}

	log.Printf("[conn:%s] found %d jobs to reconnect", connName, len(jobsToReconnect))

	successCount := 0
	for _, job := range jobsToReconnect {
		err = ReconnectJob(ctx, job.OID, nil)
		if err != nil {
			log.Printf("[job:%s] error reconnecting: %v", job.OID, err)
		} else {
			successCount++
		}
	}

	log.Printf("[conn:%s] finished reconnecting jobs: %d/%d successful", connName, successCount, len(jobsToReconnect))
}

func onConnectionDown(connName string) {
	log.Printf("[conn:%s] connection became disconnected", connName)
}

func GetJobConnStatus(jobId string) string {
	jobControllerLock.Lock()
	defer jobControllerLock.Unlock()
	status, exists := jobConnStates[jobId]
	if !exists {
		return JobConnStatus_Disconnected
	}
	return status
}

func SetJobConnStatus(jobId string, status string) {
	jobControllerLock.Lock()
	defer jobControllerLock.Unlock()
	if status == JobConnStatus_Disconnected {
		delete(jobConnStates, jobId)
	} else {
		jobConnStates[jobId] = status
	}
}

func GetConnectedJobIds() []string {
	jobControllerLock.Lock()
	defer jobControllerLock.Unlock()
	var connectedJobIds []string
	for jobId, status := range jobConnStates {
		if status == JobConnStatus_Connected {
			connectedJobIds = append(connectedJobIds, jobId)
		}
	}
	return connectedJobIds
}

func GetNumJobsRunning() int {
	ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancelFn()
	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		return 0
	}
	count := 0
	for _, job := range allJobs {
		if job.JobManagerStatus == JobManagerStatus_Running {
			count++
		}
	}
	return count
}

func GetNumJobsConnected() int {
	jobControllerLock.Lock()
	defer jobControllerLock.Unlock()
	count := 0
	for _, status := range jobConnStates {
		if status == JobConnStatus_Connected {
			count++
		}
	}
	return count
}

func CheckJobConnected(ctx context.Context, jobId string) (*waveobj.Job, error) {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return nil, fmt.Errorf("failed to get job: %w", err)
	}

	isConnected, err := conncontroller.IsConnected(job.Connection)
	if err != nil {
		return nil, fmt.Errorf("error checking connection status: %w", err)
	}
	if !isConnected {
		return nil, fmt.Errorf("connection %q is not connected", job.Connection)
	}

	jobConnStatus := GetJobConnStatus(jobId)
	if jobConnStatus != JobConnStatus_Connected {
		return nil, fmt.Errorf("job is not connected (status: %s)", jobConnStatus)
	}

	return job, nil
}

type StartJobParams struct {
	ConnName string
	JobKind  string
	Cmd      string
	Args     []string
	Env      map[string]string
	TermSize *waveobj.TermSize
	BlockId  string
}

func StartJob(ctx context.Context, params StartJobParams) (string, error) {
	if params.ConnName == "" {
		return "", fmt.Errorf("connection name is required")
	}
	if params.JobKind != JobKind_Shell && params.JobKind != JobKind_Task {
		return "", fmt.Errorf("jobkind must be %q or %q", JobKind_Shell, JobKind_Task)
	}
	if params.Cmd == "" {
		return "", fmt.Errorf("command is required")
	}
	if params.TermSize == nil {
		params.TermSize = &waveobj.TermSize{Rows: 24, Cols: 80}
	}

	isConnected, err := conncontroller.IsConnected(params.ConnName)
	if err != nil {
		return "", fmt.Errorf("error checking connection status: %w", err)
	}
	if !isConnected {
		return "", fmt.Errorf("connection %q is not connected", params.ConnName)
	}

	jobId := uuid.New().String()
	jobAuthToken, err := utilfn.RandomHexString(32)
	if err != nil {
		return "", fmt.Errorf("failed to generate job auth token: %w", err)
	}

	jobAccessClaims := &wavejwt.WaveJwtClaims{
		MainServer: true,
		JobId:      jobId,
	}
	jobAccessToken, err := wavejwt.Sign(jobAccessClaims)
	if err != nil {
		return "", fmt.Errorf("failed to generate job access token: %w", err)
	}

	job := &waveobj.Job{
		OID:              jobId,
		Connection:       params.ConnName,
		JobKind:          params.JobKind,
		Cmd:              params.Cmd,
		CmdArgs:          params.Args,
		CmdEnv:           params.Env,
		CmdTermSize:      *params.TermSize,
		JobAuthToken:     jobAuthToken,
		JobManagerStatus: JobManagerStatus_Init,
		AttachedBlockId:  params.BlockId,
		WaveVersion:      wavebase.WaveVersion,
		Meta:             make(waveobj.MetaMapType),
	}

	err = wstore.DBInsert(ctx, job)
	if err != nil {
		return "", fmt.Errorf("failed to create job in database: %w", err)
	}
	if params.BlockId != "" {
		// AttachJobToBlock will send status
		err = AttachJobToBlock(ctx, jobId, params.BlockId)
		if err != nil {
			return "", fmt.Errorf("failed to attach job to block: %w", err)
		}
	}
	bareRpc := wshclient.GetBareRpcClient()
	broker := bareRpc.StreamBroker
	readerRouteId := wshclient.GetBareRpcClientRouteId()
	writerRouteId := wshutil.MakeJobRouteId(jobId)
	reader, streamMeta := broker.CreateStreamReader(readerRouteId, writerRouteId, DefaultStreamRwnd)
	jobStreamIds.Set(jobId, streamMeta.Id)

	fileOpts := wshrpc.FileOpts{
		MaxSize:  10 * 1024 * 1024,
		Circular: true,
	}
	err = filestore.WFS.MakeFile(ctx, jobId, JobOutputFileName, wshrpc.FileMeta{}, fileOpts)
	if err != nil {
		return "", fmt.Errorf("failed to create WaveFS file: %w", err)
	}

	clientId := wstore.GetClientId()
	publicKey := wavejwt.GetPublicKey()
	publicKeyBase64 := base64.StdEncoding.EncodeToString(publicKey)
	jobEnv := envutil.CopyAndAddToEnvMap(params.Env, "WAVETERM_JOBID", jobId)
	startJobData := wshrpc.CommandRemoteStartJobData{
		Cmd:                params.Cmd,
		Args:               params.Args,
		Env:                jobEnv,
		TermSize:           *params.TermSize,
		StreamMeta:         streamMeta,
		JobAuthToken:       jobAuthToken,
		JobId:              jobId,
		MainServerJwtToken: jobAccessToken,
		ClientId:           clientId,
		PublicKeyBase64:    publicKeyBase64,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeConnectionRouteId(params.ConnName),
		Timeout: 30000,
	}

	writeSessionSeparatorToTerminal(params.BlockId, params.TermSize.Cols)

	log.Printf("[job:%s] sending RemoteStartJobCommand to connection %s, cmd=%q, args=%v", jobId, params.ConnName, params.Cmd, params.Args)
	log.Printf("[job:%s] env=%v", jobId, params.Env)
	rtnData, err := wshclient.RemoteStartJobCommand(bareRpc, startJobData, rpcOpts)
	if err != nil {
		log.Printf("[job:%s] RemoteStartJobCommand failed: %v", jobId, err)
		errMsg := fmt.Sprintf("failed to start job: %v", err)
		var updatedJob *waveobj.Job
		wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
			job.JobManagerStatus = JobManagerStatus_Done
			job.JobManagerDoneReason = JobDoneReason_StartupError
			job.JobManagerStartupError = errMsg
			updatedJob = job
		})
		sendBlockJobStatusEventByJob(ctx, updatedJob)
		telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
			Event: "job:done",
			Props: telemetrydata.TEventProps{
				JobDoneReason: JobDoneReason_StartupError,
				JobKind:       params.JobKind,
			},
		})
		return "", fmt.Errorf("failed to start remote job: %w", err)
	}

	log.Printf("[job:%s] RemoteStartJobCommand succeeded, cmdpid=%d cmdstartts=%d jobmanagerpid=%d jobmanagerstartts=%d", jobId, rtnData.CmdPid, rtnData.CmdStartTs, rtnData.JobManagerPid, rtnData.JobManagerStartTs)
	var updatedJob *waveobj.Job
	err = wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
		job.CmdPid = rtnData.CmdPid
		job.CmdStartTs = rtnData.CmdStartTs
		job.JobManagerPid = rtnData.JobManagerPid
		job.JobManagerStartTs = rtnData.JobManagerStartTs
		job.JobManagerStatus = JobManagerStatus_Running
		updatedJob = job
	})
	if err != nil {
		log.Printf("[job:%s] warning: failed to update job status to running: %v", jobId, err)
	} else {
		log.Printf("[job:%s] job status updated to running", jobId)
		sendBlockJobStatusEventByJob(ctx, updatedJob)
	}

	telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
		Event: "job:start",
		Props: telemetrydata.TEventProps{
			JobKind: params.JobKind,
		},
	})

	go func() {
		defer func() {
			panichandler.PanicHandler("jobcontroller:runOutputLoop", recover())
		}()
		runOutputLoop(context.Background(), jobId, streamMeta.Id, reader)
	}()

	return jobId, nil
}

func doWFSAppend(ctx context.Context, oref waveobj.ORef, fileName string, data []byte) error {
	err := filestore.WFS.AppendData(ctx, oref.OID, fileName, data)
	if err != nil {
		return err
	}
	wps.Broker.Publish(wps.WaveEvent{
		Event: wps.Event_BlockFile,
		Scopes: []string{
			oref.String(),
		},
		Data: &wps.WSFileEventData{
			ZoneId:   oref.OID,
			FileName: fileName,
			FileOp:   wps.FileOp_Append,
			Data64:   base64.StdEncoding.EncodeToString(data),
		},
	})
	return nil
}

func handleAppendJobFile(ctx context.Context, jobId string, fileName string, data []byte) error {
	err := doWFSAppend(ctx, waveobj.MakeORef(waveobj.OType_Job, jobId), fileName, data)
	if err != nil {
		return fmt.Errorf("error appending to job file: %w", err)
	}

	job, err := wstore.DBGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return fmt.Errorf("error getting job: %w", err)
	}
	if job != nil && job.AttachedBlockId != "" {
		err = doWFSAppend(ctx, waveobj.MakeORef(waveobj.OType_Block, job.AttachedBlockId), fileName, data)
		if err != nil {
			return fmt.Errorf("error appending to block file: %w", err)
		}
	}

	return nil
}

func runOutputLoop(ctx context.Context, jobId string, streamId string, reader *streamclient.Reader) {
	defer reader.Close()
	defer func() {
		log.Printf("[job:%s] [stream:%s] output loop finished", jobId, streamId)
	}()

	log.Printf("[job:%s] [stream:%s] output loop started", jobId, streamId)
	buf := make([]byte, 4096)
	for {
		n, err := reader.Read(buf)
		currentStreamId, _ := jobStreamIds.GetEx(jobId)
		if currentStreamId != streamId {
			log.Printf("[job:%s] [stream:%s] stream superseded by [stream:%s], exiting output loop", jobId, streamId, currentStreamId)
			break
		}
		if n > 0 {
			appendErr := handleAppendJobFile(ctx, jobId, JobOutputFileName, buf[:n])
			if appendErr != nil {
				log.Printf("[job:%s] error appending data to WaveFS: %v", jobId, appendErr)
			}
		}

		if err == io.EOF {
			log.Printf("[job:%s] stream ended (EOF)", jobId)
			updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
				job.StreamDone = true
			})
			if updateErr != nil {
				log.Printf("[job:%s] error updating job stream status: %v", jobId, updateErr)
			}
			tryTerminateJobManager(ctx, jobId)
			break
		}

		if err != nil {
			log.Printf("[job:%s] stream error: %v", jobId, err)
			streamErr := err.Error()
			updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
				job.StreamDone = true
				job.StreamError = streamErr
			})
			if updateErr != nil {
				log.Printf("[job:%s] error updating job stream error: %v", jobId, updateErr)
			}
			tryTerminateJobManager(ctx, jobId)
			break
		}
	}
}

func HandleCmdJobExited(ctx context.Context, jobId string, data wshrpc.CommandJobCmdExitedData) error {
	var updatedJob *waveobj.Job
	err := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
		job.CmdExitError = data.ExitErr
		job.CmdExitCode = data.ExitCode
		job.CmdExitSignal = data.ExitSignal
		job.CmdExitTs = data.ExitTs
		updatedJob = job
	})
	if err != nil {
		return fmt.Errorf("failed to update job exit status: %w", err)
	}
	sendBlockJobStatusEventByJob(ctx, updatedJob)
	tryTerminateJobManager(ctx, jobId)

	shouldWrite := jobTerminationMessageWritten.TestAndSet(jobId, true, func(val bool, exists bool) bool {
		return !exists || !val
	})
	if shouldWrite {
		resetTerminalState(ctx, updatedJob.AttachedBlockId)
		msg := "shell terminated"
		if updatedJob.CmdExitCode != nil && *updatedJob.CmdExitCode != 0 {
			msg = fmt.Sprintf("shell terminated (exit code %d)", *updatedJob.CmdExitCode)
		} else if updatedJob.CmdExitSignal != "" {
			msg = fmt.Sprintf("shell terminated (signal %s)", updatedJob.CmdExitSignal)
		}
		writeMutedMessageToTerminal(updatedJob.AttachedBlockId, "["+msg+"]")
	}
	return nil
}

func tryTerminateJobManager(ctx context.Context, jobId string) {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		log.Printf("[job:%s] error getting job for termination check: %v", jobId, err)
		return
	}

	if job.JobManagerStatus != JobManagerStatus_Running {
		return
	}

	cmdExited := job.CmdExitTs != 0

	if !cmdExited || !job.StreamDone {
		log.Printf("[job:%s] not ready for termination: exited=%v streamDone=%v", jobId, cmdExited, job.StreamDone)
		return
	}

	log.Printf("[job:%s] both job cmd exited and stream finished, terminating job manager", jobId)

	err = TerminateJobManager(ctx, jobId)
	if err != nil {
		log.Printf("[job:%s] error terminating job manager: %v", jobId, err)
	}
}

func TerminateAndDetachJob(ctx context.Context, jobId string) {
	err := TerminateJobManager(ctx, jobId)
	if err != nil {
		log.Printf("[job:%s] error terminating job manager: %v", jobId, err)
	}
	err = DetachJobFromBlock(ctx, jobId, true)
	if err != nil {
		log.Printf("[job:%s] error detaching job from block: %v", jobId, err)
	}
}

func TerminateJobManager(ctx context.Context, jobId string) error {
	_, err, _ := terminateJobManagerGroup.Do(jobId, func() (any, error) {
		err := doTerminateJobManager(ctx, jobId)
		return nil, err
	})
	return err
}

func doTerminateJobManager(ctx context.Context, jobId string) error {
	var shouldTerminate bool
	var job *waveobj.Job
	err := wstore.DBUpdateFn(ctx, jobId, func(j *waveobj.Job) {
		job = j
		if j.JobManagerStatus == JobManagerStatus_Done {
			shouldTerminate = false
			return
		}
		j.TerminateOnReconnect = true
		shouldTerminate = true
	})
	if err != nil {
		return fmt.Errorf("failed to set TerminateOnReconnect: %w", err)
	}

	if !shouldTerminate {
		log.Printf("[job:%s] already terminated, skipping", jobId)
		return nil
	}

	return remoteTerminateJobManager(ctx, job)
}

func DisconnectJob(ctx context.Context, jobId string) error {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return fmt.Errorf("failed to get job: %w", err)
	}

	bareRpc := wshclient.GetBareRpcClient()
	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeConnectionRouteId(job.Connection),
		Timeout: 5000,
	}

	disconnectData := wshrpc.CommandRemoteDisconnectFromJobManagerData{
		JobId: jobId,
	}

	err = wshclient.RemoteDisconnectFromJobManagerCommand(bareRpc, disconnectData, rpcOpts)
	if err != nil {
		return fmt.Errorf("failed to send disconnect command: %w", err)
	}

	log.Printf("[job:%s] job disconnect command sent successfully", jobId)
	return nil
}

func remoteTerminateJobManager(ctx context.Context, job *waveobj.Job) error {
	log.Printf("[job:%s] terminating job manager", job.OID)

	shouldWrite := jobTerminationMessageWritten.TestAndSet(job.OID, true, func(val bool, exists bool) bool {
		return !exists || !val
	})
	if shouldWrite {
		resetTerminalState(ctx, job.AttachedBlockId)
		writeMutedMessageToTerminal(job.AttachedBlockId, "[shell terminated]")
	}

	if job.JobManagerStatus == JobManagerStatus_Done {
		log.Printf("[job:%s] job manager already marked as done, skipping termination", job.OID)
		return nil
	}

	bareRpc := wshclient.GetBareRpcClient()
	terminateData := wshrpc.CommandRemoteTerminateJobManagerData{
		JobId:             job.OID,
		JobManagerPid:     job.JobManagerPid,
		JobManagerStartTs: job.JobManagerStartTs,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeConnectionRouteId(job.Connection),
		Timeout: 5000,
	}

	err := wshclient.RemoteTerminateJobManagerCommand(bareRpc, terminateData, rpcOpts)
	if err != nil {
		log.Printf("[job:%s] error terminating job manager: %v", job.OID, err)
		return fmt.Errorf("failed to terminate job manager: %w", err)
	}

	var updatedJob *waveobj.Job
	updateErr := wstore.DBUpdateFn(ctx, job.OID, func(job *waveobj.Job) {
		job.JobManagerStatus = JobManagerStatus_Done
		job.JobManagerDoneReason = JobDoneReason_Terminated
		job.TerminateOnReconnect = false
		if !job.StreamDone {
			job.StreamDone = true
			job.StreamError = "job manager terminated"
		}
		updatedJob = job
	})
	if updateErr != nil {
		log.Printf("[job:%s] error updating job status after termination: %v", job.OID, updateErr)
	} else {
		sendBlockJobStatusEventByJob(ctx, updatedJob)
	}

	telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
		Event: "job:done",
		Props: telemetrydata.TEventProps{
			JobDoneReason: JobDoneReason_Terminated,
			JobKind:       job.JobKind,
		},
	})

	log.Printf("[job:%s] job manager terminated successfully", job.OID)
	return nil
}

func ReconnectJob(ctx context.Context, jobId string, rtOpts *waveobj.RuntimeOpts) error {
	_, err, _ := reconnectGroup.Do(jobId, func() (any, error) {
		return nil, doReconnectJob(ctx, jobId, rtOpts)
	})
	return err
}

func doReconnectJob(ctx context.Context, jobId string, rtOpts *waveobj.RuntimeOpts) error {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return fmt.Errorf("failed to get job: %w", err)
	}

	_, err = CheckJobConnected(ctx, jobId)
	if err == nil {
		log.Printf("[job:%s] already connected, skipping reconnect", jobId)
		return nil
	}
	log.Printf("[job:%s] not connected, proceeding with reconnect: %v", jobId, err)

	isConnected, err := conncontroller.IsConnected(job.Connection)
	if err != nil {
		return fmt.Errorf("error checking connection status: %w", err)
	}
	if !isConnected {
		return fmt.Errorf("connection %q is not connected", job.Connection)
	}

	if job.TerminateOnReconnect {
		return remoteTerminateJobManager(ctx, job)
	}

	if rtOpts == nil {
		rtOpts = &waveobj.RuntimeOpts{
			TermSize: job.CmdTermSize,
		}
	}

	bareRpc := wshclient.GetBareRpcClient()

	jobAccessClaims := &wavejwt.WaveJwtClaims{
		MainServer: true,
		JobId:      jobId,
	}
	jobAccessToken, err := wavejwt.Sign(jobAccessClaims)
	if err != nil {
		return fmt.Errorf("failed to generate job access token: %w", err)
	}

	reconnectData := wshrpc.CommandRemoteReconnectToJobManagerData{
		JobId:              jobId,
		JobAuthToken:       job.JobAuthToken,
		MainServerJwtToken: jobAccessToken,
		JobManagerPid:      job.JobManagerPid,
		JobManagerStartTs:  job.JobManagerStartTs,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeConnectionRouteId(job.Connection),
		Timeout: 5000,
	}

	log.Printf("[job:%s] sending RemoteReconnectToJobManagerCommand to connection %s", jobId, job.Connection)
	rtnData, err := wshclient.RemoteReconnectToJobManagerCommand(bareRpc, reconnectData, rpcOpts)
	if err != nil {
		log.Printf("[job:%s] RemoteReconnectToJobManagerCommand failed: %v", jobId, err)
		return fmt.Errorf("failed to reconnect to job manager: %w", err)
	}

	if !rtnData.Success {
		log.Printf("[job:%s] RemoteReconnectToJobManagerCommand returned error: %s", jobId, rtnData.Error)
		if rtnData.JobManagerGone {
			var updatedJob *waveobj.Job
			updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
				job.JobManagerStatus = JobManagerStatus_Done
				job.JobManagerDoneReason = JobDoneReason_Gone
				updatedJob = job
			})
			if updateErr != nil {
				log.Printf("[job:%s] error updating job manager running status: %v", jobId, updateErr)
			} else {
				sendBlockJobStatusEventByJob(ctx, updatedJob)
			}
			telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
				Event: "job:done",
				Props: telemetrydata.TEventProps{
					JobDoneReason: JobDoneReason_Gone,
					JobKind:       job.JobKind,
				},
			})
			writeJobTerminationMessage(ctx, jobId, updatedJob, "[session gone]")
			return fmt.Errorf("job manager has exited: %s", rtnData.Error)
		}
		return fmt.Errorf("failed to reconnect to job manager: %s", rtnData.Error)
	}

	log.Printf("[job:%s] RemoteReconnectToJobManagerCommand succeeded, waiting for route", jobId)

	routeId := wshutil.MakeJobRouteId(jobId)
	waitCtx, cancelFn := context.WithTimeout(ctx, 2*time.Second)
	defer cancelFn()
	err = wshutil.DefaultRouter.WaitForRegister(waitCtx, routeId)
	if err != nil {
		return fmt.Errorf("route did not establish after successful reconnection: %w", err)
	}
	SetJobConnStatus(jobId, JobConnStatus_Connected)
	sendBlockJobStatusEventByJob(ctx, job)

	telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
		Event: "job:reconnect",
		Props: telemetrydata.TEventProps{
			JobKind: job.JobKind,
		},
	})

	log.Printf("[job:%s] route established, restarting streaming", jobId)
	return restartStreaming(ctx, jobId, true, rtOpts)
}

func ReconnectJobsForConn(ctx context.Context, connName string) error {
	isConnected, err := conncontroller.IsConnected(connName)
	if err != nil {
		return fmt.Errorf("error checking connection status: %w", err)
	}
	if !isConnected {
		return fmt.Errorf("connection %q is not connected", connName)
	}

	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		return fmt.Errorf("failed to get jobs: %w", err)
	}

	var jobsToReconnect []*waveobj.Job
	for _, job := range allJobs {
		if job.Connection == connName && isJobManagerRunning(job) {
			jobsToReconnect = append(jobsToReconnect, job)
		}
	}

	log.Printf("[conn:%s] found %d jobs to reconnect", connName, len(jobsToReconnect))

	for _, job := range jobsToReconnect {
		err = ReconnectJob(ctx, job.OID, nil)
		if err != nil {
			log.Printf("[job:%s] error reconnecting: %v", job.OID, err)
		}
	}

	return nil
}

func restartStreaming(ctx context.Context, jobId string, knownConnected bool, rtOpts *waveobj.RuntimeOpts) error {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)
	if err != nil {
		return fmt.Errorf("failed to get job: %w", err)
	}

	termSize := job.CmdTermSize
	if rtOpts != nil && rtOpts.TermSize.Rows > 0 && rtOpts.TermSize.Cols > 0 {
		termSize = rtOpts.TermSize
		err = wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
			job.CmdTermSize = termSize
		})
		if err != nil {
			log.Printf("[job:%s] warning: failed to update termsize in DB: %v", jobId, err)
		}
	}

	if !knownConnected {
		isConnected, err := conncontroller.IsConnected(job.Connection)
		if err != nil {
			return fmt.Errorf("error checking connection status: %w", err)
		}
		if !isConnected {
			return fmt.Errorf("connection %q is not connected", job.Connection)
		}

		jobConnStatus := GetJobConnStatus(jobId)
		if jobConnStatus != JobConnStatus_Connected {
			return fmt.Errorf("job manager is not connected (status: %s)", jobConnStatus)
		}
	}

	var currentSeq int64 = 0
	var totalGap int64 = 0
	waveFile, err := filestore.WFS.Stat(ctx, jobId, JobOutputFileName)
	if err == nil {
		currentSeq = waveFile.Size
		totalGap = getMetaInt64(waveFile.Meta, MetaKey_TotalGap)
		currentSeq += totalGap
	}

	bareRpc := wshclient.GetBareRpcClient()
	broker := bareRpc.StreamBroker
	readerRouteId := wshclient.GetBareRpcClientRouteId()
	writerRouteId := wshutil.MakeJobRouteId(jobId)
	reader, streamMeta := broker.CreateStreamReaderWithSeq(readerRouteId, writerRouteId, DefaultStreamRwnd, currentSeq)
	jobStreamIds.Set(jobId, streamMeta.Id)

	prepareData := wshrpc.CommandJobPrepareConnectData{
		StreamMeta: *streamMeta,
		Seq:        currentSeq,
		TermSize:   termSize,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeJobRouteId(jobId),
		Timeout: 5000,
	}

	log.Printf("[job:%s] sending JobPrepareConnectCommand with seq=%d (fileSize=%d, totalGap=%d)", jobId, currentSeq, waveFile.Size, totalGap)
	rtnData, err := wshclient.JobPrepareConnectCommand(bareRpc, prepareData, rpcOpts)
	if err != nil {
		reader.Close()
		return fmt.Errorf("failed to prepare connect: %w", err)
	}

	if rtnData.HasExited {
		exitCodeStr := "nil"
		if rtnData.ExitCode != nil {
			exitCodeStr = fmt.Sprintf("%d", *rtnData.ExitCode)
		}
		log.Printf("[job:%s] job has already exited: code=%s signal=%q err=%q", jobId, exitCodeStr, rtnData.ExitSignal, rtnData.ExitErr)
		exitData := wshrpc.CommandJobCmdExitedData{
			ExitCode:   rtnData.ExitCode,
			ExitSignal: rtnData.ExitSignal,
			ExitErr:    rtnData.ExitErr,
			ExitTs:     time.Now().UnixMilli(),
		}
		HandleCmdJobExited(ctx, jobId, exitData)
	}

	if rtnData.StreamDone {
		log.Printf("[job:%s] stream is already done: error=%q", jobId, rtnData.StreamError)
		updateErr := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
			if !job.StreamDone {
				job.StreamDone = true
				if rtnData.StreamError != "" {
					job.StreamError = rtnData.StreamError
				}
			}
		})
		if updateErr != nil {
			log.Printf("[job:%s] error updating job stream status: %v", jobId, updateErr)
		}
	}

	if rtnData.StreamDone && rtnData.HasExited {
		reader.Close()
		log.Printf("[job:%s] both stream done and job exited, calling tryExitJobManager", jobId)
		tryTerminateJobManager(ctx, jobId)
		return nil
	}

	if rtnData.StreamDone {
		reader.Close()
		log.Printf("[job:%s] stream already done, no need to restart streaming", jobId)
		return nil
	}

	if rtnData.Seq > currentSeq {
		gap := rtnData.Seq - currentSeq
		totalGap += gap
		log.Printf("[job:%s] detected gap: our seq=%d, server seq=%d, gap=%d, new totalGap=%d", jobId, currentSeq, rtnData.Seq, gap, totalGap)

		metaErr := filestore.WFS.WriteMeta(ctx, jobId, JobOutputFileName, wshrpc.FileMeta{
			MetaKey_TotalGap: totalGap,
		}, true)
		if metaErr != nil {
			log.Printf("[job:%s] error updating totalgap metadata: %v", jobId, metaErr)
		}

		reader.UpdateNextSeq(rtnData.Seq)
	}

	log.Printf("[job:%s] sending JobStartStreamCommand", jobId)
	startStreamData := wshrpc.CommandJobStartStreamData{}
	err = wshclient.JobStartStreamCommand(bareRpc, startStreamData, rpcOpts)
	if err != nil {
		reader.Close()
		return fmt.Errorf("failed to start stream: %w", err)
	}

	go func() {
		defer func() {
			panichandler.PanicHandler("jobcontroller:RestartStreaming:runOutputLoop", recover())
		}()
		runOutputLoop(context.Background(), jobId, streamMeta.Id, reader)
	}()

	log.Printf("[job:%s] streaming restarted successfully", jobId)
	return nil
}

// this function must be kept up to date with getBlockTermDurableAtom in frontend/app/store/global.ts
func IsBlockTermDurable(block *waveobj.Block) bool {
	if block == nil {
		return false
	}

	// Check if view is "term", and controller is "shell"
	if block.Meta.GetString(waveobj.MetaKey_View, "") != "term" || block.Meta.GetString(waveobj.MetaKey_Controller, "") != "shell" {
		return false
	}

	// 1. Check if block has a JobId
	if block.JobId != "" {
		return true
	}

	// 2. Check if connection is local or WSL (not durable)
	connName := block.Meta.GetString(waveobj.MetaKey_Connection, "")
	if conncontroller.IsLocalConnName(connName) || conncontroller.IsWslConnName(connName) {
		return false
	}

	// 3. Check config hierarchy: blockmeta → connection → global (default true)
	// Check block meta first
	if val, exists := block.Meta[waveobj.MetaKey_TermDurable]; exists {
		if boolVal, ok := val.(bool); ok {
			return boolVal
		}
	}
	// Check connection config
	fullConfig := wconfig.GetWatcher().GetFullConfig()
	if connName != "" {
		if connConfig, exists := fullConfig.Connections[connName]; exists {
			if connConfig.TermDurable != nil {
				return *connConfig.TermDurable
			}
		}
	}
	// Check global settings
	if fullConfig.Settings.TermDurable != nil {
		return *fullConfig.Settings.TermDurable
	}
	// Default to true for non-local connections
	return true
}

func IsBlockIdTermDurable(blockId string) bool {
	block, err := wstore.DBGet[*waveobj.Block](context.Background(), blockId)
	if err != nil || block == nil {
		return false
	}
	return IsBlockTermDurable(block)
}

func DeleteJob(ctx context.Context, jobId string) error {
	SetJobConnStatus(jobId, JobConnStatus_Disconnected)
	jobTerminationMessageWritten.Delete(jobId)
	err := filestore.WFS.DeleteZone(ctx, jobId)
	if err != nil {
		log.Printf("[job:%s] warning: error deleting WaveFS zone: %v", jobId, err)
	}
	return wstore.DBDelete(ctx, waveobj.OType_Job, jobId)
}

func AttachJobToBlock(ctx context.Context, jobId string, blockId string) error {
	err := wstore.WithTx(ctx, func(tx *wstore.TxWrap) error {
		var oldJobId string

		err := wstore.DBUpdateFn(tx.Context(), blockId, func(block *waveobj.Block) {
			oldJobId = block.JobId
			block.JobId = jobId
		})
		if err != nil {
			return fmt.Errorf("failed to update block: %w", err)
		}

		if oldJobId != "" && oldJobId != jobId {
			err = wstore.DBUpdateFn(tx.Context(), oldJobId, func(oldJob *waveobj.Job) {
				if oldJob.AttachedBlockId == blockId {
					oldJob.AttachedBlockId = ""
				}
			})
			if err != nil {
				log.Printf("[job:%s] warning: could not detach old job: %v", oldJobId, err)
			}
		}

		err = wstore.DBUpdateFnErr(tx.Context(), jobId, func(job *waveobj.Job) error {
			if job.AttachedBlockId != "" && job.AttachedBlockId != blockId {
				return fmt.Errorf("job %s already attached to block %s", jobId, job.AttachedBlockId)
			}
			job.AttachedBlockId = blockId
			return nil
		})
		if err != nil {
			return fmt.Errorf("failed to update job: %w", err)
		}

		log.Printf("[job:%s] attached to block:%s", jobId, blockId)
		return nil
	})
	if err != nil {
		return err
	}

	SendBlockJobStatusEvent(ctx, blockId)
	wcore.SendWaveObjUpdate(waveobj.MakeORef(waveobj.OType_Block, blockId))
	return nil
}

func DetachJobFromBlock(ctx context.Context, jobId string, updateBlock bool) error {
	var blockId string
	var blockUpdated bool
	err := wstore.WithTx(ctx, func(tx *wstore.TxWrap) error {
		job, err := wstore.DBMustGet[*waveobj.Job](tx.Context(), jobId)
		if err != nil {
			return fmt.Errorf("failed to get job: %w", err)
		}

		blockId = job.AttachedBlockId
		if blockId == "" {
			return nil
		}

		if updateBlock {
			block, err := wstore.DBGet[*waveobj.Block](tx.Context(), blockId)
			if err == nil && block != nil {
				err = wstore.DBUpdateFn(tx.Context(), blockId, func(block *waveobj.Block) {
					block.JobId = ""
				})
				if err != nil {
					log.Printf("[job:%s] warning: failed to clear JobId from block:%s: %v", jobId, blockId, err)
				} else {
					blockUpdated = true
				}
			}
		}

		err = wstore.DBUpdateFn(tx.Context(), jobId, func(job *waveobj.Job) {
			job.AttachedBlockId = ""
		})
		if err != nil {
			return fmt.Errorf("failed to update job: %w", err)
		}

		log.Printf("[job:%s] detached from block:%s", jobId, blockId)
		return nil
	})
	if err != nil {
		return err
	}

	if blockId != "" {
		SendBlockJobStatusEvent(ctx, blockId)
		if blockUpdated {
			wcore.SendWaveObjUpdate(waveobj.MakeORef(waveobj.OType_Block, blockId))
		}
	}

	return nil
}

func SendInput(ctx context.Context, data wshrpc.CommandJobInputData) error {
	jobId := data.JobId

	if data.TermSize != nil {
		err := wstore.DBUpdateFn(ctx, jobId, func(job *waveobj.Job) {
			job.CmdTermSize = *data.TermSize
		})
		if err != nil {
			log.Printf("[job:%s] warning: failed to update termsize in DB: %v", jobId, err)
		}
	}

	_, err := CheckJobConnected(ctx, jobId)
	if err != nil {
		return err
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:      wshutil.MakeJobRouteId(jobId),
		Timeout:    5000,
		NoResponse: false,
	}

	bareRpc := wshclient.GetBareRpcClient()
	err = wshclient.JobInputCommand(bareRpc, data, rpcOpts)
	if err != nil {
		return fmt.Errorf("failed to send input to job: %w", err)
	}

	return nil
}

func resetTerminalState(logCtx context.Context, blockId string) {
	if blockId == "" {
		return
	}
	ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout)
	defer cancelFn()
	if isFileEmpty(ctx, blockId) {
		return
	}
	blocklogger.Debugf(logCtx, "[conndebug] resetTerminalState: resetting terminal state for block\n")
	resetSeq := shellutil.GetTerminalResetSeq()
	resetSeq += "\r\n"
	err := doWFSAppend(ctx, waveobj.MakeORef(waveobj.OType_Block, blockId), JobOutputFileName, []byte(resetSeq))
	if err != nil {
		log.Printf("error appending terminal reset to block file: %v\n", err)
	}
}

func isFileEmpty(ctx context.Context, blockId string) bool {
	if blockId == "" {
		return true
	}
	file, statErr := filestore.WFS.Stat(ctx, blockId, JobOutputFileName)
	if statErr == fs.ErrNotExist {
		return true
	}
	if statErr != nil {
		log.Printf("error statting block output file: %v\n", statErr)
		return true
	}
	return file.Size == 0
}

func writeSessionSeparatorToTerminal(blockId string, termWidth int) {
	if blockId == "" {
		return
	}
	ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout)
	defer cancelFn()
	if isFileEmpty(ctx, blockId) {
		return
	}
	separatorLine := "\r\n"
	err := doWFSAppend(ctx, waveobj.MakeORef(waveobj.OType_Block, blockId), JobOutputFileName, []byte(separatorLine))
	if err != nil {
		log.Printf("error writing session separator to terminal (blockid=%s): %v", blockId, err)
	}
}

// msg should not have a terminating newline
func writeMutedMessageToTerminal(blockId string, msg string) {
	if blockId == "" {
		return
	}
	ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout)
	defer cancelFn()
	fullMsg := "\x1b[90m" + msg + "\x1b[0m\r\n"
	err := doWFSAppend(ctx, waveobj.MakeORef(waveobj.OType_Block, blockId), JobOutputFileName, []byte(fullMsg))
	if err != nil {
		log.Printf("error writing muted message to terminal (blockid=%s): %v", blockId, err)
	}
}

func writeJobTerminationMessage(ctx context.Context, jobId string, job *waveobj.Job, msg string) {
	if job == nil {
		return
	}
	shouldWrite := jobTerminationMessageWritten.TestAndSet(jobId, true, func(val bool, exists bool) bool {
		return !exists || !val
	})
	if shouldWrite {
		resetTerminalState(ctx, job.AttachedBlockId)
		writeMutedMessageToTerminal(job.AttachedBlockId, msg)
	}
}