Cover pause and resume in the e2e run

The idempotency assertion compared counter deltas between the two runs,
but per-run counters are reset at the start of every run, so the second
run's row already showed that run alone — the delta was negative and the
script failed before reaching anything else.

With that fixed, a second account of 3000 messages exercises the new
stop path end to end: pause mid-folder, assert the task and the account
settle into paused, resume, and assert the resumed run covers every
message while skipping the ones the paused stretch had already copied.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

internal/httpapi/accounts.go

@@ -154,6 +154,82 @@ func (s *Server) handleCreateAccount(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusCreated, map[string]int64{"id": id})
 }
 
+// handleUpdateAccountCredentials fixes the logins/passwords of an existing
+// account — typically after an import brought in a wrong password and the
+// connection test failed. An empty password field keeps the stored one, so the
+// operator can correct one side without retyping the other.
+func (s *Server) handleUpdateAccountCredentials(w http.ResponseWriter, r *http.Request) {
+	taskID, err := pathID(r, "id")
+	if err != nil {
+		http.Error(w, "bad id", http.StatusBadRequest)
+		return
+	}
+	accID, err := pathID(r, "accountId")
+	if err != nil {
+		http.Error(w, "bad account id", http.StatusBadRequest)
+		return
+	}
+	task, err := s.store.GetTask(r.Context(), taskID)
+	if err != nil {
+		http.Error(w, "not found", http.StatusNotFound)
+		return
+	}
+	acc, ok := s.findAccount(r, taskID, accID)
+	if !ok {
+		http.Error(w, "account not found", http.StatusNotFound)
+		return
+	}
+	if task.Status == "running" || acc.Status == "running" {
+		http.Error(w, "cannot change credentials while the account is running", http.StatusConflict)
+		return
+	}
+	var body struct {
+		SrcLogin string `json:"src_login"`
+		SrcPass  string `json:"src_pass"`
+		DstLogin string `json:"dst_login"`
+		DstPass  string `json:"dst_pass"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		http.Error(w, "bad json", http.StatusBadRequest)
+		return
+	}
+	// Same trimming as account creation: pasted logins/passwords often carry a
+	// stray space or newline that the IMAP server rejects.
+	body.SrcLogin = strings.TrimSpace(body.SrcLogin)
+	body.DstLogin = strings.TrimSpace(body.DstLogin)
+	body.SrcPass = strings.TrimSpace(body.SrcPass)
+	body.DstPass = strings.TrimSpace(body.DstPass)
+	if body.SrcLogin == "" || body.DstLogin == "" {
+		http.Error(w, "src_login and dst_login are required", http.StatusBadRequest)
+		return
+	}
+	encrypt := func(pass string) (*string, error) {
+		if pass == "" {
+			return nil, nil // keep the stored password
+		}
+		enc, err := crypto.Encrypt(s.cfg.EncKey, []byte(pass))
+		if err != nil {
+			return nil, err
+		}
+		return &enc, nil
+	}
+	srcEnc, err := encrypt(body.SrcPass)
+	if err != nil {
+		http.Error(w, "encrypt", http.StatusInternalServerError)
+		return
+	}
+	dstEnc, err := encrypt(body.DstPass)
+	if err != nil {
+		http.Error(w, "encrypt", http.StatusInternalServerError)
+		return
+	}
+	if err := s.store.UpdateAccountCredentials(r.Context(), accID, body.SrcLogin, body.DstLogin, srcEnc, dstEnc); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
 // findAccount returns the account with accID under taskID, or ok=false.
 func (s *Server) findAccount(r *http.Request, taskID, accID int64) (store.Account, bool) {
 	accs, err := s.store.ListAccountsByTask(r.Context(), taskID)

internal/httpapi/router.go

@@ -27,9 +27,13 @@ func (s *Server) Router() http.Handler {
 	api.HandleFunc("GET /api/tasks/{id}/runs", s.handleListRuns)
 	api.HandleFunc("GET /api/tasks/{id}/accounts/{accountId}/errors", s.handleListAccountErrors)
 	api.HandleFunc("DELETE /api/tasks/{id}/accounts/{accountId}", s.handleDeleteAccount)
+	api.HandleFunc("PUT /api/tasks/{id}/accounts/{accountId}/credentials", s.handleUpdateAccountCredentials)
 	api.HandleFunc("POST /api/tasks/{id}/import", s.handleImportCSV)
 	api.HandleFunc("POST /api/tasks/{id}/test", s.handleTestAccounts)
 	api.HandleFunc("POST /api/tasks/{id}/run", s.handleRun)
+	api.HandleFunc("POST /api/tasks/{id}/pause", s.handlePauseRun)
+	api.HandleFunc("POST /api/tasks/{id}/cancel", s.handleCancelRun)
+	api.HandleFunc("POST /api/tasks/{id}/resume", s.handleResumeRun)
 	api.HandleFunc("POST /api/tasks/{id}/accounts/{accountId}/cancel", s.handleCancelAccount)
 	api.HandleFunc("POST /api/tasks/{id}/accounts/{accountId}/probe", s.handleProbeAccountFolders)
 	api.HandleFunc("PUT /api/tasks/{id}/accounts/{accountId}/folder-mapping", s.handleSetAccountFolderMapping)

internal/httpapi/run.go

@@ -127,6 +127,56 @@ func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusAccepted, map[string]int64{"run_id": runID})
 }
 
+// handlePauseRun stops the live run but keeps the unfinished accounts
+// resumable; handleCancelRun stops it for good. Both are no-ops (409) when the
+// task has no run in flight.
+func (s *Server) handlePauseRun(w http.ResponseWriter, r *http.Request) {
+	s.stopRun(w, r, s.orch.PauseTask)
+}
+
+func (s *Server) handleCancelRun(w http.ResponseWriter, r *http.Request) {
+	s.stopRun(w, r, s.orch.CancelTask)
+}
+
+func (s *Server) stopRun(w http.ResponseWriter, r *http.Request, stop func(int64) bool) {
+	taskID, err := pathID(r, "id")
+	if err != nil {
+		http.Error(w, "bad id", http.StatusBadRequest)
+		return
+	}
+	if !stop(taskID) {
+		http.Error(w, "task is not running", http.StatusConflict)
+		return
+	}
+	w.WriteHeader(http.StatusAccepted)
+}
+
+// handleResumeRun restarts a paused task with the accounts its pause left
+// unfinished; already-copied messages are skipped by the migration journal.
+func (s *Server) handleResumeRun(w http.ResponseWriter, r *http.Request) {
+	taskID, err := pathID(r, "id")
+	if err != nil {
+		http.Error(w, "bad id", http.StatusBadRequest)
+		return
+	}
+	runID, err := s.orch.ResumeTask(r.Context(), taskID)
+	switch {
+	case errors.Is(err, orchestrator.ErrNothingToResume):
+		http.Error(w, "no paused accounts to resume", http.StatusConflict)
+		return
+	case errors.Is(err, orchestrator.ErrNotTested):
+		http.Error(w, "accounts must pass connection tests first", http.StatusConflict)
+		return
+	case errors.Is(err, orchestrator.ErrAlreadyRunning):
+		http.Error(w, "task is already running", http.StatusConflict)
+		return
+	case err != nil:
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	writeJSON(w, http.StatusAccepted, map[string]int64{"run_id": runID})
+}
+
 func (s *Server) handleCancelAccount(w http.ResponseWriter, r *http.Request) {
 	taskID, err := pathID(r, "id")
 	if err != nil {

internal/orchestrator/orchestrator.go

@@ -17,6 +17,7 @@ import (
 var ErrNotTested = errors.New("accounts not fully tested")
 var ErrAlreadyRunning = errors.New("task already running")
 var ErrNoAccountsSelected = errors.New("no matching accounts selected")
+var ErrNothingToResume = errors.New("no paused accounts to resume")
 
 // maxAccountErrors caps how many individual error rows one account records per
 // run, so a corrupt mailbox producing thousands of failures can't bloat the
@@ -62,6 +63,38 @@ func planFolders(folders []string, mapping map[string]string, excluded []string)
 	return plan
 }
 
+// A run stops either on its own or because the operator intervened. Pausing and
+// cancelling take the same path — stop the in-flight work — and differ only in
+// the status left behind: paused accounts are what Resume picks up again.
+type stopReason int32
+
+const (
+	stopNone stopReason = iota
+	stopPaused
+	stopCancelled
+)
+
+func (r stopReason) accountStatus() string {
+	if r == stopPaused {
+		return "paused"
+	}
+	return "cancelled"
+}
+
+// runHandle is the live state of one task's run: the cancel that stops every
+// account under it, plus why it was stopped.
+type runHandle struct {
+	cancel context.CancelFunc
+	reason atomic.Int32
+}
+
+func (h *runHandle) stopWith(r stopReason) {
+	h.reason.CompareAndSwap(int32(stopNone), int32(r))
+	h.cancel()
+}
+
+func (h *runHandle) stopReason() stopReason { return stopReason(h.reason.Load()) }
+
 type Orchestrator struct {
 	store       *store.Store
 	hub         *wshub.Hub
@@ -70,10 +103,66 @@ type Orchestrator struct {
 
 	mu      sync.Mutex
 	cancels map[int64]context.CancelFunc // account_id -> cancel of its in-flight copy
+	runs    map[int64]*runHandle         // task_id -> live run
 }
 
 func New(s *store.Store, hub *wshub.Hub, encKey []byte, concurrency int) *Orchestrator {
-	return &Orchestrator{store: s, hub: hub, encKey: encKey, concurrency: concurrency, cancels: map[int64]context.CancelFunc{}}
+	return &Orchestrator{
+		store: s, hub: hub, encKey: encKey, concurrency: concurrency,
+		cancels: map[int64]context.CancelFunc{},
+		runs:    map[int64]*runHandle{},
+	}
+}
+
+// PauseTask stops the task's live run, leaving every unfinished account
+// "paused" so ResumeTask can pick them up. Returns false if nothing is running.
+func (o *Orchestrator) PauseTask(taskID int64) bool { return o.stopRun(taskID, stopPaused) }
+
+// CancelTask stops the task's live run and marks every unfinished account
+// "cancelled". Returns false if nothing is running.
+func (o *Orchestrator) CancelTask(taskID int64) bool { return o.stopRun(taskID, stopCancelled) }
+
+func (o *Orchestrator) stopRun(taskID int64, reason stopReason) bool {
+	o.mu.Lock()
+	h, ok := o.runs[taskID]
+	o.mu.Unlock()
+	if !ok {
+		return false
+	}
+	h.stopWith(reason)
+	return true
+}
+
+func (o *Orchestrator) registerRun(taskID int64, h *runHandle) {
+	o.mu.Lock()
+	o.runs[taskID] = h
+	o.mu.Unlock()
+}
+
+func (o *Orchestrator) unregisterRun(taskID int64) {
+	o.mu.Lock()
+	delete(o.runs, taskID)
+	o.mu.Unlock()
+}
+
+// ResumeTask restarts a paused task with exactly the accounts the pause left
+// unfinished. Everything already copied is skipped by the migration journal, so
+// each account continues where it stopped.
+func (o *Orchestrator) ResumeTask(ctx context.Context, taskID int64) (int64, error) {
+	accs, err := o.store.ListAccountsByTask(ctx, taskID)
+	if err != nil {
+		return 0, err
+	}
+	ids := make([]int64, 0, len(accs))
+	for _, a := range accs {
+		if a.Status == "paused" {
+			ids = append(ids, a.ID)
+		}
+	}
+	if len(ids) == 0 {
+		return 0, ErrNothingToResume
+	}
+	return o.Run(ctx, taskID, "manual", ids)
 }
 
 // CancelAccount aborts the in-flight copy for one account, if it is running.
@@ -231,11 +320,21 @@ func (o *Orchestrator) Run(ctx context.Context, taskID int64, trigger string, ac
 	}
 	o.hub.Publish(wshub.Event{Type: "run_started", TaskID: taskID, Data: map[string]any{"run_id": runID}})
 
-	go o.runAll(context.WithoutCancel(ctx), task, runID, accs, srcEP, dstEP, trigger)
+	// dbCtx outlives the request so status/counter writes still land after a
+	// pause or cancel; runCtx is what Pause/Cancel actually stop, and every
+	// account's IMAP work hangs off it.
+	dbCtx := context.WithoutCancel(ctx)
+	runCtx, runCancel := context.WithCancel(dbCtx)
+	h := &runHandle{cancel: runCancel}
+	o.registerRun(taskID, h)
+
+	go o.runAll(dbCtx, runCtx, h, task, runID, accs, srcEP, dstEP, trigger)
 	return runID, nil
 }
 
-func (o *Orchestrator) runAll(ctx context.Context, task store.Task, runID int64, accs []store.Account, srcEP, dstEP imapx.Endpoint, trigger string) {
+func (o *Orchestrator) runAll(ctx, runCtx context.Context, h *runHandle, task store.Task, runID int64, accs []store.Account, srcEP, dstEP imapx.Endpoint, trigger string) {
+	defer o.unregisterRun(task.ID)
+	defer h.cancel()
 	defer func() {
 		if r := recover(); r != nil {
 			slog.Error("run coordinator panicked", "task", task.ID, "run", runID, "panic", r)
@@ -256,7 +355,17 @@ func (o *Orchestrator) runAll(ctx context.Context, task store.Task, runID int64,
 	sem := make(chan struct{}, o.concurrency)
 	var wg sync.WaitGroup
 
-	for _, a := range accs {
+	for i, a := range accs {
+		// Stopped mid-queue: the accounts that never started are marked with the
+		// same status as the ones that were interrupted, so a pause leaves the
+		// whole remainder resumable and a cancel leaves it cancelled.
+		if runCtx.Err() != nil {
+			st := h.stopReason().accountStatus()
+			for _, rest := range accs[i:] {
+				_ = o.store.SetAccountStatus(ctx, rest.ID, st)
+			}
+			break
+		}
 		wg.Add(1)
 		sem <- struct{}{}
 		go func(a store.Account) {
@@ -273,7 +382,7 @@ func (o *Orchestrator) runAll(ctx context.Context, task store.Task, runID int64,
 					mu.Unlock()
 				}
 			}()
-			c, s, e := o.runAccount(ctx, task, runID, a, srcEP, dstEP)
+			c, s, e := o.runAccount(ctx, runCtx, h, task, runID, a, srcEP, dstEP)
 			mu.Lock()
 			totCopied += c
 			totSkipped += s
@@ -283,23 +392,32 @@ func (o *Orchestrator) runAll(ctx context.Context, task store.Task, runID int64,
 	}
 	wg.Wait()
 
+	reason := h.stopReason()
 	status := "done"
-	if totErr > 0 {
+	switch {
+	case reason == stopPaused:
+		status = "paused"
+	case reason == stopCancelled:
+		status = "cancelled"
+	case totErr > 0:
 		status = "done_with_errors"
 	}
 	_ = o.store.FinishRun(ctx, runID, status, totCopied, totSkipped, totErr)
 	_ = o.store.SetTaskStatus(ctx, task.ID, status)
 	o.hub.Publish(wshub.Event{Type: "run_done", TaskID: task.ID,
-		Data: map[string]any{"run_id": runID, "copied": totCopied, "skipped": totSkipped, "errors": totErr}})
+		Data: map[string]any{"run_id": runID, "status": status,
+			"copied": totCopied, "skipped": totSkipped, "errors": totErr}})
 
-	if shouldBreak(trigger, totErr) {
+	// An operator stopping the run is not a schedule failure, so leave the
+	// breaker alone even when the accounts that did run reported errors.
+	if reason == stopNone && shouldBreak(trigger, totErr) {
 		_ = o.store.SetTaskBroken(ctx, task.ID)
 		o.hub.Publish(wshub.Event{Type: "task_broken", TaskID: task.ID,
 			Data: map[string]any{"task_id": task.ID, "errors": totErr}})
 	}
 }
 
-func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint) (int64, int64, int64) {
+func (o *Orchestrator) runAccount(ctx, runCtx context.Context, h *runHandle, task store.Task, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint) (int64, int64, int64) {
 	o.hub.Publish(wshub.Event{Type: "account_started", TaskID: task.ID, Data: map[string]any{
 		"account_id": a.ID,
 		"src_login":  a.SrcLogin, "src_host": srcEP.Host, "src_port": srcEP.Port,
@@ -310,10 +428,11 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 	_ = o.store.ResetAccountCounters(ctx, a.ID) // start from zero; IncAccountCounters is additive
 	_ = o.store.ClearAccountErrors(ctx, a.ID)   // drop last run's per-error rows
 
-	// Per-account cancellable context: IMAP work uses actx (so CancelAccount
+	// Per-account cancellable context: IMAP work uses actx, so both CancelAccount
-	// stops it); DB writes keep the parent ctx so status/counters persist even
+	// and a task-wide pause/cancel (which cancels runCtx) stop it. DB writes keep
-	// after cancellation. ctx is context.WithoutCancel from runAll.
+	// ctx — the uncancellable one from runAll — so status/counters persist even
-	actx, cancel := context.WithCancel(ctx)
+	// after cancellation.
+	actx, cancel := context.WithCancel(runCtx)
 	o.registerCancel(a.ID, cancel)
 	defer func() {
 		o.unregisterCancel(a.ID)
@@ -322,20 +441,20 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 
 	srcPass, err := crypto.Decrypt(o.encKey, a.SrcPassEnc)
 	if err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
 	}
 	dstPass, err := crypto.Decrypt(o.encKey, a.DstPassEnc)
 	if err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "dst", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
 	}
 
 	src, err := imapx.Connect(actx, srcEP)
 	if err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
 	}
 	if err := src.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil {
 		_ = src.Logout().Wait()
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
 	}
 	// srcClient holds the LIVE source connection. A body-read timeout (server
 	// under-delivering a literal) forces a mid-run reconnect via reconnectSrc,
@@ -348,11 +467,11 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 
 	dst, err := imapx.Connect(actx, dstEP)
 	if err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "dst", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
 	}
 	defer func() { _ = dst.Logout().Wait() }()
 	if err := dst.Login(a.DstLogin, string(dstPass)).Wait(); err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "dst", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
 	}
 
 	// reconnectSrc dials and logs in a fresh source client, swaps it in as the
@@ -422,7 +541,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 	folders, err := imapx.ListFolders(src)
 	touch()
 	if err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err)
+		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
 	}
 
 	// Planning pass: decide folders from the account's own config, then EXAMINE
@@ -547,11 +666,18 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 	}
 
 	if actx.Err() != nil {
-		_ = o.store.SetAccountStatus(ctx, a.ID, "cancelled")
+		// A task-wide pause leaves the account resumable; anything else (per-account
-		o.hub.Publish(wshub.Event{Type: "cancelled", TaskID: task.ID,
+		// cancel, stall watchdog, task-wide cancel) leaves it cancelled.
+		st := "cancelled"
+		if runCtx.Err() != nil {
+			st = h.stopReason().accountStatus()
+		}
+		_ = o.store.SetAccountStatus(ctx, a.ID, st)
+		o.hub.Publish(wshub.Event{Type: st, TaskID: task.ID,
 			Data: map[string]any{"account_id": a.ID, "src_login": a.SrcLogin,
 				"copied": copied, "skipped": skipped, "errors": errs}})
-		slog.Info("account cancelled", "account", a.ID, "src_login", a.SrcLogin, "copied", copied, "skipped", skipped)
+		slog.Info("account stopped", "account", a.ID, "src_login", a.SrcLogin,
+			"status", st, "copied", copied, "skipped", skipped)
 		return copied, skipped, errs
 	}
 
@@ -567,11 +693,16 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 	return copied, skipped, errs
 }
 
-func (o *Orchestrator) accountFailed(ctx context.Context, taskID, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint, side string, err error) (int64, int64, int64) {
+func (o *Orchestrator) accountFailed(ctx, runCtx context.Context, h *runHandle, taskID, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint, side string, err error) (int64, int64, int64) {
-	// A cancellation surfacing as an error is a cancel, not a failure.
+	// A cancellation surfacing as an error is a stop, not a failure — and a
+	// task-wide pause must still leave the account resumable.
 	if errors.Is(err, context.Canceled) {
-		_ = o.store.SetAccountStatus(ctx, a.ID, "cancelled")
+		st := "cancelled"
-		o.hub.Publish(wshub.Event{Type: "cancelled", TaskID: taskID,
+		if runCtx.Err() != nil {
+			st = h.stopReason().accountStatus()
+		}
+		_ = o.store.SetAccountStatus(ctx, a.ID, st)
+		o.hub.Publish(wshub.Event{Type: st, TaskID: taskID,
 			Data: map[string]any{"account_id": a.ID, "src_login": a.SrcLogin}})
 		return 0, 0, 0
 	}

internal/orchestrator/stop_test.go

@@ -0,0 +1,83 @@
+package orchestrator
+
+import (
+	"context"
+	"testing"
+)
+
+func TestStopReasonAccountStatus(t *testing.T) {
+	if got := stopPaused.accountStatus(); got != "paused" {
+		t.Fatalf("stopPaused = %q want paused", got)
+	}
+	if got := stopCancelled.accountStatus(); got != "cancelled" {
+		t.Fatalf("stopCancelled = %q want cancelled", got)
+	}
+	// A run stopped without an operator reason (per-account cancel, stall
+	// watchdog) must not look like a pause, or Resume would pick it up.
+	if got := stopNone.accountStatus(); got != "cancelled" {
+		t.Fatalf("stopNone = %q want cancelled", got)
+	}
+}
+
+// The first stop wins: a cancel arriving after a pause must not downgrade the
+// accounts a pause already promised to keep resumable, and vice versa.
+func TestRunHandleFirstStopWins(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		first, later stopReason
+	}{
+		{"pause then cancel", stopPaused, stopCancelled},
+		{"cancel then pause", stopCancelled, stopPaused},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			h := &runHandle{cancel: cancel}
+			h.stopWith(tc.first)
+			h.stopWith(tc.later)
+			if got := h.stopReason(); got != tc.first {
+				t.Fatalf("reason = %v want %v", got, tc.first)
+			}
+		})
+	}
+}
+
+func TestRunHandleStopCancelsContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	h := &runHandle{cancel: cancel}
+	if ctx.Err() != nil {
+		t.Fatal("context cancelled before stop")
+	}
+	h.stopWith(stopPaused)
+	if ctx.Err() == nil {
+		t.Fatal("stop must cancel the run context")
+	}
+}
+
+// Pause/Cancel report false for a task with no live run, which the HTTP layer
+// turns into 409 instead of pretending it stopped something.
+func TestStopRunWithoutLiveRun(t *testing.T) {
+	o := &Orchestrator{runs: map[int64]*runHandle{}}
+	if o.PauseTask(1) {
+		t.Fatal("PauseTask must report false with no live run")
+	}
+	if o.CancelTask(1) {
+		t.Fatal("CancelTask must report false with no live run")
+	}
+
+	_, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	h := &runHandle{cancel: cancel}
+	o.registerRun(1, h)
+	if !o.PauseTask(1) {
+		t.Fatal("PauseTask must report true for a live run")
+	}
+	if got := h.stopReason(); got != stopPaused {
+		t.Fatalf("reason = %v want stopPaused", got)
+	}
+	o.unregisterRun(1)
+	if o.CancelTask(1) {
+		t.Fatal("unregistered run must not be stoppable")
+	}
+}

internal/store/accounts.go

@@ -32,6 +32,21 @@ func (s *Store) CreateAccount(ctx context.Context, a Account) (int64, error) {
 	return id, err
 }
 
+// UpdateAccountCredentials replaces an account's logins and, when a new
+// ciphertext is supplied, its passwords; a nil password keeps the stored one so
+// the operator can fix only the side that failed. Both connection tests are
+// reset to "unknown" because the previous verdicts no longer describe these
+// credentials, which also forces a re-test before the account can run.
+func (s *Store) UpdateAccountCredentials(ctx context.Context, id int64, srcLogin, dstLogin string, srcPassEnc, dstPassEnc *string) error {
+	_, err := s.Pool.Exec(ctx,
+		`UPDATE accounts SET src_login=$2, dst_login=$3,
+		 src_pass_enc=COALESCE($4, src_pass_enc), dst_pass_enc=COALESCE($5, dst_pass_enc),
+		 test_src_status='unknown', test_dst_status='unknown', last_error=''
+		 WHERE id=$1`,
+		id, srcLogin, dstLogin, srcPassEnc, dstPassEnc)
+	return err
+}
+
 // DeleteAccount removes one account (and its migrated_messages via ON DELETE CASCADE).
 func (s *Store) DeleteAccount(ctx context.Context, id int64) error {
 	_, err := s.Pool.Exec(ctx, `DELETE FROM accounts WHERE id=$1`, id)

internal/store/accounts_test.go

@@ -65,6 +65,47 @@ func TestResetAccountCounters(t *testing.T) {
 	}
 }
 
+// Fixing an imported account's credentials must replace only what the operator
+// supplied: a nil password keeps the stored ciphertext, and both test verdicts
+// go back to unknown because they described the old credentials.
+func TestUpdateAccountCredentials(t *testing.T) {
+	s := testStore(t)
+	ctx := context.Background()
+	epSrc, _ := s.CreateEndpoint(ctx, Endpoint{RoleLabel: "src", Host: "a", Port: 993, TLSMode: "ssl"})
+	epDst, _ := s.CreateEndpoint(ctx, Endpoint{RoleLabel: "dst", Host: "b", Port: 993, TLSMode: "ssl"})
+	taskID, _ := s.CreateTask(ctx, Task{Name: "t", SrcEndpointID: epSrc, DstEndpointID: epDst})
+	accID, _ := s.CreateAccount(ctx, Account{TaskID: taskID, SrcLogin: "u", SrcPassEnc: "oldsrc", DstLogin: "u2", DstPassEnc: "olddst"})
+	_ = s.SetAccountTestStatus(ctx, accID, "src", "fail")
+	_ = s.SetAccountTestStatus(ctx, accID, "dst", "ok")
+	_ = s.SetAccountError(ctx, accID, "authentication failed")
+
+	newSrc := "newsrc"
+	if err := s.UpdateAccountCredentials(ctx, accID, "u@src.example", "u@dst.example", &newSrc, nil); err != nil {
+		t.Fatalf("update: %v", err)
+	}
+
+	accs, _ := s.ListAccountsByTask(ctx, taskID)
+	if len(accs) != 1 {
+		t.Fatalf("len=%d want 1", len(accs))
+	}
+	a := accs[0]
+	if a.SrcLogin != "u@src.example" || a.DstLogin != "u@dst.example" {
+		t.Fatalf("logins not updated: %q / %q", a.SrcLogin, a.DstLogin)
+	}
+	if a.SrcPassEnc != "newsrc" {
+		t.Fatalf("src password not updated: %q", a.SrcPassEnc)
+	}
+	if a.DstPassEnc != "olddst" {
+		t.Fatalf("nil password must keep the stored one, got %q", a.DstPassEnc)
+	}
+	if a.TestSrcStatus != "unknown" || a.TestDstStatus != "unknown" {
+		t.Fatalf("test statuses not reset: %q / %q", a.TestSrcStatus, a.TestDstStatus)
+	}
+	if a.LastError != "" {
+		t.Fatalf("last_error not cleared: %q", a.LastError)
+	}
+}
+
 func TestSetAccountFolderMapping(t *testing.T) {
 	s := testStore(t)
 	ctx := context.Background()

internal/store/tasks.go

@@ -120,13 +120,16 @@ type SchedulableTask struct {
 }
 
 // ListSchedulableTasks returns tasks eligible to auto-run: schedule on, not
-// broken, not currently running — each joined with its last finished run time.
+// broken, neither running nor paused — each joined with its last finished run
+// time. A paused task waits for the operator to resume it; auto-starting a full
+// run behind their back would defeat the pause.
 func (s *Store) ListSchedulableTasks(ctx context.Context) ([]SchedulableTask, error) {
 	rows, err := s.Pool.Query(ctx,
 		`SELECT t.id, t.schedule_interval_seconds, t.schedule_anchor,
 		        (SELECT max(finished_at) FROM runs r WHERE r.task_id=t.id AND r.finished_at IS NOT NULL)
 		 FROM tasks t
-		 WHERE t.schedule_interval_seconds > 0 AND NOT t.broken AND t.status <> 'running'`)
+		 WHERE t.schedule_interval_seconds > 0 AND NOT t.broken
+		   AND t.status <> 'running' AND t.status <> 'paused'`)
 	if err != nil {
 		return nil, err
 	}

scripts/e2e.sh

@@ -158,7 +158,8 @@ wait_test_ok() {
 wait_test_ok
 
 wait_run_done() {
-	for ((i = 1; i <= 60; i++)); do
+	# Generous: the resume scenario re-scans and copies thousands of messages.
+	for ((i = 1; i <= 600; i++)); do
 		local status
 		status=$(api GET "/api/tasks/${TASK_ID}" | jq -r '.task.status')
 		if [[ "$status" == "done" ]]; then
@@ -185,16 +186,100 @@ log "POST /run (second run, expect idempotency)"
 api POST "/api/tasks/${TASK_ID}/run" >/dev/null
 wait_run_done
 
+# Counters are reset at the start of every run, so run 2's row shows run 2
+# alone: it must copy nothing and skip what run 1 already migrated.
 RES2=$(api GET "/api/tasks/${TASK_ID}")
-RUN2_COPIED_TOTAL=$(echo "$RES2" | jq -r '.accounts[0].copied')
+RUN2_COPIED=$(echo "$RES2" | jq -r '.accounts[0].copied')
-RUN2_SKIPPED_TOTAL=$(echo "$RES2" | jq -r '.accounts[0].skipped')
+RUN2_SKIPPED=$(echo "$RES2" | jq -r '.accounts[0].skipped')
 RUN2_ERRORS=$(echo "$RES2" | jq -r '.accounts[0].errors')
-RUN2_COPIED_DELTA=$((RUN2_COPIED_TOTAL - RUN1_COPIED))
+log "run 2: copied=$RUN2_COPIED skipped=$RUN2_SKIPPED errors=$RUN2_ERRORS"
-RUN2_SKIPPED_DELTA=$((RUN2_SKIPPED_TOTAL - RUN1_SKIPPED))
-log "run 2: copied_delta=$RUN2_COPIED_DELTA skipped_delta=$RUN2_SKIPPED_DELTA errors=$RUN2_ERRORS"
 
 [[ "$RUN2_ERRORS" == "0" ]] || fail "run 2 had errors"
-[[ "$RUN2_COPIED_DELTA" -eq 0 ]] || fail "run 2 copied $RUN2_COPIED_DELTA new messages (expected 0, not idempotent)"
+[[ "$RUN2_COPIED" -eq 0 ]] || fail "run 2 copied $RUN2_COPIED new messages (expected 0, not idempotent)"
-[[ "$RUN2_SKIPPED_DELTA" -gt 0 ]] || fail "run 2 skipped delta is $RUN2_SKIPPED_DELTA (expected >0)"
+[[ "$RUN2_SKIPPED" -eq "$RUN1_COPIED" ]] ||
+	fail "run 2 skipped $RUN2_SKIPPED of the $RUN1_COPIED messages run 1 copied"
 
-log "PASS: run1 copied=$RUN1_COPIED skipped=$RUN1_SKIPPED; run2 copied=$RUN2_COPIED_DELTA skipped=$RUN2_SKIPPED_DELTA (idempotent)"
+log "run1 copied=$RUN1_COPIED skipped=$RUN1_SKIPPED; run2 copied=$RUN2_COPIED skipped=$RUN2_SKIPPED (idempotent)"
+
+# ---------------------------------------------------------------------------
+# Pause / resume: a second account with enough messages that the run is still
+# in flight when the pause lands. Pausing must leave the account resumable, and
+# resuming must finish it without re-copying what the first stretch already did.
+# ---------------------------------------------------------------------------
+SRC_USER2="src2@example.com"
+DST_USER2="dst2@example.com"
+# Large enough that the copy is still in flight when the pause lands — greenmail
+# on a local socket copies well over a thousand small messages per second.
+SEED_COUNT=3000
+
+log "seeding ${SEED_COUNT} messages into ${SRC_USER2} INBOX (pause/resume scenario)"
+python3 "$SEED_PY" 127.0.0.1 3143 "$SRC_USER2" "$MAIL_PASS" "$SEED_COUNT"
+
+log "adding second account (src2 -> dst2)"
+ACCOUNT2_ID=$(api POST "/api/tasks/${TASK_ID}/accounts" \
+	"{\"src_login\":\"${SRC_USER2}\",\"src_pass\":\"${MAIL_PASS}\",\"dst_login\":\"${DST_USER2}\",\"dst_pass\":\"${MAIL_PASS}\"}" | jq -r .id)
+[[ "$ACCOUNT2_ID" =~ ^[0-9]+$ ]] || fail "bad second account id: $ACCOUNT2_ID"
+log "account2_id=$ACCOUNT2_ID"
+
+log "POST /test (both accounts)"
+api POST "/api/tasks/${TASK_ID}/test" >/dev/null
+for ((i = 1; i <= 30; i++)); do
+	BOTH_OK=$(api GET "/api/tasks/${TASK_ID}" |
+		jq -r '[.accounts[] | select(.test_src_status=="ok" and .test_dst_status=="ok")] | length')
+	[[ "$BOTH_OK" == "2" ]] && break
+	sleep 1
+done
+[[ "$BOTH_OK" == "2" ]] || fail "second account did not pass connection tests (ok count=$BOTH_OK)"
+
+# Account view for account2, by id.
+acct2() { api GET "/api/tasks/${TASK_ID}" | jq -r ".accounts[] | select(.id==${ACCOUNT2_ID}) | $1"; }
+
+log "POST /run (account2 only)"
+api POST "/api/tasks/${TASK_ID}/run" "{\"account_ids\":[${ACCOUNT2_ID}]}" >/dev/null
+
+# Per-account counters are only written to the DB when a folder completes, so
+# "copied so far" is invisible here — wait for the account to go running, give
+# the copy a few seconds of real work, then pause mid-folder.
+log "waiting for account2 to start running"
+for ((i = 1; i <= 120; i++)); do
+	[[ "$(acct2 .status)" == "running" ]] && break
+	sleep 0.5
+done
+[[ "$(acct2 .status)" == "running" ]] || fail "account2 never reached running"
+
+log "letting it copy for a few seconds, then pausing mid-folder"
+sleep 5
+curl -fsS -b "$COOKIE_JAR" -c "$COOKIE_JAR" -X POST "$BASE/api/tasks/${TASK_ID}/pause" >/dev/null ||
+	fail "pause rejected — the run finished before the pause landed, seed more messages"
+
+log "waiting for the task to settle into paused"
+for ((i = 1; i <= 60; i++)); do
+	TASK_STATUS=$(api GET "/api/tasks/${TASK_ID}" | jq -r '.task.status')
+	[[ "$TASK_STATUS" == "paused" ]] && break
+	sleep 1
+done
+[[ "$TASK_STATUS" == "paused" ]] || fail "task status=$TASK_STATUS after pause (expected paused)"
+
+ACC2_STATUS=$(acct2 .status)
+[[ "$ACC2_STATUS" == "paused" ]] || fail "account2 status=$ACC2_STATUS after pause (expected paused)"
+log "paused (folder-level counters at copied=$(acct2 .copied) of $SEED_COUNT)"
+
+log "POST /resume"
+api POST "/api/tasks/${TASK_ID}/resume" >/dev/null
+wait_run_done
+
+RESUMED_COPIED=$(acct2 .copied)
+RESUMED_SKIPPED=$(acct2 .skipped)
+RESUMED_ERRORS=$(acct2 .errors)
+RESUMED_TOTAL=$((RESUMED_COPIED + RESUMED_SKIPPED))
+log "after resume: copied=$RESUMED_COPIED skipped=$RESUMED_SKIPPED errors=$RESUMED_ERRORS"
+[[ "$RESUMED_ERRORS" == "0" ]] || fail "resumed run had errors"
+# Counters reset per run, so the resumed run alone must account for every
+# message: the ones it copied now plus the ones the paused stretch already did.
+[[ "$RESUMED_TOTAL" -eq "$SEED_COUNT" ]] || fail "resumed run covered $RESUMED_TOTAL of $SEED_COUNT messages"
+# Non-zero skipped is the proof that the paused stretch's work survived: those
+# messages are in the migration journal, so the resume did not re-copy them.
+[[ "$RESUMED_SKIPPED" -gt 0 ]] ||
+	fail "resumed run skipped nothing — the paused stretch's progress was lost"
+
+log "PASS: idempotent re-run; pause was resumable, resume re-copied $RESUMED_COPIED and skipped $RESUMED_SKIPPED of $SEED_COUNT"

web/src/api.ts

@@ -89,6 +89,14 @@ export const deleteTask = (id: number) => api(`/api/tasks/${id}`, { method: 'DEL
 export const deleteAccount = (taskId: number, accountId: number) =>
   api(`/api/tasks/${taskId}/accounts/${accountId}`, { method: 'DELETE' })
 
+// Empty password fields keep the stored ones; both connection tests reset to
+// unknown server-side, so the account must be re-tested afterwards.
+export const updateAccountCredentials = (
+  taskId: number,
+  accountId: number,
+  body: { src_login: string; src_pass: string; dst_login: string; dst_pass: string },
+) => api(`/api/tasks/${taskId}/accounts/${accountId}/credentials`, { ...jsonBody(body), method: 'PUT' })
+
 export const cancelAccount = (taskId: number, accountId: number) =>
   api(`/api/tasks/${taskId}/accounts/${accountId}/cancel`, { method: 'POST' })
 
@@ -143,6 +151,14 @@ export const testAccounts = (id: number) => api(`/api/tasks/${id}/test`, { metho
 export const runTask = (id: number, accountIds?: number[]) =>
   api(`/api/tasks/${id}/run`, accountIds?.length ? jsonBody({ account_ids: accountIds }) : { method: 'POST' })
 
+// Pause stops the run but leaves its unfinished accounts resumable; cancel ends
+// it and marks them cancelled. Resume re-runs exactly the paused accounts.
+export const pauseTask = (id: number) => api(`/api/tasks/${id}/pause`, { method: 'POST' })
+
+export const cancelTask = (id: number) => api(`/api/tasks/${id}/cancel`, { method: 'POST' })
+
+export const resumeTask = (id: number) => api<{ run_id: number }>(`/api/tasks/${id}/resume`, { method: 'POST' })
+
 export interface Run {
   id: number
   task_id: number

web/src/app.css

@@ -667,6 +667,25 @@ table.tbl a.rowlink:focus-visible {
 
 /* ---------- status badges ---------- */
 
+/* A badge that opens a dialog: the badge keeps its own look, the button only
+   contributes the affordance. */
+.badge-btn {
+  padding: 0;
+  border: 0;
+  background: none;
+  font: inherit;
+  cursor: pointer;
+}
+
+.badge-btn:hover .badge {
+  filter: brightness(1.25);
+}
+
+.badge-btn:focus-visible {
+  outline: 1px solid var(--accent);
+  outline-offset: 2px;
+}
+
 .badge {
   display: inline-flex;
   align-items: center;

web/src/components/AccountCredentialsModal.tsx

@@ -0,0 +1,116 @@
+import { useEffect, useState, type FormEvent } from 'react'
+import { Modal } from './Modal'
+import type { Account } from '../api'
+
+type Props = {
+  open: boolean
+  busy: boolean
+  account: Account | null
+  onClose: () => void
+  onSubmit: (body: { src_login: string; src_pass: string; dst_login: string; dst_pass: string }) => void
+}
+
+// Fixes the credentials of an account that failed its connection test — usually
+// a wrong password that came in through a CSV import. Passwords are never sent
+// back to the browser, so the fields start empty and an empty field means
+// "keep the stored password".
+export function AccountCredentialsModal({ open, busy, account, onClose, onSubmit }: Props) {
+  const [srcLogin, setSrcLogin] = useState('')
+  const [dstLogin, setDstLogin] = useState('')
+  const [srcPass, setSrcPass] = useState('')
+  const [dstPass, setDstPass] = useState('')
+  const [error, setError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (!open || !account) return
+    setSrcLogin(account.src_login)
+    setDstLogin(account.dst_login)
+    setSrcPass('')
+    setDstPass('')
+    setError(null)
+  }, [open, account])
+
+  function submit(e: FormEvent) {
+    e.preventDefault()
+    if (srcLogin.trim() === '' || dstLogin.trim() === '') {
+      setError('Both logins are required')
+      return
+    }
+    setError(null)
+    onSubmit({
+      src_login: srcLogin.trim(),
+      src_pass: srcPass,
+      dst_login: dstLogin.trim(),
+      dst_pass: dstPass,
+    })
+  }
+
+  return (
+    <Modal open={open} title={account ? `Edit credentials — ${account.src_login}` : 'Edit credentials'} onClose={onClose}>
+      <form onSubmit={submit}>
+        <p className="map-hint">
+          Leave a password field empty to keep the stored one. Saving resets both connection tests, so re-run{' '}
+          <strong>Test connections</strong> afterwards.
+        </p>
+        <div className="field-row">
+          <div className="field">
+            <label htmlFor="edit_src_login">Source login</label>
+            <input
+              id="edit_src_login"
+              data-modal-autofocus
+              value={srcLogin}
+              onChange={(e) => setSrcLogin(e.target.value)}
+              disabled={busy}
+              required
+            />
+          </div>
+          <div className="field">
+            <label htmlFor="edit_src_pass">Source password</label>
+            <input
+              id="edit_src_pass"
+              type="password"
+              value={srcPass}
+              onChange={(e) => setSrcPass(e.target.value)}
+              placeholder="unchanged"
+              autoComplete="new-password"
+              disabled={busy}
+            />
+          </div>
+        </div>
+        <div className="field-row">
+          <div className="field">
+            <label htmlFor="edit_dst_login">Destination login</label>
+            <input
+              id="edit_dst_login"
+              value={dstLogin}
+              onChange={(e) => setDstLogin(e.target.value)}
+              disabled={busy}
+              required
+            />
+          </div>
+          <div className="field">
+            <label htmlFor="edit_dst_pass">Destination password</label>
+            <input
+              id="edit_dst_pass"
+              type="password"
+              value={dstPass}
+              onChange={(e) => setDstPass(e.target.value)}
+              placeholder="unchanged"
+              autoComplete="new-password"
+              disabled={busy}
+            />
+          </div>
+        </div>
+        {error && <div className="error-banner">{error}</div>}
+        <div className="modal-actions">
+          <button type="button" className="btn" onClick={onClose} disabled={busy}>
+            Cancel
+          </button>
+          <button className="btn btn-primary" disabled={busy}>
+            {busy ? 'Saving…' : 'Save credentials'}
+          </button>
+        </div>
+      </form>
+    </Modal>
+  )
+}

web/src/pages/TaskDetail.tsx

@@ -1,5 +1,5 @@
 import { useEffect, useRef, useState, type ChangeEvent, type FormEvent } from 'react'
-import { cancelAccount, createAccount, deleteAccount, getTask, importCSV, importKerioCSV, probeAccountFolders, probeFolders, runTask, setAccountFolderMapping, setTaskSchedule, testAccounts, type TaskDetail as TaskDetailData } from '../api'
+import { cancelAccount, cancelTask, createAccount, deleteAccount, getTask, importCSV, importKerioCSV, pauseTask, probeAccountFolders, probeFolders, resumeTask, runTask, setAccountFolderMapping, setTaskSchedule, testAccounts, updateAccountCredentials, type Account, type TaskDetail as TaskDetailData } from '../api'
 import { connectTaskWS, type TaskEvent } from '../ws'
 import { StatusBadge } from '../components/StatusBadge'
 import { useConfirm } from '../components/ConfirmProvider'
@@ -7,6 +7,7 @@ import { FolderMappingModal } from '../components/FolderMappingModal'
 import { RunLogModal } from '../components/RunLogModal'
 import { AccountErrorsModal } from '../components/AccountErrorsModal'
 import { KerioImportModal } from '../components/KerioImportModal'
+import { AccountCredentialsModal } from '../components/AccountCredentialsModal'
 
 const emptyAccount = { src_login: '', src_pass: '', dst_login: '', dst_pass: '' }
 
@@ -58,6 +59,8 @@ function describeEvent(ev: TaskEvent): string {
     }
     case 'cancelled':
       return `CANCELLED #${d.account_id} (${d.src_login}): copied ${d.copied ?? 0}, skipped ${d.skipped ?? 0}`
+    case 'paused':
+      return `PAUSED #${d.account_id} (${d.src_login}): copied ${d.copied ?? 0}, skipped ${d.skipped ?? 0} — resumable`
     case 'error': {
       const where = d.folder ? ` folder "${d.folder}"` : d.side ? ` (${d.side} ${at})` : ''
       return `ERROR #${d.account_id}${where}: ${d.error}`
@@ -65,7 +68,7 @@ function describeEvent(ev: TaskEvent): string {
     case 'run_started':
       return `RUN started (run #${d.run_id})`
     case 'run_done':
-      return `RUN finished: copied ${d.copied}, skipped ${d.skipped}, errors ${d.errors}`
+      return `RUN ${String(d.status ?? 'finished')}: copied ${d.copied}, skipped ${d.skipped}, errors ${d.errors}`
     default:
       return JSON.stringify(ev.data)
   }
@@ -92,6 +95,7 @@ export function TaskDetail({ id }: { id: number }) {
   const [showRuns, setShowRuns] = useState(false)
   const [errorsFor, setErrorsFor] = useState<{ id: number; src_login: string } | null>(null)
   const [kerioOpen, setKerioOpen] = useState(false)
+  const [credsFor, setCredsFor] = useState<Account | null>(null)
   const [selected, setSelected] = useState<Set<number>>(new Set())
   const fileInputRef = useRef<HTMLInputElement>(null)
 
@@ -161,7 +165,7 @@ export function TaskDetail({ id }: { id: number }) {
               },
             }
           })
-        } else if (accId != null && (ev.type === 'account_started' || ev.type === 'account_done' || ev.type === 'cancelled' || (ev.type === 'error' && d.folder == null))) {
+        } else if (accId != null && (ev.type === 'account_started' || ev.type === 'account_done' || ev.type === 'cancelled' || ev.type === 'paused' || (ev.type === 'error' && d.folder == null))) {
           // terminal/reset for this account — drop live overlay, fall back to DB
           setLive((prev) => {
             if (!(accId in prev)) return prev
@@ -172,7 +176,7 @@ export function TaskDetail({ id }: { id: number }) {
         }
 
         // Structural events refresh the persisted view; `progress` is covered by live state.
-        if (['account_started', 'account_test', 'account_done', 'run_started', 'run_done', 'error', 'folder', 'cancelled', 'plan', 'task_broken'].includes(ev.type)) {
+        if (['account_started', 'account_test', 'account_done', 'run_started', 'run_done', 'error', 'folder', 'cancelled', 'paused', 'plan', 'task_broken'].includes(ev.type)) {
           reload()
         }
       }),
@@ -324,6 +328,21 @@ export function TaskDetail({ id }: { id: number }) {
     }
   }
 
+  async function saveCredentials(body: { src_login: string; src_pass: string; dst_login: string; dst_pass: string }) {
+    if (!credsFor) return
+    setBusy('add')
+    setError(null)
+    try {
+      await updateAccountCredentials(id, credsFor.id, body)
+      setCredsFor(null)
+      reload()
+    } catch (err) {
+      setError(err instanceof Error ? err.message : 'Failed to save credentials')
+    } finally {
+      setBusy(null)
+    }
+  }
+
   async function onDeleteAccount(accId: number, login: string) {
     const ok = await confirm({
       title: 'Remove account',
@@ -378,6 +397,51 @@ export function TaskDetail({ id }: { id: number }) {
     }
   }
 
+  async function onPause() {
+    setBusy('run')
+    setError(null)
+    try {
+      await pauseTask(id)
+    } catch (err) {
+      setError(err instanceof Error ? err.message : 'Failed to pause the run')
+    } finally {
+      setBusy(null)
+    }
+  }
+
+  async function onCancelRun() {
+    const ok = await confirm({
+      title: 'Cancel migration',
+      message: 'Stop the run and mark every unfinished account as cancelled? Copied messages are kept.',
+      confirmLabel: 'Cancel migration',
+      cancelLabel: 'Keep running',
+      danger: true,
+    })
+    if (!ok) return
+    setBusy('run')
+    setError(null)
+    try {
+      await cancelTask(id)
+    } catch (err) {
+      setError(err instanceof Error ? err.message : 'Failed to cancel the run')
+    } finally {
+      setBusy(null)
+    }
+  }
+
+  async function onResume() {
+    setBusy('run')
+    setError(null)
+    try {
+      await resumeTask(id)
+      reload()
+    } catch (err) {
+      setError(err instanceof Error ? err.message : 'Failed to resume the run')
+    } finally {
+      setBusy(null)
+    }
+  }
+
   async function onSchedule(intervalSeconds: number) {
     setError(null)
     try {
@@ -405,6 +469,24 @@ export function TaskDetail({ id }: { id: number }) {
 
   const { task, accounts } = data
   const isRunning = task.status === 'running'
+  // Accounts a pause left unfinished — what Resume picks up.
+  const pausedCount = accounts.filter((a) => a.status === 'paused').length
+
+  // A failed connection test is the entry point for fixing the credentials that
+  // caused it — an imported account is otherwise only deletable.
+  const testCell = (a: Account, status: string) =>
+    status === 'fail' && !isRunning && a.status !== 'running' ? (
+      <button
+        type="button"
+        className="badge-btn"
+        title="Edit credentials for this account"
+        onClick={() => setCredsFor(a)}
+      >
+        <StatusBadge status={status} />
+      </button>
+    ) : (
+      <StatusBadge status={status} />
+    )
   // A row is selectable only when both connection tests pass and no run is live.
   const selectableIds = accounts
     .filter((a) => a.test_src_status === 'ok' && a.test_dst_status === 'ok')
@@ -482,14 +564,37 @@ export function TaskDetail({ id }: { id: number }) {
           <button className="btn" onClick={onTest} disabled={busy !== null || accounts.length === 0}>
             {busy === 'test' ? 'Testing…' : 'Test connections'}
           </button>
-          <button className="btn btn-primary" onClick={onRun} disabled={busy !== null || !runReady || isRunning}>
+          {isRunning ? (
+            <>
+              <button className="btn" onClick={onPause} disabled={busy !== null}>
+                {busy === 'run' ? 'Stopping…' : 'Pause'}
+              </button>
+              <button className="btn btn-danger" onClick={onCancelRun} disabled={busy !== null}>
+                Cancel
+              </button>
+              <span className="hint">pause keeps the unfinished accounts resumable</span>
+            </>
+          ) : (
+            <>
+              {pausedCount > 0 && (
+                <button className="btn btn-primary" onClick={onResume} disabled={busy !== null}>
+                  {busy === 'run' ? 'Resuming…' : `Resume (${pausedCount})`}
+                </button>
+              )}
+              <button
+                className={pausedCount > 0 ? 'btn' : 'btn btn-primary'}
+                onClick={onRun}
+                disabled={busy !== null || !runReady}
+              >
                 {busy === 'run'
                   ? 'Starting…'
                   : effectiveSelected.length > 0
                     ? `Run selected (${effectiveSelected.length})`
                     : 'Run migration'}
               </button>
-          {!runReady && accounts.length > 0 && (
+            </>
+          )}
+          {!isRunning && !runReady && accounts.length > 0 && (
             <span className="hint">
               {effectiveSelected.length > 0
                 ? 'selected accounts must pass both connection tests'
@@ -677,12 +782,8 @@ export function TaskDetail({ id }: { id: number }) {
                         </div>
                       )}
                     </td>
-                    <td>
+                    <td>{testCell(a, a.test_src_status)}</td>
-                      <StatusBadge status={a.test_src_status} />
+                    <td>{testCell(a, a.test_dst_status)}</td>
-                    </td>
-                    <td>
-                      <StatusBadge status={a.test_dst_status} />
-                    </td>
                     <td>
                       <StatusBadge status={a.status} />
                     </td>
@@ -820,6 +921,13 @@ export function TaskDetail({ id }: { id: number }) {
         onClose={() => setKerioOpen(false)}
         onSubmit={onKerioImport}
       />
+      <AccountCredentialsModal
+        open={credsFor !== null}
+        busy={busy === 'add'}
+        account={credsFor}
+        onClose={() => setCredsFor(null)}
+        onSubmit={saveCredentials}
+      />
     </>
   )
 }