feat: fix session termination and add per-session task serialization

- Add per-session mutex to serialize tasks: stop command fully completes
  before new message starts, preventing race conditions
- Poll task-level status (not session-level) so cancellation only affects
  the specific task, not new tasks for the same session
- Send SIGTERM instead of SIGKILL on context cancel so CLI processes can
  save session state before exiting (10s grace period)
- Add GET /v1/machines/{id}/tasks/{id}/status endpoint for task-level
  status polling

Author: cici

internal/app/runner_claude.go

@@ -10,6 +10,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/shotforward/codewithphone/internal/config"
@@ -143,6 +144,12 @@ func (r *claudeRunner) RunTurn(ctx context.Context, dispatch taskDispatch, provi
 	cmd := exec.CommandContext(ctx, r.claudeBin, args...)
 	cmd.Dir = dispatch.WorkspaceRoot
 	cmd.Env = os.Environ()
+	// Send SIGTERM instead of SIGKILL on context cancel so the CLI
+	// process can save its session state before exiting.
+	cmd.Cancel = func() error {
+		return cmd.Process.Signal(syscall.SIGTERM)
+	}
+	cmd.WaitDelay = 10 * time.Second // SIGKILL after 10s if still alive
 
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {

internal/app/runner_codex.go

@@ -15,6 +15,7 @@ import (
 	"strconv"
 	"strings"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/shotforward/codewithphone/internal/config"
@@ -789,6 +790,12 @@ type codexRPCClient struct {
 
 func startCodexRPC(ctx context.Context, codexBin string) (*codexRPCClient, error) {
 	cmd := exec.CommandContext(ctx, codexBin, "app-server", "--listen", "stdio://")
+	// Send SIGTERM instead of SIGKILL on context cancel so the CLI
+	// process can save its session/thread state before exiting.
+	cmd.Cancel = func() error {
+		return cmd.Process.Signal(syscall.SIGTERM)
+	}
+	cmd.WaitDelay = 10 * time.Second
 	stdin, err := cmd.StdinPipe()
 	if err != nil {
 		return nil, err

internal/app/runner_gemini.go

@@ -18,6 +18,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/shotforward/codewithphone/internal/config"
@@ -261,6 +262,12 @@ deny_message = "No user interactive console is available. Use mcp_pocketcode_run
 	cmd := exec.CommandContext(ctx, r.geminiBin, args...)
 	cmd.Dir = dispatch.WorkspaceRoot
 	cmd.Env = append(os.Environ(), "GEMINI_CLI_HOME="+geminiHome)
+	// Send SIGTERM instead of SIGKILL on context cancel so the CLI
+	// process can save its session state before exiting.
+	cmd.Cancel = func() error {
+		return cmd.Process.Signal(syscall.SIGTERM)
+	}
+	cmd.WaitDelay = 10 * time.Second
 
 	// Ensure GEMINI_API_KEY is set when auth type is "gemini-api-key".
 	// The Gemini CLI's validateAuthMethod checks for this env var before

internal/app/server_client.go

@@ -405,6 +405,33 @@ func (c serverClient) confirmBinding(ctx context.Context, code, confirmNonce str
 	return result.MachineToken, nil
 }
 
+func (c serverClient) fetchTaskStatus(ctx context.Context, taskRunID string) (string, error) {
+	url := strings.TrimRight(c.BaseURL, "/") + "/v1/machines/" + c.MachineID + "/tasks/" + taskRunID + "/status"
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Accept", "application/json")
+	if c.MachineToken != "" {
+		req.Header.Set("X-Machine-Token", c.MachineToken)
+	}
+	resp, err := c.httpClient().Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", newHTTPStatusError("fetch task status", resp)
+	}
+	var result struct {
+		Status string `json:"status"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", err
+	}
+	return strings.TrimSpace(result.Status), nil
+}
+
 func (c serverClient) fetchSessionStatus(ctx context.Context, sessionID string) (string, error) {
 	url := strings.TrimRight(c.BaseURL, "/") + "/v1/sessions/" + sessionID
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

internal/app/service.go

@@ -56,6 +56,7 @@ type Service struct {
 	actualAddr        string // resolved listen address (useful when port=0)
 	providerSessions  map[string]string
 	sessionWorkspaces map[string]string // sessionID -> first workspaceRoot
+	sessionLocks      map[string]*sync.Mutex // per-session lock for serial task execution
 	taskWorkspaces         map[string]string // taskRunID -> workspaceRoot
 	taskWorkspaceSnapshots map[string]string // taskRunID -> snapshot.Root for "vs turn start" diffs
 	taskProfiles           map[string]turnExecutionProfile
@@ -129,6 +130,7 @@ func New(cfg config.Config, opts ...Option) *Service {
 		interactive:       true, // default: foreground with terminal
 		providerSessions:  map[string]string{},
 		sessionWorkspaces: map[string]string{},
+		sessionLocks:      map[string]*sync.Mutex{},
 		taskWorkspaces:         map[string]string{},
 		taskWorkspaceSnapshots: map[string]string{},
 		taskProfiles:           map[string]turnExecutionProfile{},
@@ -367,6 +369,20 @@ func (s *Service) getProviderSession(sessionID string) string {
 	return s.providerSessions[sessionID]
 }
 
+// getSessionLock returns a per-session mutex, creating one if needed.
+// Tasks for the same session are serialized through this lock so that
+// a stop command is fully processed before a new message starts.
+func (s *Service) getSessionLock(sessionID string) *sync.Mutex {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	lock, ok := s.sessionLocks[sessionID]
+	if !ok {
+		lock = &sync.Mutex{}
+		s.sessionLocks[sessionID] = lock
+	}
+	return lock
+}
+
 func (s *Service) setProviderSession(sessionID, providerSessionRef string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/app/task_loop.go

@@ -85,6 +85,12 @@ func (s *Service) handleCodexDispatch(ctx context.Context, dispatch taskDispatch
 }
 
 func (s *Service) handleRunnerDispatch(ctx context.Context, dispatch taskDispatch, runner turnRunner) error {
+	// Serialize tasks for the same session: wait for any previous task
+	// (including a cancelled one being drained) to finish before starting.
+	sessionLock := s.getSessionLock(dispatch.SessionID)
+	sessionLock.Lock()
+	defer sessionLock.Unlock()
+
 	dispatchStart := time.Now()
 	profile := planTurnExecution(dispatch.Prompt)
 	// Gemini session is directory-scoped; pin workspace before any task-scoped
@@ -129,7 +135,7 @@ func (s *Service) handleRunnerDispatch(ctx context.Context, dispatch taskDispatc
 	done := make(chan struct{})
 	defer close(done)
 	terminated := make(chan struct{})
-	go s.watchSessionTermination(taskCtx, dispatch.SessionID, cancel, done, terminated)
+	go s.watchSessionTermination(taskCtx, dispatch.SessionID, dispatch.TaskRunID, cancel, done, terminated)
 
 	t0 := time.Now()
 	if err := s.serverClient.postEvent(ctx, daemonEvent{
@@ -335,7 +341,7 @@ func (s *Service) handleRunnerDispatch(ctx context.Context, dispatch taskDispatc
 }
 
 
-func (s *Service) watchSessionTermination(ctx context.Context, sessionID string, cancel context.CancelFunc, done <-chan struct{}, terminated chan struct{}) {
+func (s *Service) watchSessionTermination(ctx context.Context, sessionID, taskRunID string, cancel context.CancelFunc, done <-chan struct{}, terminated chan struct{}) {
 	ticker := time.NewTicker(sessionTerminationPollInterval)
 	defer ticker.Stop()
 	for {
@@ -345,11 +351,21 @@ func (s *Service) watchSessionTermination(ctx context.Context, sessionID string,
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			status, err := s.serverClient.fetchSessionStatus(ctx, sessionID)
+			status, err := s.serverClient.fetchTaskStatus(ctx, taskRunID)
 			if err != nil {
+				// Fallback to session-level check for backward compatibility.
+				sessStatus, sessErr := s.serverClient.fetchSessionStatus(ctx, sessionID)
+				if sessErr != nil {
+					continue
+				}
+				if sessStatus == "terminated" {
+					cancel()
+					close(terminated)
+					return
+				}
 				continue
 			}
-			if status == "terminated" {
+			if status == "cancelled" {
 				cancel()
 				close(terminated)
 				return