refactor: move snapshot loop and initial prompt logic into PTYConversation (#179)

This PR refactors the snapshot loop and initial prompt handling to be fully encapsulated within PTYConversation, removing direct field manipulation from the HTTP server layer.

* Moved StartSnapshotLoop into PTYConversation.Start()
* Initial prompt sending is now handled within PTYConversation.Start()
* Fix: Status no longer flips from "changing" → "stable" → "changing" when sending the initial prompt - it stays "changing" until the agent has actually finished responding
Also:

* Fixes race condition in ./e2e
* Improves logging in ./e2e
* Removes separate test-only code paths in ./lib/screentracker
* Refactors a lot of test code (for the better)

🤖 Created using Mux (Opus 4.5).

Author: johnstcn

cmd/server/server.go

@@ -136,7 +136,6 @@ func runServer(ctx context.Context, logger *slog.Logger, argsToPass []string) er
 		fmt.Println(srv.GetOpenAPI())
 		return nil
 	}
-	srv.StartSnapshotLoop(ctx)
 	logger.Info("Starting server on port", "port", port)
 	processExitCh := make(chan error, 1)
 	go func() {

e2e/echo.go

@@ -89,11 +89,10 @@ func runEchoAgent(scriptPath string) {
 			if entry.ThinkDurationMS > 0 {
 				redrawTerminal(messages, true)
 				spinnerCtx, spinnerCancel := context.WithCancel(ctx)
-				go runSpinner(spinnerCtx)
+				spinnerDone := runSpinner(spinnerCtx)
 				time.Sleep(time.Duration(entry.ThinkDurationMS) * time.Millisecond)
-				if spinnerCancel != nil {
-					spinnerCancel()
-				}
+				spinnerCancel()
+				<-spinnerDone
 			}
 
 			messages = append(messages, st.ConversationMessage{
@@ -133,9 +132,10 @@ func runEchoAgent(scriptPath string) {
 		if entry.ThinkDurationMS > 0 {
 			redrawTerminal(messages, true)
 			spinnerCtx, spinnerCancel := context.WithCancel(ctx)
-			go runSpinner(spinnerCtx)
+			spinnerDone := runSpinner(spinnerCtx)
 			time.Sleep(time.Duration(entry.ThinkDurationMS) * time.Millisecond)
 			spinnerCancel()
+			<-spinnerDone
 		}
 
 		messages = append(messages, st.ConversationMessage{
@@ -190,21 +190,26 @@ func cleanTerminalInput(input string) string {
 	return strings.TrimSpace(input)
 }
 
-func runSpinner(ctx context.Context) {
-	spinnerChars := []string{"|", "/", "-", "\\"}
-	ticker := time.NewTicker(200 * time.Millisecond)
-	defer ticker.Stop()
-	i := 0
-
-	for {
-		select {
-		case <-ticker.C:
-			fmt.Printf("\rThinking %s", spinnerChars[i%len(spinnerChars)])
-			i++
-		case <-ctx.Done():
-			// Clear spinner on cancellation
-			fmt.Print("\r" + strings.Repeat(" ", 20) + "\r")
-			return
+func runSpinner(ctx context.Context) <-chan struct{} {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		spinnerChars := []string{"|", "/", "-", "\\"}
+		ticker := time.NewTicker(200 * time.Millisecond)
+		defer ticker.Stop()
+		i := 0
+
+		for {
+			select {
+			case <-ticker.C:
+				fmt.Printf("\rThinking %s", spinnerChars[i%len(spinnerChars)])
+				i++
+			case <-ctx.Done():
+				// Clear spinner on cancellation
+				fmt.Print("\r" + strings.Repeat(" ", 20) + "\r")
+				return
+			}
 		}
-	}
+	}()
+	return done
 }

e2e/echo_test.go

@@ -133,14 +133,11 @@ func setup(ctx context.Context, t testing.TB, p *params) ([]ScriptEntry, *agenta
 		cwd, err := os.Getwd()
 		require.NoError(t, err, "Failed to get current working directory")
 		binaryPath = filepath.Join(cwd, "..", "out", "agentapi")
-		_, err = os.Stat(binaryPath)
-		if err != nil {
-			t.Logf("Building binary at %s", binaryPath)
-			buildCmd := exec.CommandContext(ctx, "go", "build", "-o", binaryPath, ".")
-			buildCmd.Dir = filepath.Join(cwd, "..")
-			t.Logf("run: %s", buildCmd.String())
-			require.NoError(t, buildCmd.Run(), "Failed to build binary")
-		}
+		t.Logf("Building binary at %s", binaryPath)
+		buildCmd := exec.CommandContext(ctx, "go", "build", "-o", binaryPath, ".")
+		buildCmd.Dir = filepath.Join(cwd, "..")
+		t.Logf("run: %s", buildCmd.String())
+		require.NoError(t, buildCmd.Run(), "Failed to build binary")
 	}
 
 	serverPort, err := getFreePort()
@@ -254,7 +251,11 @@ func waitAgentAPIStable(ctx context.Context, t testing.TB, apiClient *agentapisd
 					return nil
 				}
 			} else {
-				t.Logf("Got %T event", evt)
+				var sb strings.Builder
+				if err := json.NewEncoder(&sb).Encode(evt); err != nil {
+					t.Logf("Failed to encode event: %v", err)
+				}
+				t.Logf("Got event: %s", sb.String())
 			}
 		case err := <-errs:
 			return fmt.Errorf("read events: %w", err)

lib/httpapi/server.go

@@ -41,7 +41,7 @@ type Server struct {
 	srv          *http.Server
 	mu           sync.RWMutex
 	logger       *slog.Logger
-	conversation *st.PTYConversation
+	conversation st.Conversation
 	agentio      *termexec.Process
 	agentType    mf.AgentType
 	emitter      *EventEmitter
@@ -244,6 +244,14 @@ func NewServer(ctx context.Context, config ServerConfig) (*Server, error) {
 		return mf.FormatToolCall(config.AgentType, message)
 	}
 
+	emitter := NewEventEmitter(1024)
+
+	// Format initial prompt into message parts if provided
+	var initialPrompt []st.MessagePart
+	if config.InitialPrompt != "" {
+		initialPrompt = FormatMessage(config.AgentType, config.InitialPrompt)
+	}
+
 	conversation := st.NewPTY(ctx, st.PTYConversationConfig{
 		AgentType:             config.AgentType,
 		AgentIO:               config.Process,
@@ -253,9 +261,17 @@ func NewServer(ctx context.Context, config ServerConfig) (*Server, error) {
 		FormatMessage:         formatMessage,
 		ReadyForInitialPrompt: isAgentReadyForInitialPrompt,
 		FormatToolCall:        formatToolCall,
-		Logger:                logger,
-	}, config.InitialPrompt)
-	emitter := NewEventEmitter(1024)
+		InitialPrompt:         initialPrompt,
+		// OnSnapshot uses a callback rather than passing the emitter directly
+		// to keep the screentracker package decoupled from httpapi concerns.
+		// This preserves clean package boundaries and avoids import cycles.
+		OnSnapshot: func(status st.ConversationStatus, messages []st.ConversationMessage, screen string) {
+			emitter.UpdateStatusAndEmitChanges(status, config.AgentType)
+			emitter.UpdateMessagesAndEmitChanges(messages)
+			emitter.UpdateScreenAndEmitChanges(screen)
+		},
+		Logger: logger,
+	})
 
 	// Create temporary directory for uploads
 	tempDir, err := os.MkdirTemp("", "agentapi-uploads-")
@@ -281,6 +297,16 @@ func NewServer(ctx context.Context, config ServerConfig) (*Server, error) {
 	// Register API routes
 	s.registerRoutes()
 
+	// Start the conversation polling loop if we have a process.
+	// Process is nil only when --print-openapi is used (no agent runs).
+	// The process is already running at this point - termexec.StartProcess()
+	// blocks until the PTY is created and the process is active. Agent
+	// readiness (waiting for the prompt) is handled asynchronously inside
+	// conversation.Start() via ReadyForInitialPrompt.
+	if config.Process != nil {
+		s.conversation.Start(ctx)
+	}
+
 	return s, nil
 }
 
@@ -336,38 +362,6 @@ func sseMiddleware(ctx huma.Context, next func(huma.Context)) {
 	next(ctx)
 }
 
-func (s *Server) StartSnapshotLoop(ctx context.Context) {
-	s.conversation.Start(ctx)
-	go func() {
-		ticker := s.clock.NewTicker(snapshotInterval)
-		defer ticker.Stop()
-		for {
-			currentStatus := s.conversation.Status()
-
-			// Send initial prompt when agent becomes stable for the first time
-			if !s.conversation.InitialPromptSent && convertStatus(currentStatus) == AgentStatusStable {
-				if err := s.conversation.Send(FormatMessage(s.agentType, s.conversation.InitialPrompt)...); err != nil {
-					s.logger.Error("Failed to send initial prompt", "error", err)
-				} else {
-					s.conversation.InitialPromptSent = true
-					s.conversation.ReadyForInitialPrompt = false
-					currentStatus = st.ConversationStatusChanging
-					s.logger.Info("Initial prompt sent successfully")
-				}
-			}
-			s.emitter.UpdateStatusAndEmitChanges(currentStatus, s.agentType)
-			s.emitter.UpdateMessagesAndEmitChanges(s.conversation.Messages())
-			s.emitter.UpdateScreenAndEmitChanges(s.conversation.Text())
-
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-			}
-		}
-	}()
-}
-
 // registerRoutes sets up all API endpoints
 func (s *Server) registerRoutes() {
 	// GET /status endpoint