attach jobs end with their command: relay watcher, once-only start banner, worker-side ttyd cleanup

Three attach-mode (-I) fixes, verified end-to-end on the t1 cluster:

1. Session hang on job finish (job_attach.go): the relay's client->worker
   pump blocked forever in ReadMessage after the worker hop closed, so the
   old wg.Wait()-before-Close teardown never ran and the client hung, then
   died with an abnormal 1006. A broker watcher now subscribes to
   EventJobFinished and on job end sends the 4001 policy close and closes
   both hops, unblocking the pumps; the teardown follows normally. The
   client (attach.go) treats 4001/1000/1001 -- and 1006-after-output as a
   final safety net -- as a clean session end.

2. "<<Starting on t1 >>" printed repeatedly (submit.go): the status poll
   printed the banner on every transition; now gated on a reportedStart
   flag. The URL is also no longer printed in attach mode -- it only makes
   sense for browse (-I --tty --browse) and DISPLAY jobs.

3. ttyd left behind after an attach job exits (xcecd_worker.c): a new
   tty_stop_attach_session(), called from process_exit_callback for TTY
   jobs, kills the per-job ttyd right away. This is insurance -- the sh -c
   exec means uv_proc monitors ttyd's own pid, so ttyd is normally gone
   before the callback fires -- but it guarantees no orphan when the
   timing lines up badly. Web-viewed sessions are untouched: they keep
   ttyd and output alive for later browsing, per design.

Co-Authored-By: Claude Code <noreply@anthropic.com>

src/frontend/apiserver/controllers/job_attach.go

@@ -15,7 +15,6 @@ import (
 
 	logger "apiserver/utils"
 
-
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
 	"go.uber.org/zap"
@@ -166,7 +165,7 @@ func (j JobController) AttachJob(ctx *gin.Context) {
 		zap.Int64("jobid", job.Unique_JobID), zap.String("user", logonUser),
 		zap.String("worker_url", workerURL))
 
-	attachRelay(clientConn, workerConn)
+	attachRelay(uint64(job.Unique_JobID), clientConn, workerConn)
 	logger.Info("attach: relay finished", zap.Int64("jobid", job.Unique_JobID))
 }
 
@@ -196,13 +195,17 @@ func attachTTYDURL(raw string) (string, error) {
 	return u.String(), nil
 }
 
-// attachClose sends one close frame (best-effort) and stops reading.
+// attachClose sends one close frame (best-effort) and waits for the peer's
+// closing handshake before the caller tears the TCP connection down. Without
+// the drain the deferred Close() can race the frame out the socket and the
+// client sees 1006 (abnormal closure) instead of our policy code.
 func attachClose(conn *websocket.Conn, code int, text string) {
 	deadline := time.Now().Add(attachWriteTimeout)
 	conn.WriteControl(websocket.CloseMessage,
 		websocket.FormatCloseMessage(code, text), deadline)
-	// Drain so the close frame is actually processed by gorilla's internal
-	// control handler before the connection is torn down.
+	// Drain until the peer answers the handshake (or the deadline): each
+	// NextReader pumps gorilla's internal control-frame handler, which
+	// completes the close exchange on this side.
 	conn.SetReadDeadline(deadline)
 	for {
 		if _, _, err := conn.NextReader(); err != nil {
@@ -215,12 +218,13 @@ func attachClose(conn *websocket.Conn, code int, text string) {
 // goes away. Each direction runs in its own goroutine; the first direction
 // to finish tears the whole relay down (and closes both connections, which
 // unblocks the other direction's in-flight Read).
-func attachRelay(client, worker *websocket.Conn) {
-	// Oversized pipes: terminal output can burst; the pipes absorb jitter
-	// between the two hops. On ping/pong: both hops are terminal sessions,
-	// so application-level traffic flows constantly when attached; the
-	// deadlines below exist to catch a dead peer at all, not to time out
-	// interactive silence mid-command.
+//
+// The relay also watches the job's terminal state: an attach session is over
+// the moment its job finishes, even if the worker-side websocket lingers
+// (ttyd's --exit-with-program does not always fire promptly with a client
+// still attached). The watch fires a policy close to the client in that
+// case; the pumps' teardown below runs as usual.
+func attachRelay(jobID uint64, client, worker *websocket.Conn) {
 	var wg sync.WaitGroup
 	wg.Add(2)
 	done := make(chan struct{})
@@ -228,16 +232,61 @@ func attachRelay(client, worker *websocket.Conn) {
 	go func() {
 		defer wg.Done()
 		if !attachPump(worker, client, done) {
-			// worker side ended on its own: the session (or its job) ended
+			// worker side ended on its own: the session (or its job) ended.
+			// Send the policy close and complete the handshake BEFORE the
+			// teardown goroutine closes the socket -- attachClose blocks
+			// until the client answers or the deadline passes, so the
+			// client reliably reads 4001 instead of a bare 1006.
 			attachClose(client, attachCloseJobEnded, "interactive session ended")
 		}
 	}()
 	go func() {
 		defer wg.Done()
+		// client-initiated end: forward nothing, teardown below suffices.
 		attachPump(client, worker, done)
 	}()
 
-	// Unblock both read pumps when either direction finishes.
+	// The job ending is a session end regardless of the worker hop: send
+	// the policy close and tear the relay down. WriteControl (not the full
+	// attachClose handshake) here: the watcher may race a pump already
+	// closing this connection, and the teardown below follows immediately,
+	// so a best-effort frame is enough -- the client tolerates both a 4001
+	// and an abrupt end after output.
+	go func() {
+		ch := backend.DefaultBroker.Subscribe(4)
+		defer backend.DefaultBroker.Unsubscribe(ch)
+		for {
+			select {
+			case <-done:
+				return
+			case ev, ok := <-ch:
+				if !ok {
+					return
+				}
+				if ev.JobID == jobID && ev.Type == backend.EventJobFinished {
+					select {
+					case <-done:
+					default:
+						logger.Info("attach: job finished, ending relayed session",
+							zap.Uint64("jobid", jobID))
+						_ = client.WriteControl(websocket.CloseMessage,
+							websocket.FormatCloseMessage(attachCloseJobEnded,
+								"interactive session ended (job finished)"),
+							time.Now().Add(attachWriteTimeout))
+					}
+					// unblock the pumps; teardown closes both hops
+					go func() {
+						client.Close()
+						worker.Close()
+					}()
+					return
+				}
+			}
+		}
+	}()
+
+	// Unblock both read pumps when either direction finishes (including the
+	// attachClose handshake window above).
 	go func() {
 		wg.Wait()
 		close(done)
@@ -251,8 +300,9 @@ func attachRelay(client, worker *websocket.Conn) {
 // attachPump copies one direction: reads frames from src, writes them to
 // dst. Binary and text frames pass through with their type; close/other
 // control frames end the pump (gorilla surfaces close via the read error).
-// Returns false when src ended first (its peer outlived it), true when the
-// dst side ended.
+// Returns false when src ended on its own (its peer outlived it, and the
+// caller may want to send it a policy close); true when the end was caused
+// by dst or by our own teardown (nothing left to tell).
 func attachPump(src, dst *websocket.Conn, done <-chan struct{}) bool {
 	for {
 		msgType, data, err := src.ReadMessage()
@@ -268,6 +318,8 @@ func attachPump(src, dst *websocket.Conn, done <-chan struct{}) bool {
 		}
 		_ = dst.SetWriteDeadline(time.Now().Add(attachWriteTimeout))
 		if err := dst.WriteMessage(msgType, data); err != nil {
+			// dst went away first; unblock src's reads via teardown and
+			// report "not src-ended" so no policy close is sent to dst.
 			return true
 		}
 	}

src/frontend/xcecli/cmd/jobs/attach.go

@@ -49,7 +49,9 @@ func attachEndpoint(serverURL, jobID string) (string, error) {
 }
 
 // runAttachSession connects to the relay, puts the local terminal in raw
-// mode and copies bytes both ways until the session ends.
+// mode and copies bytes both ways until the session ends. sawOutput tracks
+// whether the session ever delivered terminal data, so an abrupt close
+// after real output is reported as a normal end instead of an error.
 func runAttachSession(serverURL, token, jobID string) error {
 	endpoint, err := attachEndpoint(serverURL, jobID)
 	if err != nil {
@@ -75,6 +77,8 @@ func runAttachSession(serverURL, token, jobID string) error {
 	}
 	defer conn.Close()
 
+	sawOutput := false
+
 	// terminal raw mode around the whole session
 	oldState, err := term.MakeRaw(int(os.Stdin.Fd()))
 	if err != nil {
@@ -132,8 +136,18 @@ func runAttachSession(serverURL, token, jobID string) error {
 	for {
 		mt, data, err := conn.ReadMessage()
 		if err != nil {
-			// 4001 = session ended cleanly (job finished / server side closed)
+			// 4001 = session ended cleanly (job finished / server side
+			// closed). 1006 often carries the same fact: ttyd tears the TCP
+			// connection down immediately after a fast-exiting command (its
+			// -Q path) and the relay's close frame can race that teardown,
+			// so treat an abrupt end with prior session output the same as
+			// a clean 4001 rather than scaring the user with an error.
 			if websocket.IsCloseError(err, 4001, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+				fmt.Println("<<interactive session ended>>")
+				return nil
+			}
+			if websocket.IsCloseError(err, websocket.CloseAbnormalClosure) && sawOutput {
+				fmt.Println("<<interactive session ended>>")
 				return nil
 			}
 			return fmt.Errorf("session ended: [%s]", attachErrText(err))
@@ -143,6 +157,7 @@ func runAttachSession(serverURL, token, jobID string) error {
 		}
 		switch data[0] {
 		case '0': // OUTPUT
+			sawOutput = true
 			if _, err := os.Stdout.Write(data[1:]); err != nil {
 				return nil
 			}

src/frontend/xcecli/cmd/jobs/submit.go

@@ -58,6 +58,7 @@ type JobSubmitter struct {
 	useTTY         bool
 	useXJob        bool
 	useBrowse      bool
+	reportedStart  bool
 	projectName    string
 	askedHosts     string
 	launchType     string
@@ -510,25 +511,27 @@ func (js *JobSubmitter) RunJobsSubmit() error {
 		if err != nil {
 			return err
 		}
-		fmt.Printf("interactive session URL: %s\n", url)
 
 		if js.useTTY {
 			// TTY jobs attach by default: the session is displayed right in
 			// this terminal through the apiserver relay (worker ttyd ports are
-			// not reachable from user clients in production). --browse instead
-			// opens the ttyd web URL in the browser.
-			if js.useBrowse {
-				if err := openBrowser(url); err != nil {
-					errMsg := request.ExtractErrorMessage(err)
-					fmt.Printf("warning: failed to open browser: [%s]\n", errMsg)
-					fmt.Println("Please manually open the URL in your browser.")
-				}
-				return nil
+			// not reachable from user clients in production). The URL is an
+			// implementation detail of the relay -- not printed unless the
+			// user asked for the web view. --browse opens the ttyd web URL
+			// in the browser instead of attaching.
+			if !js.useBrowse {
+				return runAttachSession(serverInfo.URL, serverInfo.Token, jobID)
 			}
-			return runAttachSession(serverInfo.URL, serverInfo.Token, jobID)
+			if err := openBrowser(url); err != nil {
+				errMsg := request.ExtractErrorMessage(err)
+				fmt.Printf("warning: failed to open browser: [%s]\n", errMsg)
+				fmt.Println("Please manually open the URL in your browser.")
+			}
+			return nil
 		}
 
 		// XJOB (VNC) has no terminal rendering -- browser remains the path
+		fmt.Printf("interactive session URL: %s\n", url)
 		if err := openBrowser(url); err != nil {
 			errMsg := request.ExtractErrorMessage(err)
 			fmt.Printf("warning: failed to open browser: [%s]\n", errMsg)
@@ -576,9 +579,15 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin
 		case "PEND":
 			return row.JobStatus, nil
 		case "RUN":
-			var execHosts models.ExecHosts
-			if err := json.Unmarshal([]byte(row.ExecHosts.String()), &execHosts); err == nil && len(execHosts.HostSlots) > 0 {
-				fmt.Println("<<Starting on", execHosts.HostSlots[0].HostName, ">>")
+			// Print the dispatch line once: waitForInteractiveURL keeps
+			// polling while RUN until the session URL arrives, and this
+			// poll loop re-entered here would repeat the message.
+			if !js.reportedStart {
+				var execHosts models.ExecHosts
+				if err := json.Unmarshal([]byte(row.ExecHosts.String()), &execHosts); err == nil && len(execHosts.HostSlots) > 0 {
+					fmt.Println("<<Starting on", execHosts.HostSlots[0].HostName, ">>")
+					js.reportedStart = true
+				}
 			}
 			return row.JobStatus, nil
 		default:

src/worker/src/xcecd_worker.c

@@ -60,6 +60,7 @@ static void check_and_reconnect();
 static void worker_reconnect_callback(bool connected);
 static int  xjob_start_vnc();
 static void xjob_stop_vnc();
+static void tty_stop_attach_session();
 static bool check_process_status(pid_t target_pid, int expected_signal,
                                  int timeout_seconds);
 
@@ -515,6 +516,12 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
         xjob_stop_vnc();
     }
 
+    /* attach-mode TTY job: the session ends with the command (web-viewed
+     * TTY jobs keep ttyd for later browsing -- see tty_stop_attach_session) */
+    if (TTY == mydata.transfer_info->interactive_mode) {
+        tty_stop_attach_session();
+    }
+
     if (job_runtime_info.job_post_exec[0] != '\0') {
         job_post_exec();
     }
@@ -710,6 +717,38 @@ static void xjob_stop_vnc()
                      jobidStr);
 }
 
+/* An attached TTY job (interactive=1, the xcecli attach path) is done the
+ * moment its command exits: the user watches the same terminal, so there is
+ * nothing left to serve. Kill the ttyd we spawned for the job right away --
+ * ttyd's own -Q teardown cannot be relied on here: with its uv/libwebsockets
+ * foreign event loop it logs "exiting due to --exit-with-program" but stays
+ * alive while a websocket client (the apiserver relay) is still connected,
+ * which would leave the attach hanging and the ttyd orphaned.
+ *
+ * Web-viewed sessions (--browse / DISPLAY) intentionally keep ttyd and the
+ * output alive after the command ends until a browser connects or the job
+ * is reaped; tty_stop_attach_session is only called for the attach path. */
+static void tty_stop_attach_session()
+{
+    pid_t ttyd_pid = uv_proc_get_pid(&uv_proc_manager);
+
+    if (ttyd_pid <= 0) {
+        return;
+    }
+
+    char jobidStr[64];
+    fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
+    XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Terminating ttyd %d of attach job %s",
+                     ttyd_pid, jobidStr);
+
+    signal_children_recursive(SIGKILL, ttyd_pid);
+    if (kill(ttyd_pid, SIGKILL) != 0 && errno != ESRCH) {
+        XCECD_WORKER_LOG(XCE_LOG_WARN,
+                         "Failed to kill ttyd %d of job %s: %s", ttyd_pid,
+                         jobidStr, strerror(errno));
+    }
+}
+
 static void worker_job_callback(job_assignment_t* info)
 {
     char* job_cmd           = NULL;