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>

.gitignore

@@ -10,3 +10,4 @@
 # local cache of the impeccable design hook
 .impeccable/
 **/.impeccable/
+*.csv

docker-compose-traefik.yml

@@ -8,7 +8,7 @@ services:
     volumes:
       - pgdata:/var/lib/postgresql
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U imap"]
+      test: ["CMD-SHELL", "pg_isready -U imap -d imapcopier"]
       interval: 5s
       timeout: 3s
       retries: 5

docker-compose.yml

@@ -8,7 +8,7 @@ services:
     volumes:
       - pgdata:/var/lib/postgresql
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U imap"]
+      test: ["CMD-SHELL", "pg_isready -U imap -d imapcopier"]
       interval: 5s
       timeout: 3s
       retries: 5

internal/csvimport/csvimport.go

@@ -1,9 +1,11 @@
 package csvimport
 
 import (
+	"bufio"
 	"encoding/csv"
 	"fmt"
 	"io"
+	"regexp"
 	"strings"
 )
 
@@ -52,3 +54,110 @@ func Parse(r io.Reader) ([]Row, error) {
 	}
 	return rows, nil
 }
+
+// Kerio Connect exports users as a semicolon-separated file whose first line is
+// a header. Only these columns matter; the rest (quotas, last login, …) is
+// ignored. The file carries no domain — the caller supplies it.
+const (
+	kerioColName        = 0 // login without the domain part
+	kerioColDescription = 2 // Kerio keeps the plaintext password here
+	kerioColEnable      = 3 // "Yes" / "No"
+	kerioMinColumns     = 4
+)
+
+// domainRe accepts a bare DNS domain: labels of alphanumerics/hyphens with at
+// least one dot and no scheme, user part, or whitespace.
+var domainRe = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+$`)
+
+func normalizeDomain(domain string) (string, error) {
+	d := strings.ToLower(strings.TrimSpace(domain))
+	if d == "" {
+		return "", fmt.Errorf("domain is required")
+	}
+	if !domainRe.MatchString(d) {
+		return "", fmt.Errorf("invalid domain %q", domain)
+	}
+	return d, nil
+}
+
+// skipBOM consumes a leading UTF-8 byte-order mark, which Kerio writes into its
+// exports and encoding/csv would otherwise glue onto the first header field.
+func skipBOM(br *bufio.Reader) error {
+	b, err := br.Peek(3)
+	if err != nil && err != io.EOF {
+		return err
+	}
+	if len(b) == 3 && b[0] == 0xEF && b[1] == 0xBB && b[2] == 0xBF {
+		_, _ = br.Discard(3)
+	}
+	return nil
+}
+
+// ParseKerio reads a Kerio Connect user export and maps every enabled, non-admin
+// account onto both sides of a migration: the login and password are identical
+// on source and destination, only the server differs. Rows for disabled accounts
+// and the built-in admin are skipped.
+func ParseKerio(r io.Reader, domain string) ([]Row, error) {
+	d, err := normalizeDomain(domain)
+	if err != nil {
+		return nil, err
+	}
+
+	br := bufio.NewReader(r)
+	if err := skipBOM(br); err != nil {
+		return nil, err
+	}
+
+	cr := csv.NewReader(br)
+	cr.Comma = ';'
+	cr.FieldsPerRecord = -1 // проверяем сами
+	cr.LazyQuotes = true    // FullName нередко содержит одиночную кавычку
+
+	header, err := cr.Read()
+	if err != nil {
+		return nil, fmt.Errorf("cannot read header: %w", err)
+	}
+	if len(header) < kerioMinColumns || !strings.EqualFold(strings.TrimSpace(header[kerioColName]), "Name") {
+		return nil, fmt.Errorf("not a Kerio export: expected a header starting with Name;FullName;Description;Enable")
+	}
+
+	var rows []Row
+	seen := map[string]bool{}
+	for {
+		rec, err := cr.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, err
+		}
+		line, _ := cr.FieldPos(0)
+		if len(rec) == 1 && strings.TrimSpace(rec[0]) == "" {
+			continue
+		}
+		if len(rec) < kerioMinColumns {
+			return nil, fmt.Errorf("line %d: expected at least %d columns, got %d", line, kerioMinColumns, len(rec))
+		}
+		name := strings.ToLower(strings.TrimSpace(rec[kerioColName]))
+		if name == "" {
+			return nil, fmt.Errorf("line %d: Name is empty", line)
+		}
+		if !strings.EqualFold(strings.TrimSpace(rec[kerioColEnable]), "Yes") || name == "admin" {
+			continue
+		}
+		pass := strings.TrimSpace(rec[kerioColDescription])
+		if pass == "" {
+			return nil, fmt.Errorf("line %d: no password in the Description column for %q", line, name)
+		}
+		if seen[name] {
+			return nil, fmt.Errorf("line %d: duplicate Name %q", line, name)
+		}
+		seen[name] = true
+		login := name + "@" + d
+		rows = append(rows, Row{SrcLogin: login, SrcPass: pass, DstLogin: login, DstPass: pass})
+	}
+	if len(rows) == 0 {
+		return nil, fmt.Errorf("no enabled accounts found in the export")
+	}
+	return rows, nil
+}

internal/csvimport/csvimport_test.go

@@ -49,3 +49,100 @@ func TestParseZeroRowsErrors(t *testing.T) {
 		t.Fatal("expected error when no rows parsed")
 	}
 }
+
+const kerioHeader = "Name;FullName;Description;Enable;DataSource;Authentication;Role;Groups;MailAddress\n"
+
+func TestParseKerioOK(t *testing.T) {
+	in := kerioHeader +
+		"j.doe;Jane Doe;SrcPass11;Yes;Internal;Internal;No rights;all;j.doe\n" +
+		"k.smith;Kim Smith;SrcPass22;Yes;Internal;Internal;No rights;all;k.smith\n"
+	rows, err := ParseKerio(strings.NewReader(in), "example.test")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("want 2 rows, got %d: %+v", len(rows), rows)
+	}
+	want := Row{
+		SrcLogin: "j.doe@example.test", SrcPass: "SrcPass11",
+		DstLogin: "j.doe@example.test", DstPass: "SrcPass11",
+	}
+	if rows[0] != want {
+		t.Fatalf("row 0: got %+v, want %+v", rows[0], want)
+	}
+}
+
+func TestParseKerioSkipsDisabledAndAdmin(t *testing.T) {
+	in := kerioHeader +
+		"admin;;AdminPass1;Yes;Internal;Internal;Account admin;;admin\n" +
+		"o.disabled;;OffPass111;No;Internal;Internal;No rights;all;o.disabled\n" +
+		"info;;InfoPass11;Yes;Internal;Internal;No rights;all;info\n"
+	rows, err := ParseKerio(strings.NewReader(in), "example.test")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(rows) != 1 || rows[0].SrcLogin != "info@example.test" {
+		t.Fatalf("only the enabled non-admin row must survive, got %+v", rows)
+	}
+}
+
+func TestParseKerioStripsBOM(t *testing.T) {
+	in := "\ufeff" + kerioHeader + "info;;InfoPass11;Yes;Internal;Internal;No rights;all;info\n"
+	rows, err := ParseKerio(strings.NewReader(in), "example.test")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("want 1 row, got %+v", rows)
+	}
+}
+
+func TestParseKerioRejectsMissingHeader(t *testing.T) {
+	in := "j.doe;Jane Doe;SrcPass11;Yes;Internal;Internal;No rights;all;j.doe\n"
+	if _, err := ParseKerio(strings.NewReader(in), "example.test"); err == nil {
+		t.Fatal("a file without the Kerio header must error")
+	}
+}
+
+func TestParseKerioRejectsBadDomain(t *testing.T) {
+	in := kerioHeader + "info;;InfoPass11;Yes;Internal;Internal;No rights;all;info\n"
+	for _, domain := range []string{"", "   ", "@example.test", "exam ple.test", "example", "info@example.test"} {
+		if _, err := ParseKerio(strings.NewReader(in), domain); err == nil {
+			t.Fatalf("domain %q must be rejected", domain)
+		}
+	}
+}
+
+func TestParseKerioTrimsAndLowercasesDomain(t *testing.T) {
+	in := kerioHeader + "info;;InfoPass11;Yes;Internal;Internal;No rights;all;info\n"
+	rows, err := ParseKerio(strings.NewReader(in), "  Example.TEST  ")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if rows[0].SrcLogin != "info@example.test" {
+		t.Fatalf("domain must be trimmed and lowercased, got %q", rows[0].SrcLogin)
+	}
+}
+
+func TestParseKerioRejectsEmptyPassword(t *testing.T) {
+	in := kerioHeader + "info;;;Yes;Internal;Internal;No rights;all;info\n"
+	if _, err := ParseKerio(strings.NewReader(in), "example.test"); err == nil {
+		t.Fatal("an enabled account without a password must error")
+	}
+}
+
+func TestParseKerioRejectsDuplicateName(t *testing.T) {
+	in := kerioHeader +
+		"info;;InfoPass11;Yes;Internal;Internal;No rights;all;info\n" +
+		"info;;aaaaaaAa1;Yes;Internal;Internal;No rights;all;info\n"
+	if _, err := ParseKerio(strings.NewReader(in), "example.test"); err == nil {
+		t.Fatal("duplicate Name must error")
+	}
+}
+
+func TestParseKerioZeroRowsErrors(t *testing.T) {
+	in := kerioHeader + "admin;;AdminPass1;Yes;Internal;Internal;Account admin;;admin\n"
+	if _, err := ParseKerio(strings.NewReader(in), "example.test"); err == nil {
+		t.Fatal("expected error when every row is filtered out")
+	}
+}

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/endpoints.go

@@ -2,8 +2,11 @@ package httpapi
 
 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"net/http"
 
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/vasyansk/imap-copier/internal/store"
 )
 
@@ -54,6 +57,36 @@ func (s *Server) handleUpdateEndpoint(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
+// handleDeleteEndpoint removes an endpoint that no task references. A referenced
+// endpoint is refused with 409 rather than a foreign-key error, and the same
+// status covers the race where a task is created between check and delete.
+func (s *Server) handleDeleteEndpoint(w http.ResponseWriter, r *http.Request) {
+	id, err := pathID(r, "id")
+	if err != nil {
+		http.Error(w, "bad id", http.StatusBadRequest)
+		return
+	}
+	used, err := s.store.CountTasksUsingEndpoint(r.Context(), id)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if used > 0 {
+		http.Error(w, fmt.Sprintf("endpoint is used by %d task(s) — delete them first", used), http.StatusConflict)
+		return
+	}
+	if err := s.store.DeleteEndpoint(r.Context(), id); err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && pgErr.Code == "23503" {
+			http.Error(w, "endpoint is used by a task — delete it first", http.StatusConflict)
+			return
+		}
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
 func (s *Server) handleListEndpoints(w http.ResponseWriter, r *http.Request) {
 	eps, err := s.store.ListEndpoints(r.Context())
 	if err != nil {

internal/httpapi/router.go

@@ -15,6 +15,7 @@ func (s *Server) Router() http.Handler {
 	api.HandleFunc("GET /api/endpoints", s.handleListEndpoints)
 	api.HandleFunc("POST /api/endpoints", s.handleCreateEndpoint)
 	api.HandleFunc("PUT /api/endpoints/{id}", s.handleUpdateEndpoint)
+	api.HandleFunc("DELETE /api/endpoints/{id}", s.handleDeleteEndpoint)
 	api.HandleFunc("GET /api/tasks", s.handleListTasks)
 	api.HandleFunc("POST /api/tasks", s.handleCreateTask)
 	api.HandleFunc("GET /api/tasks/{id}", s.handleGetTask)
@@ -26,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

@@ -14,6 +14,17 @@ import (
 	"github.com/vasyansk/imap-copier/internal/store"
 )
 
+// parseImportRows picks the CSV dialect from the "format" form field. A Kerio
+// Connect export holds one login/password pair and no domain, so the operator
+// supplies the domain alongside the file; anything else is the plain 4-column
+// src/dst format.
+func parseImportRows(r *http.Request, file io.Reader) ([]csvimport.Row, error) {
+	if r.FormValue("format") == "kerio" {
+		return csvimport.ParseKerio(file, r.FormValue("domain"))
+	}
+	return csvimport.Parse(file)
+}
+
 func (s *Server) handleImportCSV(w http.ResponseWriter, r *http.Request) {
 	taskID, err := pathID(r, "id")
 	if err != nil {
@@ -26,7 +37,7 @@ func (s *Server) handleImportCSV(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer file.Close()
-	rows, err := csvimport.Parse(file)
+	rows, err := parseImportRows(r, file)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
@@ -116,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/httpapi/run_test.go

@@ -1,7 +1,9 @@
 package httpapi
 
 import (
+	"io"
 	"mime/multipart"
+	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
@@ -27,6 +29,60 @@ func TestImportCSVFailsOnBadEncKey(t *testing.T) {
 	}
 }
 
+// importReq builds a multipart import request with the given CSV payload and
+// extra form fields, then returns it with the uploaded file ready to read.
+func importReq(t *testing.T, csv string, fields map[string]string) (*http.Request, io.Reader) {
+	t.Helper()
+	body := &strings.Builder{}
+	mw := multipart.NewWriter(body)
+	for k, v := range fields {
+		_ = mw.WriteField(k, v)
+	}
+	fw, _ := mw.CreateFormFile("file", "a.csv")
+	fw.Write([]byte(csv))
+	mw.Close()
+	req := httptest.NewRequest("POST", "/api/tasks/1/import", strings.NewReader(body.String()))
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+	req.SetPathValue("id", "1")
+	file, _, err := req.FormFile("file")
+	if err != nil {
+		t.Fatalf("form file: %v", err)
+	}
+	return req, file
+}
+
+const kerioCSV = "Name;FullName;Description;Enable;DataSource\n" +
+	"info;;InfoPass11;Yes;Internal\n"
+
+func TestParseImportRowsUsesKerioParserWithDomain(t *testing.T) {
+	req, file := importReq(t, kerioCSV, map[string]string{"format": "kerio", "domain": "example.test"})
+	rows, err := parseImportRows(req, file)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(rows) != 1 || rows[0].SrcLogin != "info@example.test" || rows[0].DstLogin != "info@example.test" {
+		t.Fatalf("kerio rows must carry the supplied domain on both sides, got %+v", rows)
+	}
+}
+
+func TestParseImportRowsKerioRequiresDomain(t *testing.T) {
+	req, file := importReq(t, kerioCSV, map[string]string{"format": "kerio"})
+	if _, err := parseImportRows(req, file); err == nil {
+		t.Fatal("kerio import without a domain must error")
+	}
+}
+
+func TestParseImportRowsDefaultsToPlainFormat(t *testing.T) {
+	req, file := importReq(t, "a@x,p1,a@y,p2\n", nil)
+	rows, err := parseImportRows(req, file)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(rows) != 1 || rows[0].SrcLogin != "a@x" || rows[0].DstPass != "p2" {
+		t.Fatalf("no format field must keep the 4-column parser, got %+v", rows)
+	}
+}
+
 func TestParseRunAccountIDs(t *testing.T) {
 	// empty body => nil (run all)
 	req := httptest.NewRequest("POST", "/api/tasks/1/run", strings.NewReader(""))

internal/imapx/account.go

@@ -4,13 +4,12 @@ import (
 	"context"
 
 	"github.com/emersion/go-imap/v2"
-	"github.com/emersion/go-imap/v2/imapclient"
 )
 
 // FolderMessageCount opens a folder read-only (EXAMINE) and returns how many
 // messages it holds — used to plan an accurate overall progress total before
 // copying begins. It does not fetch any message bodies.
-func FolderMessageCount(c *imapclient.Client, folder string) (int64, error) {
+func FolderMessageCount(c *Client, folder string) (int64, error) {
 	sel, err := c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait()
 	if err != nil {
 		return 0, err
@@ -19,7 +18,7 @@ func FolderMessageCount(c *imapclient.Client, folder string) (int64, error) {
 }
 
 // ListFolders returns the mailbox names visible on an already-connected, logged-in client.
-func ListFolders(c *imapclient.Client) ([]string, error) {
+func ListFolders(c *Client) ([]string, error) {
 	mboxes, err := c.List("", "*", nil).Collect()
 	if err != nil {
 		return nil, err

internal/imapx/copy.go

@@ -6,7 +6,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net"
+	"os"
 	"strings"
 	"time"
 
@@ -35,8 +37,37 @@ type CopyDeps struct {
 	// text — so the orchestrator can persist individual errors for the
 	// per-account error modal. Folder-level errors are reported by the caller.
 	OnError func(ref, msg string)
+	// OnActivity is called repeatedly WHILE a single message body is streamed
+	// (each FETCH read chunk and each APPEND write chunk). Copying one large
+	// message can take longer than the orchestrator's stall timeout; without an
+	// in-body signal the watchdog can't tell a slow-but-live transfer from a
+	// wedged connection and cancels a healthy copy. May be nil.
+	OnActivity func()
+	// ReconnectSrc dials and logs in a FRESH source client, returning it. It is
+	// called when a body read times out (server under-delivered a literal),
+	// which leaves the current src connection desynced and unusable. CopyFolder
+	// swaps to the returned client, re-EXAMINEs the folder, and resumes. The
+	// implementation is expected to also update any external reference to the
+	// live src client (e.g. so a cancel path closes the right connection). If
+	// nil, a body-read timeout aborts the folder instead of recovering.
+	ReconnectSrc func() (*Client, error)
 }
 
+// ErrBodyTimeout means a message body did not finish transferring within the
+// idle deadline — the server stopped sending mid-literal. It is almost always a
+// server announcing a BODY[] literal larger than the bytes it actually sends,
+// which makes go-imap wait forever for bytes that never come. Distinct from a
+// closed connection so the caller can skip just this one message and resume.
+var ErrBodyTimeout = errors.New("message body read timed out")
+
+// bodyIdleTimeout bounds how long a body read may go with NO bytes arriving
+// before it is abandoned. It is generous enough for legitimately slow servers
+// (even ~15 KB/s links keep bytes flowing far more often than this) yet well
+// under the orchestrator's multi-minute stall watchdog, so an under-delivered
+// literal is caught quickly and locally instead of stalling the whole account.
+// A var (not const) so tests can shorten it.
+var bodyIdleTimeout = 30 * time.Second
+
 // CopyResult summarizes the outcome of one CopyFolder run.
 type CopyResult struct {
 	Copied  int
@@ -88,7 +119,7 @@ func metaBatches(total, batchSize uint32) []imap.SeqRange {
 // held in memory only for the duration of a single FETCH->APPEND and is
 // never written to disk. Messages already migrated (per deps.IsMigrated)
 // are skipped without re-fetching their bodies.
-func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dstFolder string, deps CopyDeps) (CopyResult, error) {
+func CopyFolder(ctx context.Context, src, dst *Client, srcFolder, dstFolder string, deps CopyDeps) (CopyResult, error) {
 	var res CopyResult
 
 	sel, err := src.Select(srcFolder, &imap.SelectOptions{ReadOnly: true}).Wait()
@@ -180,9 +211,28 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst
 		if err := ctx.Err(); err != nil {
 			return res, err
 		}
-		if err := streamOne(src, dst, dstFolder, q.uid, q.flags, q.internalDate); err != nil {
+		if err := streamOne(src, dst, dstFolder, q.uid, q.flags, q.internalDate, deps.OnActivity); err != nil {
 			res.Errors++
 			reportErr(msgRef(q.uid, q.subject), "copy message: "+err.Error())
+			// A body-read timeout means the server under-delivered this message's
+			// literal; the src connection is now desynced. Mark the message
+			// migrated so this and future runs skip it (it is un-fetchable via a
+			// conforming client), then reconnect src and resume the folder with
+			// the remaining queued messages.
+			if errors.Is(err, ErrBodyTimeout) && deps.ReconnectSrc != nil {
+				if merr := deps.MarkMigrated(dstFolder, q.key); merr != nil {
+					reportErr(msgRef(q.uid, q.subject), "mark skipped: "+merr.Error())
+				}
+				newSrc, rerr := deps.ReconnectSrc()
+				if rerr != nil {
+					return res, fmt.Errorf("reconnect src after body timeout in %q: %w", srcFolder, rerr)
+				}
+				src = newSrc
+				if _, serr := src.Select(srcFolder, &imap.SelectOptions{ReadOnly: true}).Wait(); serr != nil {
+					return res, fmt.Errorf("re-examine %q after reconnect: %w", srcFolder, serr)
+				}
+				continue
+			}
 			// A closed/broken connection won't recover: every remaining APPEND
 			// would fail identically. Abort the folder instead of logging
 			// thousands of the same error; a re-run resumes via dedup.
@@ -204,11 +254,63 @@ func CopyFolder(ctx context.Context, src, dst *imapclient.Client, srcFolder, dst
 	return res, nil
 }
 
+// slowMessage marks how long one message's FETCH or APPEND phase may take
+// before it is logged as anomalous. Well below the orchestrator's 3-minute
+// stall timeout, so a message that trips the watchdog always leaves a log line
+// naming the phase (FETCH vs APPEND) and size — turning a silent stall into
+// evidence of which side and which message is the culprit.
+const slowMessage = 20 * time.Second
+
+// deadlineReader arms an idle read deadline on the source socket before every
+// read of a message body, so a server that stops sending mid-literal — having
+// announced a larger BODY[] size than it actually delivers — trips the deadline
+// instead of blocking go-imap forever waiting for bytes that never arrive. It
+// also pings onActivity as bytes arrive, feeding the orchestrator's stall
+// watchdog. The deadline is refreshed on each read, so it bounds IDLE time
+// (no bytes) rather than total transfer time — a legitimately slow but steady
+// download never trips it.
+type deadlineReader struct {
+	c    *Client
+	r    io.Reader
+	idle time.Duration
+	on   func()
+}
+
+func (d deadlineReader) Read(p []byte) (int, error) {
+	_ = d.c.SetReadDeadline(time.Now().Add(d.idle))
+	n, err := d.r.Read(p)
+	if n > 0 && d.on != nil {
+		d.on()
+	}
+	return n, err
+}
+
+// touchWriter wraps the APPEND write stream and pings onActivity on every
+// non-empty write, so a long upload keeps the stall watchdog fed byte-by-byte.
+type touchWriter struct {
+	w  io.Writer
+	on func()
+}
+
+func (t touchWriter) Write(p []byte) (int, error) {
+	n, err := t.w.Write(p)
+	if n > 0 && t.on != nil {
+		t.on()
+	}
+	return n, err
+}
+
 // streamOne FETCHes BODY[] for one message and APPENDs it into dst without
 // spooling to disk. The body is buffered in RAM only for the duration of
-// this single FETCH->APPEND round trip.
-func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flags []imap.Flag, internalDate time.Time) error {
+// this single FETCH->APPEND round trip. onActivity (may be nil) fires as bytes
+// move in either direction, feeding the orchestrator's stall watchdog.
+//
+// The body read is guarded by an idle deadline on src: if the server goes
+// silent mid-literal, streamOne returns ErrBodyTimeout rather than hanging, so
+// CopyFolder can skip the message and reconnect.
+func streamOne(src, dst *Client, dstFolder string, uid imap.UID, flags []imap.Flag, internalDate time.Time, onActivity func()) error {
 	bodySection := &imap.FetchItemBodySection{}
+	fetchStart := time.Now()
 	fetchCmd := src.Fetch(imap.UIDSetNum(uid), &imap.FetchOptions{
 		BodySection: []*imap.FetchItemBodySection{bodySection},
 	})
@@ -219,33 +321,41 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag
 		return fmt.Errorf("no message for uid %v", uid)
 	}
 	var body []byte
+	var readErr error
 	for {
 		item := msg.Next()
 		if item == nil {
 			break
 		}
 		if d, ok := item.(imapclient.FetchItemDataBodySection); ok {
-			b, err := io.ReadAll(d.Literal)
-			if err != nil {
-				return err
-			}
-			body = b
+			body, readErr = io.ReadAll(deadlineReader{c: src, r: d.Literal, idle: bodyIdleTimeout, on: onActivity})
 		}
 	}
+	// Clear the deadline before any further I/O on src (fetchCmd.Close reads the
+	// command's completion off the same socket).
+	_ = src.SetReadDeadline(time.Time{})
+	if readErr != nil {
+		if errors.Is(readErr, os.ErrDeadlineExceeded) {
+			return fmt.Errorf("%w: uid %v (server sent fewer bytes than the announced literal)", ErrBodyTimeout, uid)
+		}
+		return readErr
+	}
 	if err := fetchCmd.Close(); err != nil {
 		return err
 	}
 	if body == nil {
 		return fmt.Errorf("empty body uid %v", uid)
 	}
+	fetchDur := time.Since(fetchStart)
 
+	appendStart := time.Now()
 	appendCmd := dst.Append(dstFolder, int64(len(body)), &imap.AppendOptions{Flags: keepFlags(flags), Time: internalDate})
 	// Append acquires go-imap's per-client encoder mutex and holds it until
 	// Close() calls enc.end(). Close() MUST run on every path: if io.Copy
 	// fails mid-write (server stall, idle timeout), returning without Close()
 	// leaks the mutex and the NEXT Append on this client deadlocks forever on
 	// beginCommand. Close() is idempotent and always releases the lock.
-	_, copyErr := io.Copy(appendCmd, bytes.NewReader(body))
+	_, copyErr := io.Copy(touchWriter{w: appendCmd, on: onActivity}, bytes.NewReader(body))
 	closeErr := appendCmd.Close()
 	if copyErr != nil {
 		return fmt.Errorf("append body uid %v: %w", uid, copyErr)
@@ -253,8 +363,19 @@ func streamOne(src, dst *imapclient.Client, dstFolder string, uid imap.UID, flag
 	if closeErr != nil {
 		return closeErr
 	}
-	_, err := appendCmd.Wait()
-	return err
+	if _, err := appendCmd.Wait(); err != nil {
+		return err
+	}
+	appendDur := time.Since(appendStart)
+
+	// One message that individually eats a large slice of the stall budget is
+	// the prime suspect behind a "no progress" cancel; name it, its size, and
+	// which phase was slow so the culprit is visible in the logs.
+	if fetchDur > slowMessage || appendDur > slowMessage {
+		slog.Warn("slow message copy", "uid", uid, "bytes", len(body),
+			"fetch", fetchDur.Round(time.Millisecond), "append", appendDur.Round(time.Millisecond))
+	}
+	return nil
 }
 
 // keepFlags drops \Recent: it cannot be set via APPEND. go-imap v2 beta.8

internal/imapx/copy_test.go

@@ -169,6 +169,56 @@ func TestCopyFolderPreservesInternalDate(t *testing.T) {
 	}
 }
 
+// TestCopyFolderReportsActivityDuringBody proves CopyFolder invokes OnActivity
+// while a message body is being transferred (FETCH/APPEND), not only between
+// messages. This is what keeps the orchestrator's stall watchdog from killing a
+// single large-but-live message whose transfer legitimately exceeds the stall
+// timeout: without an in-body activity signal, one slow message looks identical
+// to a wedged connection.
+func TestCopyFolderReportsActivityDuringBody(t *testing.T) {
+	ep := testEP(t)
+	ctx := context.Background()
+
+	seedInbox(t, ep, "actsrc@localhost", "p", 1)
+
+	src, err := Connect(ctx, ep)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = src.Logout().Wait() }()
+	if err := src.Login("actsrc@localhost", "p").Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	dst, err := Connect(ctx, ep)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = dst.Logout().Wait() }()
+	if err := dst.Login("actdst@localhost", "p").Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	var activity int
+	deps := CopyDeps{
+		IsMigrated:   func(string) (bool, error) { return false, nil },
+		MarkMigrated: func(_, _ string) error { return nil },
+		OnProgress:   func(_, _ int) {},
+		OnActivity:   func() { activity++ },
+	}
+
+	r, err := CopyFolder(ctx, src, dst, "INBOX", "INBOX", deps)
+	if err != nil {
+		t.Fatalf("CopyFolder: %v", err)
+	}
+	if r.Copied != 1 {
+		t.Fatalf("copied=%d want 1", r.Copied)
+	}
+	if activity == 0 {
+		t.Fatal("OnActivity never called during body transfer")
+	}
+}
+
 // Требует два ящика на greenmail. Первый запуск копирует N, второй — 0 (все skipped).
 func TestCopyFolderIdempotent(t *testing.T) {
 	ep := testEP(t) // plain greenmail

internal/imapx/dial.go

@@ -18,6 +18,29 @@ type Endpoint struct {
 
 func (e Endpoint) addr() string { return fmt.Sprintf("%s:%d", e.Host, e.Port) }
 
+// Client wraps an imapclient.Client together with the raw network connection it
+// runs over. The embedded *imapclient.Client provides the full IMAP API; the
+// retained conn lets callers impose a read deadline on the socket for the
+// duration of a body transfer.
+//
+// This defends against servers that announce a BODY[] literal LARGER than the
+// bytes they actually send (a protocol violation observed on some webmail
+// servers). go-imap reads a literal strictly by its announced size, so a short
+// literal makes it block forever waiting for bytes that never arrive. A
+// deadline around the body read turns that infinite hang into a timeout the
+// copier can recover from.
+type Client struct {
+	*imapclient.Client
+	conn net.Conn
+}
+
+// SetReadDeadline sets (or, with the zero time, clears) a deadline on the
+// underlying socket. Used to bound a single body read; always cleared again
+// once the read completes so it never affects idle periods.
+func (c *Client) SetReadDeadline(t time.Time) error {
+	return c.conn.SetReadDeadline(t)
+}
+
 // dialTimeout bounds establishing the TCP connection.
 const dialTimeout = 30 * time.Second
 
@@ -27,7 +50,7 @@ const dialTimeout = 30 * time.Second
 // during a long src scan) from one stuck mid-response, and would wrongly close
 // idle connections. Stall detection is done at the orchestrator level via a
 // progress watchdog; go-imap's own per-command timeouts bound active commands.
-func dialOnce(ctx context.Context, ep Endpoint) (*imapclient.Client, error) {
+func dialOnce(ctx context.Context, ep Endpoint) (*Client, error) {
 	d := &net.Dialer{Timeout: dialTimeout}
 	raw, err := d.DialContext(ctx, "tcp", ep.addr())
 	if err != nil {
@@ -42,16 +65,27 @@ func dialOnce(ctx context.Context, ep Endpoint) (*imapclient.Client, error) {
 			_ = raw.Close()
 			return nil, err
 		}
-		return waitGreeting(imapclient.New(tlsConn, nil))
+		c, err := waitGreeting(imapclient.New(tlsConn, nil))
+		if err != nil {
+			return nil, err
+		}
+		return &Client{Client: c, conn: tlsConn}, nil
 	case "starttls":
+		// Deadline goes on the raw TCP conn: it sits beneath the TLS layer that
+		// NewStartTLS negotiates, and a TCP read deadline still interrupts the
+		// TLS read above it.
 		opts := &imapclient.Options{TLSConfig: &tls.Config{ServerName: ep.Host}}
 		c, err := imapclient.NewStartTLS(raw, opts)
 		if err != nil {
 			return nil, err
 		}
-		return c, nil
+		return &Client{Client: c, conn: raw}, nil
 	case "plain":
-		return waitGreeting(imapclient.New(raw, nil))
+		c, err := waitGreeting(imapclient.New(raw, nil))
+		if err != nil {
+			return nil, err
+		}
+		return &Client{Client: c, conn: raw}, nil
 	default:
 		_ = raw.Close()
 		return nil, fmt.Errorf("unknown tls_mode %q", ep.TLSMode)
@@ -69,7 +103,7 @@ func waitGreeting(c *imapclient.Client) (*imapclient.Client, error) {
 	return c, nil
 }
 
-func Connect(ctx context.Context, ep Endpoint) (*imapclient.Client, error) {
+func Connect(ctx context.Context, ep Endpoint) (*Client, error) {
 	const attempts = 3
 	var lastErr error
 	for i := 0; i < attempts; i++ {

internal/imapx/keepalive.go

@@ -0,0 +1,45 @@
+package imapx
+
+import (
+	"context"
+	"time"
+)
+
+// KeepaliveInterval is how often an otherwise-idle IMAP connection is pinged
+// with NOOP so the server does not drop it.
+//
+// The destination connection sits completely idle for the entire duration of
+// the source-side metadata scan (Pass 1 of CopyFolder), which on a large
+// mailbox runs for many minutes across all folders. With no traffic, the
+// server closes the idle connection; go-imap's reader then tears the client
+// down, and every subsequent APPEND fails with "use of closed network
+// connection" — aborting each folder and copying nothing. A periodic NOOP
+// keeps the connection warm. 60s is well under the idle timeout of any common
+// IMAP server.
+const KeepaliveInterval = 60 * time.Second
+
+// Keepalive pings c with a NOOP every interval until ctx is cancelled, keeping
+// an idle connection from being dropped by the server. It is meant to run in
+// its own goroutine.
+//
+// It is safe to run concurrently with other commands on c: go-imap serializes
+// command submission and supports multiple in-flight commands over a single
+// connection (RFC 9051 §5.5 pipelining). A NOOP issued while another command is
+// in flight simply queues behind it and completes when the server responds.
+//
+// Keepalive returns when ctx is done or when a NOOP fails — a failed NOOP means
+// the connection is already gone, so there is nothing left to keep alive.
+func Keepalive(ctx context.Context, c *Client, interval time.Duration) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			if err := c.Noop().Wait(); err != nil {
+				return
+			}
+		}
+	}
+}

internal/imapx/keepalive_test.go

@@ -0,0 +1,89 @@
+package imapx
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestKeepaliveReturnsOnContextCancel proves Keepalive is a well-behaved
+// goroutine: it exits promptly when its context is cancelled instead of
+// leaking.
+func TestKeepaliveReturnsOnContextCancel(t *testing.T) {
+	ep := testEP(t)
+	ctx := context.Background()
+
+	c, err := Connect(ctx, ep)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = c.Logout().Wait() }()
+	if err := c.Login("ka1@localhost", "p").Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	kctx, cancel := context.WithCancel(ctx)
+	done := make(chan struct{})
+	go func() { Keepalive(kctx, c, 10*time.Millisecond); close(done) }()
+
+	// Let a few NOOPs fire, then cancel and require a prompt return.
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Keepalive did not return within 2s of context cancel")
+	}
+}
+
+// TestKeepaliveDoesNotDisruptCopy runs Keepalive on the destination connection
+// at an aggressive interval while CopyFolder is APPENDing to it, proving the
+// concurrent NOOPs do not corrupt in-flight commands (the real risk of pinging
+// a connection that is also being used for real work).
+func TestKeepaliveDoesNotDisruptCopy(t *testing.T) {
+	ep := testEP(t)
+	ctx := context.Background()
+
+	const n = 8
+	seedInbox(t, ep, "kasrc@localhost", "p", n)
+
+	src, err := Connect(ctx, ep)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = src.Logout().Wait() }()
+	if err := src.Login("kasrc@localhost", "p").Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	dst, err := Connect(ctx, ep)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = dst.Logout().Wait() }()
+	if err := dst.Login("kadst@localhost", "p").Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	kctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	go Keepalive(kctx, dst, 1*time.Millisecond)
+
+	seen := map[string]bool{}
+	deps := CopyDeps{
+		IsMigrated:   func(k string) (bool, error) { return seen[k], nil },
+		MarkMigrated: func(_, k string) error { seen[k] = true; return nil },
+		OnProgress:   func(_, _ int) {},
+	}
+
+	r, err := CopyFolder(kctx, src, dst, "INBOX", "INBOX", deps)
+	if err != nil {
+		t.Fatalf("CopyFolder with concurrent keepalive: %v", err)
+	}
+	if r.Copied != n {
+		t.Fatalf("copied=%d want %d", r.Copied, n)
+	}
+	if r.Errors != 0 {
+		t.Fatalf("errors=%d want 0", r.Errors)
+	}
+}

internal/imapx/underflow_test.go

@@ -0,0 +1,218 @@
+package imapx
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/emersion/go-imap/v2"
+)
+
+// underflowServer is a minimal IMAP server that reproduces the amega.kz bug: it
+// answers a BODY[] FETCH by announcing a literal LARGER than the bytes it then
+// sends, and afterwards goes silent — exactly what makes go-imap's strict
+// literal reader block forever. LOGIN/EXAMINE and the Pass-1 metadata FETCH are
+// answered normally so a full CopyFolder can reach the poisoned message.
+//
+// It serves one connection per accept and keeps accepting, so a reconnect gets
+// a fresh, well-behaved session (its second EXAMINE reports zero messages, so
+// the resumed folder simply finishes).
+type underflowServer struct {
+	ln      net.Listener
+	mu      sync.Mutex
+	accepts int // how many connections have been accepted
+	stop    chan struct{}
+}
+
+func newUnderflowServer(t *testing.T) *underflowServer {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	s := &underflowServer{ln: ln, stop: make(chan struct{})}
+	go s.serve()
+	return s
+}
+
+func (s *underflowServer) addr() Endpoint {
+	a := s.ln.Addr().(*net.TCPAddr)
+	return Endpoint{Host: "127.0.0.1", Port: a.Port, TLSMode: "plain"}
+}
+
+func (s *underflowServer) close() { close(s.stop); _ = s.ln.Close() }
+
+func (s *underflowServer) serve() {
+	for {
+		conn, err := s.ln.Accept()
+		if err != nil {
+			return
+		}
+		s.mu.Lock()
+		s.accepts++
+		first := s.accepts == 1
+		s.mu.Unlock()
+		go s.handle(conn, first)
+	}
+}
+
+// handle drives one connection. On the FIRST connection the mailbox reports one
+// message and its BODY[] fetch under-delivers; on any later connection (i.e.
+// after a reconnect) the mailbox is empty so the resumed folder completes.
+func (s *underflowServer) handle(conn net.Conn, first bool) {
+	defer func() { _ = conn.Close() }()
+	br := bufio.NewReader(conn)
+	fmt.Fprint(conn, "* OK IMAP4rev1 ready\r\n")
+	for {
+		line, err := br.ReadString('\n')
+		if err != nil {
+			return
+		}
+		fields := strings.Fields(line)
+		if len(fields) == 0 {
+			continue
+		}
+		tag := fields[0]
+		up := strings.ToUpper(line)
+		switch {
+		case strings.Contains(up, "LOGIN"):
+			fmt.Fprintf(conn, "%s OK LOGIN completed\r\n", tag)
+		case strings.Contains(up, "EXAMINE"), strings.Contains(up, "SELECT"):
+			n := 0
+			if first {
+				n = 1
+			}
+			fmt.Fprintf(conn, "* %d EXISTS\r\n", n)
+			fmt.Fprint(conn, "* OK [UIDVALIDITY 1] ok\r\n")
+			fmt.Fprintf(conn, "%s OK [READ-ONLY] EXAMINE completed\r\n", tag)
+		case strings.Contains(up, "BODY["), strings.Contains(up, "BODY.PEEK"):
+			// Poison: announce 100000 bytes, send 10, then stall until shutdown.
+			fmt.Fprint(conn, "* 1 FETCH (UID 1 BODY[] {100000}\r\n")
+			fmt.Fprint(conn, "0123456789")
+			<-s.stop
+			return
+		case strings.Contains(up, "FETCH"):
+			// Pass-1 metadata fetch for the single message.
+			fmt.Fprint(conn, "* 1 FETCH (UID 1 RFC822.SIZE 100 FLAGS () "+
+				"INTERNALDATE \"01-Jan-2020 00:00:00 +0000\" "+
+				"ENVELOPE (\"Wed, 01 Jan 2020 00:00:00 +0000\" \"poison\" NIL NIL NIL NIL NIL NIL NIL \"<poison@x>\"))\r\n")
+			fmt.Fprintf(conn, "%s OK FETCH completed\r\n", tag)
+		case strings.Contains(up, "LOGOUT"):
+			fmt.Fprintf(conn, "* BYE\r\n%s OK LOGOUT completed\r\n", tag)
+			return
+		case strings.Contains(up, "CREATE"), strings.Contains(up, "NOOP"):
+			fmt.Fprintf(conn, "%s OK completed\r\n", tag)
+		default:
+			fmt.Fprintf(conn, "%s OK completed\r\n", tag)
+		}
+	}
+}
+
+// TestStreamOneTimesOutOnLiteralUnderflow proves a server that announces a
+// larger BODY[] literal than it sends makes streamOne return ErrBodyTimeout
+// (bounded by bodyIdleTimeout) instead of hanging forever.
+func TestStreamOneTimesOutOnLiteralUnderflow(t *testing.T) {
+	restore := shortenBodyIdle(200 * time.Millisecond)
+	defer restore()
+
+	srv := newUnderflowServer(t)
+	defer srv.close()
+	ctx := context.Background()
+
+	src, err := Connect(ctx, srv.addr())
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer func() { _ = src.Close() }()
+	if err := src.Login("u", "p").Wait(); err != nil {
+		t.Fatalf("login: %v", err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- streamOne(src, src, "INBOX", imap.UID(1), nil, time.Time{}, nil)
+	}()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, ErrBodyTimeout) {
+			t.Fatalf("want ErrBodyTimeout, got %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("streamOne hung on literal underflow instead of timing out")
+	}
+}
+
+// TestCopyFolderSkipsAndReconnectsOnUnderflow proves CopyFolder does not hang on
+// a poisoned message: it marks the message migrated (so future runs skip it),
+// invokes ReconnectSrc, and returns with the error counted — the folder is not
+// wedged.
+func TestCopyFolderSkipsAndReconnectsOnUnderflow(t *testing.T) {
+	restore := shortenBodyIdle(200 * time.Millisecond)
+	defer restore()
+
+	srv := newUnderflowServer(t)
+	defer srv.close()
+	ctx := context.Background()
+
+	src, err := Connect(ctx, srv.addr())
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer func() { _ = src.Close() }()
+	if err := src.Login("u", "p").Wait(); err != nil {
+		t.Fatalf("login: %v", err)
+	}
+
+	var marked []string
+	var reconnected bool
+	deps := CopyDeps{
+		IsMigrated:   func(string) (bool, error) { return false, nil },
+		MarkMigrated: func(_, k string) error { marked = append(marked, k); return nil },
+		OnProgress:   func(_, _ int) {},
+		ReconnectSrc: func() (*Client, error) {
+			reconnected = true
+			nc, derr := Connect(ctx, srv.addr())
+			if derr != nil {
+				return nil, derr
+			}
+			if lerr := nc.Login("u", "p").Wait(); lerr != nil {
+				return nil, lerr
+			}
+			return nc, nil
+		},
+	}
+
+	done := make(chan CopyResult, 1)
+	go func() {
+		r, _ := CopyFolder(ctx, src, src, "INBOX", "INBOX", deps)
+		done <- r
+	}()
+
+	select {
+	case r := <-done:
+		if r.Errors == 0 {
+			t.Fatalf("expected the poisoned message counted as an error, got %+v", r)
+		}
+		if !reconnected {
+			t.Fatal("ReconnectSrc was never called")
+		}
+		if len(marked) != 1 {
+			t.Fatalf("poisoned message should be marked migrated once, got %v", marked)
+		}
+	case <-time.After(8 * time.Second):
+		t.Fatal("CopyFolder hung on a poisoned message instead of skipping it")
+	}
+}
+
+func shortenBodyIdle(d time.Duration) func() {
+	old := bodyIdleTimeout
+	bodyIdleTimeout = d
+	return func() { bodyIdleTimeout = old }
+}

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,37 +392,47 @@ 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,
 		"dst_login": a.DstLogin, "dst_host": dstEP.Host, "dst_port": dstEP.Port,
 	}})
 	_ = o.store.SetAccountStatus(ctx, a.ID, "running")
-	_ = o.store.SetAccountError(ctx, a.ID, "")     // clear any error from a previous run
-	_ = 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
+	_ = o.store.SetAccountError(ctx, a.ID, "")  // clear any error from a previous run
+	_ = 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
-	// stops it); DB writes keep the parent ctx so status/counters persist even
-	// after cancellation. ctx is context.WithoutCancel from runAll.
-	actx, cancel := context.WithCancel(ctx)
+	// Per-account cancellable context: IMAP work uses actx, so both CancelAccount
+	// and a task-wide pause/cancel (which cancels runCtx) stop it. DB writes keep
+	// ctx — the uncancellable one from runAll — so status/counters persist even
+	// after cancellation.
+	actx, cancel := context.WithCancel(runCtx)
 	o.registerCancel(a.ID, cancel)
 	defer func() {
 		o.unregisterCancel(a.ID)
@@ -322,38 +441,77 @@ 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)
 	}
-	defer func() { _ = src.Logout().Wait() }()
 	if err := src.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil {
-		return o.accountFailed(ctx, task.ID, runID, a, srcEP, dstEP, "src", err)
+		_ = src.Logout().Wait()
+		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,
+	// which swaps this pointer. The cancel goroutine and the deferred logout
+	// below both read through it, so they always act on the current connection
+	// rather than a stale one that was already replaced and logged out.
+	var srcClient atomic.Pointer[imapx.Client]
+	srcClient.Store(src)
+	defer func() { _ = srcClient.Load().Logout().Wait() }()
+
 	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
+	// live connection, and logs the old (desynced) one out. CopyFolder calls it
+	// to recover after a message body times out: the server left the connection
+	// mid-literal, so it can't be reused. Bound to actx, so a cancelled account
+	// fails the dial instead of reconnecting.
+	reconnectSrc := func() (*imapx.Client, error) {
+		nc, err := imapx.Connect(actx, srcEP)
+		if err != nil {
+			return nil, err
+		}
+		if err := nc.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil {
+			_ = nc.Logout().Wait()
+			return nil, err
+		}
+		if old := srcClient.Swap(nc); old != nil {
+			_ = old.Logout().Wait()
+		}
+		slog.Info("reconnected src after message body timeout", "account", a.ID, "src_login", a.SrcLogin)
+		return nc, nil
 	}
 
 	// On cancel, close the connections so any in-flight network read (a slow
 	// FETCH/Collect that ctx.Err() checks can't interrupt) unblocks immediately.
 	go func() {
 		<-actx.Done()
-		_ = src.Close()
+		_ = srcClient.Load().Close()
 		_ = dst.Close()
 	}()
 
+	// Keep the destination connection warm. It sits idle for the whole
+	// source-side metadata scan (Pass 1 of CopyFolder), which on a large
+	// mailbox runs for minutes; without traffic the server drops it and every
+	// subsequent APPEND fails with "use of closed network connection", copying
+	// nothing. Periodic NOOPs prevent that. Only dst needs it — src is
+	// continuously busy scanning/fetching. Bound to actx so it stops with the
+	// account.
+	go imapx.Keepalive(actx, dst, imapx.KeepaliveInterval)
+
 	// Progress watchdog: track the last time we saw scan/copy activity; if it
 	// goes quiet for stallTimeout, cancel the account so the connections close
 	// and this worker unwinds (it would otherwise block forever on a silent
@@ -383,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
@@ -432,6 +590,14 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 		IsMigrated:   func(k string) (bool, error) { return o.store.IsMigrated(ctx, a.ID, k) },
 		MarkMigrated: func(folder, k string) error { return o.store.MarkMigrated(ctx, a.ID, folder, k) },
 		OnError:      func(ref, msg string) { addErr("message", curFolder, ref, msg) },
+		// Fires as bytes move within a single message's FETCH/APPEND, so the
+		// stall watchdog sees a large-but-live transfer as progress instead of
+		// cancelling it as a wedged connection.
+		OnActivity: touch,
+		// Recovers from a message body timeout (server under-delivering a
+		// literal) by swapping in a fresh source connection so the folder can
+		// resume with the remaining messages.
+		ReconnectSrc: reconnectSrc,
 		OnProgress: func(c, s int) {
 			touch()
 			now := time.Now()
@@ -478,7 +644,7 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
 		if actx.Err() != nil {
 			break // cancelled — stop scheduling more folders
 		}
-		res, err := imapx.CopyFolder(actx, src, dst, fp.src, fp.dst, deps)
+		res, err := imapx.CopyFolder(actx, srcClient.Load(), dst, fp.src, fp.dst, deps)
 		folderErr := int64(0)
 		if err != nil && actx.Err() == nil {
 			slog.Warn("folder copy error", "account", a.ID, "src_login", a.SrcLogin, "folder", fp.src, "err", err)
@@ -500,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")
-		o.hub.Publish(wshub.Event{Type: "cancelled", TaskID: task.ID,
+		// A task-wide pause leaves the account resumable; anything else (per-account
+		// 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
 	}
 
@@ -520,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) {
-	// A cancellation surfacing as an error is a cancel, not a failure.
+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 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")
-		o.hub.Publish(wshub.Event{Type: "cancelled", TaskID: taskID,
+		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: 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/crud_test.go

@@ -18,6 +18,41 @@ func TestUpdateEndpoint(t *testing.T) {
 	}
 }
 
+func TestDeleteEndpoint(t *testing.T) {
+	s := testStore(t)
+	ctx := context.Background()
+	id, _ := s.CreateEndpoint(ctx, Endpoint{RoleLabel: "src", Host: "a.com", Port: 993, TLSMode: "ssl"})
+	if err := s.DeleteEndpoint(ctx, id); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	eps, _ := s.ListEndpoints(ctx)
+	if len(eps) != 0 {
+		t.Fatalf("endpoint not deleted: %d remain", len(eps))
+	}
+}
+
+func TestDeleteEndpointUsedByTaskRefused(t *testing.T) {
+	s := testStore(t)
+	ctx := context.Background()
+	ep1, _ := s.CreateEndpoint(ctx, Endpoint{RoleLabel: "s", Host: "a", Port: 993, TLSMode: "ssl"})
+	ep2, _ := s.CreateEndpoint(ctx, Endpoint{RoleLabel: "d", Host: "b", Port: 993, TLSMode: "ssl"})
+	if _, err := s.CreateTask(ctx, Task{Name: "t", SrcEndpointID: ep1, DstEndpointID: ep2}); err != nil {
+		t.Fatalf("create task: %v", err)
+	}
+	for _, id := range []int64{ep1, ep2} {
+		n, err := s.CountTasksUsingEndpoint(ctx, id)
+		if err != nil {
+			t.Fatalf("count: %v", err)
+		}
+		if n != 1 {
+			t.Fatalf("count for ep %d = %d, want 1", id, n)
+		}
+		if err := s.DeleteEndpoint(ctx, id); err == nil {
+			t.Fatalf("delete of referenced endpoint %d succeeded, want FK violation", id)
+		}
+	}
+}
+
 func TestDeleteAccountCascadesJournal(t *testing.T) {
 	s := testStore(t)
 	ctx := context.Background()

internal/store/endpoints.go

@@ -26,6 +26,23 @@ func (s *Store) UpdateEndpoint(ctx context.Context, e Endpoint) error {
 	return err
 }
 
+// CountTasksUsingEndpoint reports how many tasks reference the endpoint on
+// either side, so a delete can be refused with a meaningful message instead of
+// surfacing a raw foreign-key violation.
+func (s *Store) CountTasksUsingEndpoint(ctx context.Context, id int64) (int, error) {
+	var n int
+	err := s.Pool.QueryRow(ctx,
+		`SELECT count(*) FROM tasks WHERE src_endpoint_id=$1 OR dst_endpoint_id=$1`, id).Scan(&n)
+	return n, err
+}
+
+// DeleteEndpoint removes an endpoint. Tasks reference endpoints without ON
+// DELETE CASCADE, so Postgres rejects the delete while any task still uses it.
+func (s *Store) DeleteEndpoint(ctx context.Context, id int64) error {
+	_, err := s.Pool.Exec(ctx, `DELETE FROM endpoints WHERE id=$1`, id)
+	return err
+}
+
 func (s *Store) GetEndpoint(ctx context.Context, id int64) (Endpoint, error) {
 	var e Endpoint
 	err := s.Pool.QueryRow(ctx,

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_SKIPPED_TOTAL=$(echo "$RES2" | jq -r '.accounts[0].skipped')
+RUN2_COPIED=$(echo "$RES2" | jq -r '.accounts[0].copied')
+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))
-RUN2_SKIPPED_DELTA=$((RUN2_SKIPPED_TOTAL - RUN1_SKIPPED))
-log "run 2: copied_delta=$RUN2_COPIED_DELTA skipped_delta=$RUN2_SKIPPED_DELTA errors=$RUN2_ERRORS"
+log "run 2: copied=$RUN2_COPIED skipped=$RUN2_SKIPPED 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_SKIPPED_DELTA" -gt 0 ]] || fail "run 2 skipped delta is $RUN2_SKIPPED_DELTA (expected >0)"
+[[ "$RUN2_COPIED" -eq 0 ]] || fail "run 2 copied $RUN2_COPIED new messages (expected 0, not idempotent)"
+[[ "$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

@@ -82,11 +82,21 @@ export const updateEndpoint = (
   body: { role_label: string; host: string; port: number; tls_mode: TLSMode },
 ) => api(`/api/endpoints/${id}`, { ...jsonBody(body), method: 'PUT' })
 
+export const deleteEndpoint = (id: number) => api(`/api/endpoints/${id}`, { method: 'DELETE' })
+
 export const deleteTask = (id: number) => api(`/api/tasks/${id}`, { method: 'DELETE' })
 
 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' })
 
@@ -141,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
@@ -177,3 +195,13 @@ export const importCSV = (id: number, file: File) => {
   fd.append('file', file)
   return api<{ imported: number }>(`/api/tasks/${id}/import`, { method: 'POST', body: fd })
 }
+
+// A Kerio Connect export carries no domain, so the operator supplies it here;
+// the login and password apply to both sides of the migration.
+export const importKerioCSV = (id: number, file: File, domain: string) => {
+  const fd = new FormData()
+  fd.append('file', file)
+  fd.append('format', 'kerio')
+  fd.append('domain', domain)
+  return api<{ imported: number }>(`/api/tasks/${id}/import`, { method: 'POST', body: fd })
+}

web/src/app.css

@@ -488,11 +488,23 @@
   cursor: pointer;
 }
 
+.map-toolbar {
+  display: flex;
+  align-items: center;
+  gap: 14px;
+  margin-bottom: 12px;
+  flex-wrap: wrap;
+}
+
+.map-toolbar .btn {
+  padding: 6px 12px;
+  font-size: 11px;
+}
+
 .map-all {
   display: flex;
   align-items: center;
   gap: 8px;
-  margin-bottom: 12px;
   font-size: 11px;
   text-transform: uppercase;
   letter-spacing: 0.08em;
@@ -655,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;
@@ -905,11 +936,24 @@ table.tbl a.rowlink:hover {
 
 .upload-row {
   display: flex;
-  align-items: center;
+  align-items: flex-start;
   gap: 12px;
   flex-wrap: wrap;
 }
 
+/* One import route per column: the action on top, its sample file underneath. */
+.upload-item {
+  display: flex;
+  flex-direction: column;
+  align-items: stretch;
+  gap: 6px;
+}
+
+.upload-item .link-btn {
+  align-self: center;
+  text-align: center;
+}
+
 .file-btn {
   position: relative;
   overflow: hidden;

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/components/FolderMappingModal.tsx

@@ -20,6 +20,21 @@ function defaultDst(src: string, dstFolders: string[], initial: Record<string, s
   return src
 }
 
+// Exchange/Kerio special folders and their mailcow counterparts. Keyed by the
+// lowercased source name so casing differences between servers don't matter.
+const DEFAULT_TARGETS: Record<string, string> = {
+  'deleted items': 'Trash',
+  'deleted messages': 'Trash',
+  'junk e-mail': 'Junk',
+  'junk email': 'Junk',
+  spam: 'Junk',
+  'sent items': 'Sent',
+  'sent messages': 'Sent',
+}
+
+// Source folders with no counterpart on mailcow — unchecked by the defaults.
+const DEFAULT_EXCLUDED = new Set(['public folders'])
+
 export function FolderMappingModal({
   open, srcFolders, dstFolders, initialMapping, initialExcluded, accountLabel, onConfirm, onCancel,
 }: Props) {
@@ -31,16 +46,40 @@ export function FolderMappingModal({
 
   // Options per select: all destination folders, plus the source name itself
   // (marked "create") when it does not already exist on the destination.
+  const valueFor = (src: string) => choice[src] ?? defaultDst(src, dstFolders, initialMapping)
+
+  // Options per select: all destination folders, plus any name not present there
+  // — the source folder itself and the current selection — marked "create".
   const options = useMemo(() => {
     const set = new Set(dstFolders)
-    return (src: string) => {
+    return (src: string, current: string) => {
       const opts = [...dstFolders]
-      if (!set.has(src)) opts.unshift(src)
+      if (!set.has(current)) opts.unshift(current)
+      if (!set.has(src) && src !== current) opts.unshift(src)
       return opts
     }
   }, [dstFolders])
 
-  const valueFor = (src: string) => choice[src] ?? defaultDst(src, dstFolders, initialMapping)
+  // Collapse the Exchange/Kerio folder layout onto mailcow's in one click: the
+  // per-account mapping is otherwise repetitive work when importing many users.
+  function applyDefaults() {
+    const nextChoice = { ...choice }
+    const nextSynced = { ...synced }
+    for (const src of srcFolders) {
+      const key = src.trim().toLowerCase()
+      if (DEFAULT_EXCLUDED.has(key)) {
+        nextSynced[src] = false
+        continue
+      }
+      const target = DEFAULT_TARGETS[key]
+      if (target) {
+        nextChoice[src] = target
+        nextSynced[src] = true
+      }
+    }
+    setChoice(nextChoice)
+    setSynced(nextSynced)
+  }
 
   function confirm() {
     const mapping: Record<string, string> = {}
@@ -70,17 +109,27 @@ export function FolderMappingModal({
           Route each source folder to an existing destination folder. Leaving a folder mapped to its own name
           creates it on the destination if missing (e.g. map <code>Спам</code> → <code>Spam</code> to avoid duplicates).
         </p>
-        <label className="map-all">
-          <input
-            type="checkbox"
-            checked={srcFolders.every((f) => synced[f] !== false)}
-            onChange={(e) => {
-              const on = e.target.checked
-              setSynced(Object.fromEntries(srcFolders.map((f) => [f, on])))
-            }}
-          />
-          sync all folders
-        </label>
+        <div className="map-toolbar">
+          <button
+            type="button"
+            className="btn btn-ghost"
+            onClick={applyDefaults}
+            title="Map Deleted Items → Trash, Junk E-mail → Junk, Sent Items → Sent and skip Public Folders"
+          >
+            By default
+          </button>
+          <label className="map-all">
+            <input
+              type="checkbox"
+              checked={srcFolders.every((f) => synced[f] !== false)}
+              onChange={(e) => {
+                const on = e.target.checked
+                setSynced(Object.fromEntries(srcFolders.map((f) => [f, on])))
+              }}
+            />
+            sync all folders
+          </label>
+        </div>
         <div className="map-grid">
           {srcFolders.map((src) => {
             const on = synced[src] !== false
@@ -107,10 +156,10 @@ export function FolderMappingModal({
                   disabled={!on}
                   onChange={(e) => setChoice((c) => ({ ...c, [src]: e.target.value }))}
                 >
-                  {options(src).map((f) => (
+                  {options(src, val).map((f) => (
                     <option key={f} value={f}>
                       {f}
-                      {f === src && !dstFolders.includes(src) ? ' (create)' : ''}
+                      {dstFolders.includes(f) ? '' : ' (create)'}
                     </option>
                   ))}
                 </select>

web/src/components/KerioImportModal.tsx

@@ -0,0 +1,82 @@
+import { useEffect, useState, type FormEvent } from 'react'
+import { Modal } from './Modal'
+
+// A bare DNS domain: no scheme, no user part, no whitespace. Mirrors the
+// server-side check in csvimport.normalizeDomain so bad input is caught here.
+const domainRe = /^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+$/
+
+type Props = {
+  open: boolean
+  busy: boolean
+  onClose: () => void
+  onSubmit: (file: File, domain: string) => void
+}
+
+export function KerioImportModal({ open, busy, onClose, onSubmit }: Props) {
+  const [domain, setDomain] = useState('')
+  const [file, setFile] = useState<File | null>(null)
+  const [error, setError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (!open) return
+    setDomain('')
+    setFile(null)
+    setError(null)
+  }, [open])
+
+  function submit(e: FormEvent) {
+    e.preventDefault()
+    const d = domain.trim().toLowerCase()
+    if (!domainRe.test(d)) {
+      setError('Enter a bare domain, e.g. galaxyhotel.kz')
+      return
+    }
+    if (!file) {
+      setError('Choose the Kerio export file')
+      return
+    }
+    setError(null)
+    onSubmit(file, d)
+  }
+
+  return (
+    <Modal open={open} title="Import from Kerio" onClose={onClose}>
+      <form onSubmit={submit}>
+        <p className="map-hint">
+          The Kerio user export lists a login and its password but no domain. The domain you enter is appended to every
+          login and used for both the source and the destination. Disabled accounts and <code>admin</code> are skipped.
+        </p>
+        <div className="field">
+          <label htmlFor="kerio_domain">Mail domain</label>
+          <input
+            id="kerio_domain"
+            data-modal-autofocus
+            value={domain}
+            onChange={(e) => setDomain(e.target.value)}
+            placeholder="galaxyhotel.kz"
+            disabled={busy}
+          />
+        </div>
+        <div className="field">
+          <label htmlFor="kerio_file">Export file</label>
+          <input
+            id="kerio_file"
+            type="file"
+            accept=".csv,text/csv"
+            onChange={(e) => setFile(e.target.files?.[0] ?? null)}
+            disabled={busy}
+          />
+        </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 ? 'Importing…' : 'Import'}
+          </button>
+        </div>
+      </form>
+    </Modal>
+  )
+}

web/src/pages/Endpoints.tsx

@@ -1,5 +1,6 @@
 import { useEffect, useState, type FormEvent } from 'react'
-import { createEndpoint, listEndpoints, updateEndpoint, type Endpoint, type TLSMode } from '../api'
+import { createEndpoint, deleteEndpoint, listEndpoints, updateEndpoint, type Endpoint, type TLSMode } from '../api'
+import { useConfirm } from '../components/ConfirmProvider'
 
 const emptyForm = { role_label: '', host: '', port: '993', tls_mode: 'ssl' as TLSMode }
 
@@ -9,6 +10,7 @@ export function Endpoints() {
   const [editingId, setEditingId] = useState<number | null>(null)
   const [error, setError] = useState<string | null>(null)
   const [busy, setBusy] = useState(false)
+  const confirm = useConfirm()
 
   function startEdit(ep: Endpoint) {
     setEditingId(ep.id)
@@ -29,6 +31,25 @@ export function Endpoints() {
 
   useEffect(reload, [])
 
+  async function onDelete(ep: Endpoint) {
+    const ok = await confirm({
+      title: 'Delete endpoint',
+      message: `Delete endpoint "${ep.role_label}" (${ep.host}:${ep.port})?`,
+      confirmLabel: 'Delete',
+      danger: true,
+    })
+    if (!ok) return
+    setError(null)
+    try {
+      await deleteEndpoint(ep.id)
+      // The form still edits a row that no longer exists — drop back to create mode.
+      if (editingId === ep.id) cancelEdit()
+      reload()
+    } catch (e) {
+      setError(e instanceof Error ? e.message : 'Failed to delete endpoint')
+    }
+  }
+
   async function submit(e: FormEvent) {
     e.preventDefault()
     setBusy(true)
@@ -159,6 +180,9 @@ export function Endpoints() {
                       <td className="num-cell">
                         <button type="button" className="link-btn" onClick={() => startEdit(ep)} disabled={busy}>
                           edit
+                        </button>{' '}
+                        <button type="button" className="link-btn danger" onClick={() => onDelete(ep)} disabled={busy}>
+                          delete
                         </button>
                       </td>
                     </tr>

web/src/pages/TaskDetail.tsx

@@ -1,11 +1,13 @@
 import { useEffect, useRef, useState, type ChangeEvent, type FormEvent } from 'react'
-import { cancelAccount, createAccount, deleteAccount, getTask, importCSV, 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'
 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: '' }
 
@@ -57,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}`
@@ -64,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)
   }
@@ -90,6 +94,8 @@ export function TaskDetail({ id }: { id: number }) {
   const [live, setLive] = useState<Record<number, LiveProgress>>({})
   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)
 
@@ -159,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
@@ -170,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()
         }
       }),
@@ -258,20 +264,40 @@ export function TaskDetail({ id }: { id: number }) {
     }
   }
 
-  function downloadExampleCSV() {
-    const sample = [
-      'alice@source.example,SrcPass1,alice@dest.example,DstPass1',
-      'bob@source.example,SrcPass2,bob@dest.example,DstPass2',
-      'carol@source.example,SrcPass3,carol@dest.example,DstPass3',
-    ].join('\n') + '\n'
-    const url = URL.createObjectURL(new Blob([sample], { type: 'text/csv' }))
+  function downloadCSV(name: string, content: string) {
+    const url = URL.createObjectURL(new Blob([content], { type: 'text/csv' }))
     const a = document.createElement('a')
     a.href = url
-    a.download = 'imap-copier-accounts-example.csv'
+    a.download = name
     a.click()
     URL.revokeObjectURL(url)
   }
 
+  // Plain import: comma-separated, no header, src_login,src_pass,dst_login,dst_pass.
+  function downloadExampleCSV() {
+    const sample =
+      [
+        'alice@source.example,SrcPass1,alice@dest.example,DstPass1',
+        'bob@source.example,SrcPass2,bob@dest.example,DstPass2',
+        'carol@source.example,SrcPass3,carol@dest.example,DstPass3',
+      ].join('\n') + '\n'
+    downloadCSV('imap-copier-accounts-example.csv', sample)
+  }
+
+  // Kerio Connect export: semicolon-separated with a Name;FullName;Description;Enable
+  // header. The password lives in Description; disabled rows and admin are skipped
+  // on import, and the domain is supplied in the import dialog.
+  function downloadKerioExampleCSV() {
+    const sample =
+      [
+        'Name;FullName;Description;Enable',
+        'alice;Alice Smith;SrcPass1;Yes',
+        'bob;Bob Jones;SrcPass2;Yes',
+        'carol;Carol White (disabled, skipped);SrcPass3;No',
+      ].join('\n') + '\n'
+    downloadCSV('kerio-users-example.csv', sample)
+  }
+
   async function onFileChosen(e: ChangeEvent<HTMLInputElement>) {
     const file = e.target.files?.[0]
     if (!file) return
@@ -288,6 +314,35 @@ export function TaskDetail({ id }: { id: number }) {
     }
   }
 
+  async function onKerioImport(file: File, domain: string) {
+    setBusy('import')
+    setError(null)
+    try {
+      await importKerioCSV(id, file, domain)
+      setKerioOpen(false)
+      reload()
+    } catch (err) {
+      setError(err instanceof Error ? err.message : 'Kerio import failed')
+    } finally {
+      setBusy(null)
+    }
+  }
+
+  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',
@@ -342,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 {
@@ -369,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')
@@ -446,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}>
-            {busy === 'run'
-              ? 'Starting…'
-              : effectiveSelected.length > 0
-                ? `Run selected (${effectiveSelected.length})`
-                : 'Run migration'}
-          </button>
-          {!runReady && accounts.length > 0 && (
+          {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>
+            </>
+          )}
+          {!isRunning && !runReady && accounts.length > 0 && (
             <span className="hint">
               {effectiveSelected.length > 0
                 ? 'selected accounts must pass both connection tests'
@@ -540,13 +681,23 @@ export function TaskDetail({ id }: { id: number }) {
 
           <div className="divider-label">or bulk import</div>
           <div className="upload-row">
-            <label className={`btn file-btn${busy !== null ? ' is-disabled' : ''}`}>
-              {busy === 'import' ? 'Importing…' : 'Upload CSV'}
-              <input ref={fileInputRef} type="file" accept=".csv,text/csv" onChange={onFileChosen} disabled={busy !== null} />
-            </label>
-            <button type="button" className="link-btn" onClick={downloadExampleCSV}>
-              download example.csv
-            </button>
+            <div className="upload-item">
+              <label className={`btn file-btn${busy !== null ? ' is-disabled' : ''}`}>
+                {busy === 'import' ? 'Importing…' : 'Upload CSV'}
+                <input ref={fileInputRef} type="file" accept=".csv,text/csv" onChange={onFileChosen} disabled={busy !== null} />
+              </label>
+              <button type="button" className="link-btn" onClick={downloadExampleCSV}>
+                download example.csv
+              </button>
+            </div>
+            <div className="upload-item">
+              <button type="button" className="btn" onClick={() => setKerioOpen(true)} disabled={busy !== null}>
+                Import from Kerio
+              </button>
+              <button type="button" className="link-btn" onClick={downloadKerioExampleCSV}>
+                download kerio example.csv
+              </button>
+            </div>
           </div>
         </div>
 
@@ -631,12 +782,8 @@ export function TaskDetail({ id }: { id: number }) {
                         </div>
                       )}
                     </td>
-                    <td>
-                      <StatusBadge status={a.test_src_status} />
-                    </td>
-                    <td>
-                      <StatusBadge status={a.test_dst_status} />
-                    </td>
+                    <td>{testCell(a, a.test_src_status)}</td>
+                    <td>{testCell(a, a.test_dst_status)}</td>
                     <td>
                       <StatusBadge status={a.status} />
                     </td>
@@ -768,6 +915,19 @@ export function TaskDetail({ id }: { id: number }) {
       )}
       <RunLogModal taskId={id} open={showRuns} onClose={() => setShowRuns(false)} />
       <AccountErrorsModal taskId={id} account={errorsFor} onClose={() => setErrorsFor(null)} />
+      <KerioImportModal
+        open={kerioOpen}
+        busy={busy === 'import'}
+        onClose={() => setKerioOpen(false)}
+        onSubmit={onKerioImport}
+      />
+      <AccountCredentialsModal
+        open={credsFor !== null}
+        busy={busy === 'add'}
+        account={credsFor}
+        onClose={() => setCredsFor(null)}
+        onSubmit={saveCredentials}
+      />
     </>
   )
 }