H-162: add PC conversation transfer flow (#33)

Co-authored-by: Rogee <rogee@ipao.vip>

channels/shangwutong/README.md

@@ -77,10 +77,13 @@ GET  /readyz
 GET  /metrics
 POST /internal/reconcile                    # 仅 loopback
 POST /internal/operations/accept-transfer   # 仅 loopback
+POST /internal/operations/transfer-conversation # 仅 loopback
 ```
 
 接管请求体为 `{"inbox_id":10,"conversation_id":100,"sid":"...","event_id":"...","occurred_at":"2026-08-15T07:00:00Z"}`，其中 `conversation_id` 是 GoChat internal ID。新请求只接受已映射到该 conversation 且最新商务通状态为“转接中”的 SID；相同请求返回同一队列记录。只有 operation 进入 `delivered` 才证明 `oc/accepttransfer.aspx` 返回了 `r=ok`，最终坐席归属继续以现有 `swt_state=transfer_accepted` 和 `swt_assignee_name` 同步结果为准。
 
+发起请求体字段相同。它只接受当前由其他坐席持有、最新状态为 `5` 的 SID，目标坐席从已同步的 `swt_assignee_name` 推导；Connector 以当前登录坐席为 `oname`、原坐席为 `oname1` 调用 `oc/Transfer0.aspx`。成功后仍须等待状态 `7` 再调用接受入口；接受后的归属以状态 `8` 及后续 `swt_assignee_name` 回执为准。
+
 ```bash
 shangwutong migrate up
 shangwutong migrate status

channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.down.sql

@@ -0,0 +1,22 @@
+CREATE TABLE outbound_operations_old (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  account_id INTEGER NOT NULL REFERENCES accounts(id),
+  swt_sid TEXT NOT NULL,
+  event_id TEXT NOT NULL UNIQUE,
+  operation TEXT NOT NULL CHECK(operation IN ('end_conversation', 'change_contact_name', 'accept_transfer')),
+  payload TEXT NOT NULL,
+  occurred_at DATETIME NOT NULL,
+  delivery_status TEXT NOT NULL DEFAULT 'pending' CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed')),
+  claimed_at DATETIME,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  next_attempt_at DATETIME,
+  last_error TEXT,
+  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+-- Preserve the source table if transfer_conversation rows make rollback unsafe.
+INSERT INTO outbound_operations_old SELECT * FROM outbound_operations;
+DROP TABLE outbound_operations;
+ALTER TABLE outbound_operations_old RENAME TO outbound_operations;
+CREATE INDEX idx_outbound_operations_ready ON outbound_operations(delivery_status, next_attempt_at, account_id, id);
+CREATE UNIQUE INDEX idx_outbound_one_delivering_operation_per_account ON outbound_operations(account_id) WHERE delivery_status = 'delivering';

channels/shangwutong/db/migrations/005_add_transfer_conversation_operation.up.sql

@@ -0,0 +1,23 @@
+CREATE TABLE outbound_operations_new (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  account_id INTEGER NOT NULL REFERENCES accounts(id),
+  swt_sid TEXT NOT NULL,
+  event_id TEXT NOT NULL UNIQUE,
+  operation TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  occurred_at DATETIME NOT NULL,
+  delivery_status TEXT NOT NULL DEFAULT 'pending',
+  claimed_at DATETIME,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  next_attempt_at DATETIME,
+  last_error TEXT,
+  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CHECK(operation IN ('end_conversation', 'change_contact_name', 'accept_transfer', 'transfer_conversation')),
+  CHECK(delivery_status IN ('pending', 'delivering', 'delivered', 'uncertain', 'failed'))
+);
+INSERT INTO outbound_operations_new SELECT * FROM outbound_operations;
+DROP TABLE outbound_operations;
+ALTER TABLE outbound_operations_new RENAME TO outbound_operations;
+CREATE INDEX idx_outbound_operations_ready ON outbound_operations(delivery_status, next_attempt_at, account_id, id);
+CREATE UNIQUE INDEX idx_outbound_one_delivering_operation_per_account ON outbound_operations(account_id) WHERE delivery_status = 'delivering';

channels/shangwutong/internal/command/root_test.go

@@ -87,7 +87,7 @@ func TestOperationalCommands(t *testing.T) {
 			t.Fatalf("%v produced no output", args)
 		}
 	}
-	if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 4 {
+	if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 5 {
 		t.Fatalf("backup version = %d, %v", version, err)
 	}
 

channels/shangwutong/internal/delivery/outbound.go

@@ -38,6 +38,7 @@ type MessageSender interface {
 	EndConversation(context.Context, swt.Session, string) error
 	ChangeContactName(context.Context, swt.Session, string, string, string) error
 	AcceptTransfer(context.Context, swt.Session, string) error
+	TransferConversation(context.Context, swt.Session, string, string) error
 }
 
 type ResultClient interface {
@@ -413,6 +414,17 @@ func (o *Outbound) processOperation(ctx context.Context) (bool, error) {
 			return o.sender.EndConversation(ctx, session, operation.SwtSid)
 		case "accept_transfer":
 			return o.sender.AcceptTransfer(ctx, session, operation.SwtSid)
+		case "transfer_conversation":
+			var payload struct {
+				OtherLoginName string `json:"other_login_name"`
+			}
+			if err := json.Unmarshal([]byte(operation.Payload), &payload); err != nil || strings.TrimSpace(payload.OtherLoginName) == "" {
+				return &swt.Error{Operation: operation.Operation, Code: "invalid_payload", Err: errors.New("other_login_name is required")}
+			}
+			if !session.AllowsJoiningOtherOperatorDialogue() {
+				return &swt.Error{Operation: operation.Operation, Code: "permission_denied", Err: errors.New("operator may not join another operator's dialogue")}
+			}
+			return o.sender.TransferConversation(ctx, session, operation.SwtSid, payload.OtherLoginName)
 		case "change_contact_name":
 			var envelope struct {
 				Data struct {

channels/shangwutong/internal/delivery/outbound_test.go

@@ -196,6 +196,67 @@ func TestOutboundAcceptTransferPersistsSuccessAndRetry(t *testing.T) {
 	}
 }
 
+func TestAcceptedTransferUpdatesOwnershipAndKeepsOutboundSendable(t *testing.T) {
+	ctx := context.Background()
+	database, account := deliveryDatabase(t, ctx)
+	conversationID := int64(100)
+	other := "other-agent"
+	if _, err := database.Writer().UpsertConversationMap(ctx, dbgen.UpsertConversationMapParams{
+		AccountID: account.ID, SwtSid: "visitor", GochatContactSourceID: "visitor",
+		GochatConversationID: &conversationID, SwtAssigneeName: &other,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "5", SeqID: 1, RawLine: "visitor 0 5 1"})
+	transfer, duplicate, err := database.EnqueueTransferConversation(ctx, store.OutboundOperationInput{
+		AccountID: account.ID, SWTSessionID: "visitor", EventID: "transfer-conversation:visitor:1",
+		Operation: "transfer_conversation", OccurredAt: time.Now(),
+	}, conversationID)
+	if err != nil || duplicate {
+		t.Fatalf("enqueue transfer=%#v duplicate=%v err=%v", transfer, duplicate, err)
+	}
+
+	sender := &operationSender{}
+	worker, _ := NewOutbound(database, allowedSessionStub{}, sender, &resultRecorder{}, nil, 1)
+	if worked, err := worker.processOperation(ctx); err != nil || !worked || sender.transferSID != "visitor" || sender.otherLoginName != other {
+		t.Fatalf("transfer worked=%v sender=%#v err=%v", worked, sender, err)
+	}
+
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "7", SeqID: 2, RawLine: "visitor 0 7 2"})
+	accept, duplicate, err := database.EnqueueAcceptTransfer(ctx, store.OutboundOperationInput{
+		AccountID: account.ID, SWTSessionID: "visitor", EventID: "accept-transfer:visitor:1",
+		Operation: "accept_transfer", Payload: `{}`, OccurredAt: time.Now().Add(time.Second),
+	}, conversationID)
+	if err != nil || duplicate {
+		t.Fatalf("enqueue accept=%#v duplicate=%v err=%v", accept, duplicate, err)
+	}
+	if worked, err := worker.processOperation(ctx); err != nil || !worked || sender.acceptSID != "visitor" {
+		t.Fatalf("accept worked=%v sender=%#v err=%v", worked, sender, err)
+	}
+
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 0, Text: "8", SeqID: 3, RawLine: "visitor 0 8 3"})
+	persistInboundEvent(t, database, account, swt.HeartbeatEvent{SessionID: "visitor", Kind: 31, Text: "distribute_chat|agent", SeqID: 4, RawLine: "visitor 31 distribute_chat|agent 4"})
+	inbound, _ := NewInbound(database, &inboundRecorder{}, nil, 1)
+	for range 4 {
+		if worked, err := inbound.processInbound(ctx); err != nil || !worked {
+			t.Fatalf("inbound worked=%v err=%v", worked, err)
+		}
+	}
+	mapping, err := database.Reader().GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: account.ID, SwtSid: "visitor"})
+	if err != nil || mapping.SwtAssigneeName == nil || *mapping.SwtAssigneeName != "agent" {
+		t.Fatalf("mapping=%#v err=%v", mapping, err)
+	}
+	if _, _, err := database.EnqueueOutbound(ctx, store.OutboundInput{
+		AccountID: account.ID, SWTSessionID: "visitor", EventID: "message:77:created", OccurredAt: time.Now().Add(2 * time.Second),
+		GoChatMessageID: 77, MessageType: "text", Content: stringPointer("hello"), Payload: deliveryPayload(t, nil),
+	}, false); err != nil {
+		t.Fatal(err)
+	}
+	if worked, err := worker.processOutbound(ctx); err != nil || !worked || sender.sentSID != "visitor" {
+		t.Fatalf("outbound worked=%v sender=%#v err=%v", worked, sender, err)
+	}
+}
+
 func TestOutboundRetrySkipsAlreadyDeliveredParts(t *testing.T) {
 	ctx := context.Background()
 	database, account := deliveryDatabase(t, ctx)
@@ -319,6 +380,13 @@ func (sessionStub) WithSession(ctx context.Context, _ int64, operation func(swt.
 
 func (sessionStub) InvalidateSession(context.Context, int64) error { return nil }
 
+type allowedSessionStub struct{ sessionStub }
+
+func (allowedSessionStub) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error {
+	purview := uint64(0)
+	return operation(swt.Session{BaseURL: "http://example.test/", SiteID: "site", LoginName: "agent", MAToken: "token", Purview: &purview})
+}
+
 type sessionRecorder struct{ invalidations int }
 
 func (*sessionRecorder) WithSession(ctx context.Context, _ int64, operation func(swt.Session) error) error {
@@ -354,14 +422,21 @@ func (s senderStub) ChangeContactName(context.Context, swt.Session, string, stri
 }
 
 func (s senderStub) AcceptTransfer(context.Context, swt.Session, string) error { return s.err }
-
-type operationSender struct {
-	sid       string
-	acceptSID string
-	err       error
+func (s senderStub) TransferConversation(context.Context, swt.Session, string, string) error {
+	return s.err
 }
 
-func (*operationSender) SendText(context.Context, swt.Session, string, string) (swt.SendResult, error) {
+type operationSender struct {
+	sid            string
+	acceptSID      string
+	transferSID    string
+	otherLoginName string
+	sentSID        string
+	err            error
+}
+
+func (s *operationSender) SendText(_ context.Context, _ swt.Session, sid, _ string) (swt.SendResult, error) {
+	s.sentSID = sid
 	return swt.SendResult{}, nil
 }
 
@@ -390,6 +465,11 @@ func (s *operationSender) AcceptTransfer(_ context.Context, _ swt.Session, sid s
 	return s.err
 }
 
+func (s *operationSender) TransferConversation(_ context.Context, _ swt.Session, sid, otherLoginName string) error {
+	s.transferSID, s.otherLoginName = sid, otherLoginName
+	return s.err
+}
+
 type partialSender struct {
 	textCalls  int
 	imageCalls int
@@ -423,6 +503,9 @@ func (*partialSender) ChangeContactName(context.Context, swt.Session, string, st
 }
 
 func (*partialSender) AcceptTransfer(context.Context, swt.Session, string) error { return nil }
+func (*partialSender) TransferConversation(context.Context, swt.Session, string, string) error {
+	return nil
+}
 
 func testUploadFetcher(t *testing.T) mediaFetcher {
 	t.Helper()

channels/shangwutong/internal/httpapi/server.go

@@ -185,6 +185,7 @@ func (s *Server) registerRoutes() {
 		return c.Status(http.StatusAccepted).JSON(fiber.Map{"accepted": true})
 	})
 	s.app.Post("/internal/operations/accept-transfer", s.acceptTransfer)
+	s.app.Post("/internal/operations/transfer-conversation", s.transferConversation)
 	s.app.Post("/webhooks/gochat/v1", s.handleWebhook)
 }
 
@@ -196,6 +197,52 @@ type acceptTransferRequest struct {
 	OccurredAt     time.Time `json:"occurred_at"`
 }
 
+type transferConversationRequest = acceptTransferRequest
+
+func (s *Server) transferConversation(c fiber.Ctx) error {
+	if ip := net.ParseIP(c.IP()); ip == nil || !ip.IsLoopback() {
+		return s.writeError(c, http.StatusForbidden, "forbidden", "loopback access required", false)
+	}
+	var request transferConversationRequest
+	if err := json.Unmarshal(c.Body(), &request); err != nil || request.InboxID <= 0 || request.ConversationID <= 0 || strings.TrimSpace(request.SID) == "" || strings.TrimSpace(request.EventID) == "" || request.OccurredAt.IsZero() {
+		return s.writeError(c, http.StatusUnprocessableEntity, "invalid_transfer_conversation", "inbox_id, conversation_id, sid, event_id and occurred_at are required", false)
+	}
+	request.SID, request.EventID = strings.TrimSpace(request.SID), strings.TrimSpace(request.EventID)
+	account, err := s.store.Reader().GetAccountByInboxID(c.Context(), request.InboxID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return s.writeError(c, http.StatusNotFound, "account_not_found", "connector account was not found", false)
+	}
+	if err != nil {
+		return s.writeError(c, http.StatusServiceUnavailable, "sqlite_unavailable", "account lookup failed", true)
+	}
+	queued, duplicate, err := s.store.EnqueueTransferConversation(c.Context(), store.OutboundOperationInput{
+		AccountID: account.ID, SWTSessionID: request.SID, EventID: request.EventID,
+		Operation: "transfer_conversation", OccurredAt: request.OccurredAt,
+	}, request.ConversationID)
+	if errors.Is(err, store.ErrOutboundConflict) {
+		return s.writeError(c, http.StatusConflict, "idempotency_conflict", "transfer conversation operation conflicts with an existing request", false)
+	}
+	if errors.Is(err, store.ErrOutboundAccountUnavailable) {
+		return s.writeError(c, http.StatusConflict, "account_unavailable", "connector account is disabled or offline", false)
+	}
+	if errors.Is(err, store.ErrOutboundSessionNotFound) {
+		return s.writeError(c, http.StatusConflict, "session_not_found", "sid is not mapped to the requested conversation", false)
+	}
+	if errors.Is(err, store.ErrOutboundSessionState) {
+		return s.writeError(c, http.StatusConflict, "session_not_transferable", "conversation is not held by another operator in a transferable state", false)
+	}
+	if err != nil {
+		return s.writeError(c, http.StatusServiceUnavailable, "queue_failed", "transfer conversation operation persistence failed", true)
+	}
+	status := http.StatusAccepted
+	if duplicate {
+		status = http.StatusOK
+	}
+	return c.Status(status).JSON(fiber.Map{
+		"accepted": true, "duplicate": duplicate, "event_id": request.EventID, "queue_id": queued.ID,
+	})
+}
+
 func (s *Server) acceptTransfer(c fiber.Ctx) error {
 	if ip := net.ParseIP(c.IP()); ip == nil || !ip.IsLoopback() {
 		return s.writeError(c, http.StatusForbidden, "forbidden", "loopback access required", false)

channels/shangwutong/internal/httpapi/server_test.go

@@ -408,11 +408,83 @@ func TestAcceptTransferRequestQueuesDurablyAndIsIdempotent(t *testing.T) {
 	}
 }
 
-func seedConversationState(t *testing.T, database *store.Store, accountID int64, sid string, conversationID int64, state string) {
+func TestTransferConversationRequestQueuesOtherOperatorsChat(t *testing.T) {
+	server, database, now := newTestServer(t)
+	lifecycle := webhookBody(t, now, "inbox_created", map[string]any{"channel_type": "shangwutong", "config_version": 1})
+	if response := doWebhook(t, server, lifecycle, now, "secret"); response.StatusCode != http.StatusAccepted {
+		t.Fatalf("bootstrap status = %d", response.StatusCode)
+	}
+	account, err := database.Reader().GetAccountByInboxID(context.Background(), 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	seedConversationState(t, database, account.ID, "visitor", 100, "5", "other-agent")
+	seedConversationState(t, database, account.ID, "already-transferring", 200, "7", "other-agent")
+	seedConversationState(t, database, account.ID, "owned", 300, "5", "agent")
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	served := make(chan error, 1)
+	go func() { served <- server.App().Listener(listener) }()
+	t.Cleanup(func() {
+		_ = server.Shutdown(context.Background())
+		<-served
+	})
+	request := func(value transferConversationRequest) *http.Response {
+		body, _ := json.Marshal(value)
+		response, requestErr := http.Post("http://"+listener.Addr().String()+"/internal/operations/transfer-conversation", "application/json", bytes.NewReader(body))
+		if requestErr != nil {
+			t.Fatal(requestErr)
+		}
+		return response
+	}
+
+	value := transferConversationRequest{InboxID: 10, ConversationID: 100, SID: "visitor", EventID: "transfer-conversation:visitor:1", OccurredAt: now}
+	first := request(value)
+	var firstAck map[string]any
+	if err := json.NewDecoder(first.Body).Decode(&firstAck); err != nil {
+		t.Fatal(err)
+	}
+	_ = first.Body.Close()
+	if first.StatusCode != http.StatusAccepted || firstAck["duplicate"] != false {
+		t.Fatalf("first status=%d ack=%#v", first.StatusCode, firstAck)
+	}
+	duplicate := request(value)
+	if body := readBody(duplicate); duplicate.StatusCode != http.StatusOK || !strings.Contains(body, `"duplicate":true`) {
+		t.Fatalf("duplicate status=%d body=%s", duplicate.StatusCode, body)
+	}
+	operation, err := database.Reader().GetOutboundOperationByEventID(context.Background(), value.EventID)
+	if err != nil || operation.Operation != "transfer_conversation" || !strings.Contains(operation.Payload, `"other_login_name":"other-agent"`) {
+		t.Fatalf("operation=%#v err=%v", operation, err)
+	}
+
+	for name, test := range map[string]struct {
+		request transferConversationRequest
+		code    string
+	}{
+		"wrong_state":   {transferConversationRequest{InboxID: 10, ConversationID: 200, SID: "already-transferring", EventID: "transfer-conversation:state", OccurredAt: now}, "session_not_transferable"},
+		"already_owned": {transferConversationRequest{InboxID: 10, ConversationID: 300, SID: "owned", EventID: "transfer-conversation:owned", OccurredAt: now}, "session_not_transferable"},
+		"forged_sid":    {transferConversationRequest{InboxID: 10, ConversationID: 100, SID: "missing", EventID: "transfer-conversation:missing", OccurredAt: now}, "session_not_found"},
+	} {
+		response := request(test.request)
+		body := readBody(response)
+		if response.StatusCode != http.StatusConflict || !strings.Contains(body, `"code":"`+test.code+`"`) {
+			t.Fatalf("%s status=%d body=%s", name, response.StatusCode, body)
+		}
+	}
+}
+
+func seedConversationState(t *testing.T, database *store.Store, accountID int64, sid string, conversationID int64, state string, assignee ...string) {
 	t.Helper()
-	if _, err := database.Writer().UpsertConversationMap(context.Background(), dbgen.UpsertConversationMapParams{
+	params := dbgen.UpsertConversationMapParams{
 		AccountID: accountID, SwtSid: sid, GochatContactSourceID: sid, GochatConversationID: &conversationID,
-	}); err != nil {
+	}
+	if len(assignee) > 0 {
+		params.SwtAssigneeName = &assignee[0]
+	}
+	if _, err := database.Writer().UpsertConversationMap(context.Background(), params); err != nil {
 		t.Fatal(err)
 	}
 	if _, err := database.Writer().InsertInboundEvent(context.Background(), dbgen.InsertInboundEventParams{

channels/shangwutong/internal/store/outbound.go

@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
@@ -208,6 +209,77 @@ func (s *Store) EnqueueAcceptTransfer(ctx context.Context, input OutboundOperati
 	return queued, duplicate, err
 }
 
+func (s *Store) EnqueueTransferConversation(ctx context.Context, input OutboundOperationInput, conversationID int64) (*dbgen.OutboundOperation, bool, error) {
+	if input.AccountID <= 0 || strings.TrimSpace(input.SWTSessionID) == "" || strings.TrimSpace(input.EventID) == "" ||
+		input.Operation != "transfer_conversation" || input.OccurredAt.IsZero() || conversationID <= 0 {
+		return nil, false, errors.New("valid transfer conversation fields are required")
+	}
+	var queued *dbgen.OutboundOperation
+	var duplicate bool
+	err := s.WithTx(ctx, func(queries *dbgen.Queries) error {
+		existing, err := queries.GetOutboundOperationByEventID(ctx, input.EventID)
+		if err == nil {
+			var payload struct {
+				ConversationID int64 `json:"conversation_id"`
+			}
+			if existing.AccountID != input.AccountID || existing.SwtSid != input.SWTSessionID || existing.Operation != input.Operation ||
+				!existing.OccurredAt.Equal(input.OccurredAt) || json.Unmarshal([]byte(existing.Payload), &payload) != nil || payload.ConversationID != conversationID {
+				return ErrOutboundConflict
+			}
+			queued, duplicate = existing, true
+			return nil
+		}
+		if !errors.Is(err, sql.ErrNoRows) {
+			return err
+		}
+		account, err := queries.GetAccountByID(ctx, input.AccountID)
+		if err != nil {
+			return err
+		}
+		if !AccountRunnable(account) {
+			return ErrOutboundAccountUnavailable
+		}
+		mapping, err := queries.GetConversationMap(ctx, dbgen.GetConversationMapParams{AccountID: input.AccountID, SwtSid: input.SWTSessionID})
+		if errors.Is(err, sql.ErrNoRows) || err == nil && (mapping.GochatConversationID == nil || *mapping.GochatConversationID != conversationID) {
+			return ErrOutboundSessionNotFound
+		}
+		if err != nil {
+			return err
+		}
+		otherLoginName := ""
+		if mapping.SwtAssigneeName != nil {
+			otherLoginName = strings.TrimSpace(*mapping.SwtAssigneeName)
+		}
+		currentLoginName := account.Username
+		if account.LoginName != nil && strings.TrimSpace(*account.LoginName) != "" {
+			currentLoginName = strings.TrimSpace(*account.LoginName)
+		}
+		if otherLoginName == "" || strings.EqualFold(otherLoginName, currentLoginName) {
+			return ErrOutboundSessionState
+		}
+		state, err := queries.GetLatestConversationState(ctx, dbgen.GetLatestConversationStateParams{AccountID: input.AccountID, SwtSid: input.SWTSessionID})
+		if errors.Is(err, sql.ErrNoRows) || err == nil && (state == nil || strings.TrimSpace(*state) != "5") {
+			return ErrOutboundSessionState
+		}
+		if err != nil {
+			return err
+		}
+		payload, err := json.Marshal(struct {
+			ConversationID int64  `json:"conversation_id"`
+			OtherLoginName string `json:"other_login_name"`
+		}{conversationID, otherLoginName})
+		if err != nil {
+			return err
+		}
+		queued, err = queries.InsertOutboundOperation(ctx, dbgen.InsertOutboundOperationParams{
+			AccountID: input.AccountID, SwtSid: input.SWTSessionID, EventID: input.EventID,
+			Operation: input.Operation, Payload: string(payload), OccurredAt: input.OccurredAt,
+		})
+		return err
+	})
+	return queued, duplicate, err
+}
+
 func outboundOperationConflicts(existing *dbgen.OutboundOperation, input OutboundOperationInput) bool {
 	return existing.AccountID != input.AccountID || existing.SwtSid != input.SWTSessionID || existing.Operation != input.Operation ||
 		existing.Payload != input.Payload || !existing.OccurredAt.Equal(input.OccurredAt)

channels/shangwutong/internal/store/store_test.go

@@ -211,6 +211,36 @@ func TestRollbackMigrationRefusesAcceptTransferOperations(t *testing.T) {
 	}
 }
 
+func TestRollbackMigrationRefusesTransferConversationOperations(t *testing.T) {
+	db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE outbound_operations (
+		id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, swt_sid TEXT NOT NULL,
+		event_id TEXT NOT NULL UNIQUE, operation TEXT NOT NULL, payload TEXT NOT NULL,
+		occurred_at DATETIME NOT NULL, delivery_status TEXT NOT NULL, claimed_at DATETIME,
+		attempts INTEGER NOT NULL, next_attempt_at DATETIME, last_error TEXT,
+		created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`INSERT INTO outbound_operations(account_id,swt_sid,event_id,operation,payload,occurred_at,delivery_status,attempts,created_at,updated_at) VALUES (1,'sid','event','transfer_conversation','{}',CURRENT_TIMESTAMP,'pending',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)`); err != nil {
+		t.Fatal(err)
+	}
+	rollback, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", "005_add_transfer_conversation_operation.down.sql"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(string(rollback)); err == nil {
+		t.Fatal("expected rollback to reject transfer_conversation")
+	}
+	var count int
+	if err := db.QueryRow("SELECT COUNT(*) FROM outbound_operations").Scan(&count); err != nil || count != 1 {
+		t.Fatalf("operation count = %d, err=%v", count, err)
+	}
+}
+
 func TestRollbackMigrationKeepsOldOperationsWhenRenameIsAbsent(t *testing.T) {
 	db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db"))
 	if err != nil {
@@ -590,7 +620,7 @@ func TestOnlineBackupCanBeOpenedReadOnly(t *testing.T) {
 		t.Fatalf("backup mode = %v", info.Mode().Perm())
 	}
 	version, err := InspectDatabase(ctx, backup)
-	if err != nil || version != 4 {
+	if err != nil || version != 5 {
 		t.Fatalf("backup version = %d, %v", version, err)
 	}
 }