package channeladmin

import (
	"context"
	"fmt"
	"os"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/app"
	"github.com/tidbcloud/AgentTeam/internal/store/channel"
	receiptstore "github.com/tidbcloud/AgentTeam/internal/store/channeladmin"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/plan"
	"github.com/tidbcloud/AgentTeam/internal/store/workspace"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

var channelManagementTiDB testkit.TiDBCoreDBFixture

func TestMain(m *testing.M) {
	os.Exit(testkit.RunTestsWithTiDBCoreDBFixture(m, "channelmanagement", &channelManagementTiDB))
}

// fakeChannelStore records CreateChannelInTx calls and, critically, observes
// whether a receipt row already exists in the caller's transaction when the
// mutation runs. That observation is the deterministic RED/GREEN hinge for the
// receipt-first concurrency contract (#t2347): the receipt must be reserved
// before the mutation so a concurrent duplicate can never perform the mutation.
type fakeChannelStore struct {
	mu                            sync.Mutex
	createCalls                   int
	receiptReservedBeforeMutation bool
	mutationErr                   error
}

func (f *fakeChannelStore) CreateChannelInTx(_ context.Context, tx coredb.Tx, workspaceID, ownerMemberID, name, description, visibility string) (message.ChannelSummary, error) {
	var count int
	if err := tx.QueryRow(context.Background(), `
SELECT COUNT(*) FROM channel_operation_receipts
`).Scan(&count); err != nil {
		return message.ChannelSummary{}, err
	}
	f.mu.Lock()
	f.createCalls++
	f.receiptReservedBeforeMutation = count > 0
	mutationErr := f.mutationErr
	f.mu.Unlock()
	if mutationErr != nil {
		return message.ChannelSummary{}, mutationErr
	}
	return message.ChannelSummary{
		ChannelID:   "chn_fake_" + name,
		Slug:        "fake-" + name,
		DisplayName: name,
		Description: description,
		Visibility:  visibility,
	}, nil
}

func (f *fakeChannelStore) ListManagedIMChannelsForUserPage(
	context.Context,
	string,
	string,
	channel.ManagedIMChannelListOptions,
) (channel.ManagedIMChannelPage, error) {
	return channel.ManagedIMChannelPage{}, nil
}

type fakeOwnerResolver struct {
	userID   string
	memberID string
}

func (o fakeOwnerResolver) AgentManagedByUser(context.Context, string, string) (string, error) {
	return o.userID, nil
}

type fakeWorkspaceResolver struct {
	workspaceID string
	memberID    string
	userID      string
}

func (r fakeWorkspaceResolver) GetWorkspaceForUser(_ context.Context, _ string, workspaceID string) (workspace.Summary, error) {
	return workspace.Summary{WorkspaceID: workspaceID, MemberID: r.memberID, MemberUserID: r.userID}, nil
}

type recordingRealtime struct {
	mu                sync.Mutex
	channelUpdated    int
	membersUpdated    int
	subscribedRooms   []string
	unsubscribedRooms []string
	wakeClears        []wakeClear
	unreadUpdated     []unread.UnreadDelta
}

type wakeClear struct {
	scopeID string
	agentID string
}

func (r *recordingRealtime) EmitChannelUpdated(string) {
	r.mu.Lock()
	r.channelUpdated++
	r.mu.Unlock()
}

func (r *recordingRealtime) EmitChannelMembersUpdated(string, string) {
	r.mu.Lock()
	r.membersUpdated++
	r.mu.Unlock()
}

func (r *recordingRealtime) SubscribeScopeForUser(_, _, scopeID string) {
	r.mu.Lock()
	r.subscribedRooms = append(r.subscribedRooms, scopeID)
	r.mu.Unlock()
}

func (r *recordingRealtime) UnsubscribeScopeForUser(_, _, scopeID string) {
	r.mu.Lock()
	r.unsubscribedRooms = append(r.unsubscribedRooms, scopeID)
	r.mu.Unlock()
}

func (r *recordingRealtime) ClearAgentWakeStatusesForChannel(_, scopeID, agentID string) {
	r.mu.Lock()
	r.wakeClears = append(r.wakeClears, wakeClear{scopeID: scopeID, agentID: agentID})
	r.mu.Unlock()
}

func (r *recordingRealtime) EmitUnreadUpdated(_ string, delta unread.UnreadDelta) {
	r.mu.Lock()
	r.unreadUpdated = append(r.unreadUpdated, delta)
	r.mu.Unlock()
}

func newChannelManagementTestService(t *testing.T, ctx context.Context, ownerUserID, ownerMemberID string) (*Service, *fakeChannelStore, *recordingRealtime, coredb.DB) {
	t.Helper()
	return buildChannelManagementService(t, ctx, channelManagementTiDB.SerializedRollbackDB(t, ctx), ownerUserID, ownerMemberID)
}

// newPooledTestDB opens a real pooled TiDB for the package fixture so genuine
// concurrency (separate connections, real commits, primary-key lock ordering)
// can be exercised. The isolated migrated database is reclaimed by the testkit.
func newPooledTestDB(t *testing.T, ctx context.Context) coredb.DB {
	t.Helper()
	rootDSN := channelManagementTiDB.RootDSN(t)
	dsn := testkit.StartInitializedTiDBWithRootDSN(t, rootDSN)
	sqlDB, err := testkit.OpenTiDBContext(ctx, dsn)
	require.NoError(t, err)
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	t.Cleanup(func() { _ = sqlDB.Close() })
	return coredb.WrapSQLDB(sqlDB)
}

// buildChannelManagementService seeds the workspace and assembles the service on
// the given db. The seed runs inside one transaction so SET FOREIGN_KEY_CHECKS
// and the workspace insert share a connection (required for the pooled DB used
// by concurrency tests; harmless for the rollback fixture).
func buildChannelManagementService(t *testing.T, ctx context.Context, db coredb.DB, ownerUserID, ownerMemberID string) (*Service, *fakeChannelStore, *recordingRealtime, coredb.DB) {
	t.Helper()
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		if _, err := tx.Exec(ctx, `SET FOREIGN_KEY_CHECKS = 0`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO workspaces (id, organization_id, slug, name, created_at, updated_at) VALUES ($1, 'org_test', 'wsp-channel-mgmt', 'test', NOW(), NOW())`, "wsp_channel_mgmt"); err != nil {
			return err
		}
		// Human owner member + its machine + the agent, so the tx-aware owner
		// resolver (access.AgentManagedByUser + LoadHumanWorkspaceMemberForUpdate)
		// resolves ownerUserID -> mem_owner on the caller's transaction.
		if _, err := tx.Exec(ctx, `INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, status, created_at, updated_at) VALUES ('mem_owner', 'wsp_channel_mgmt', 'human', 'owner', $1, 'owner', 'active', NOW(), NOW())`, ownerUserID); err != nil {
			return err
		}
		// The Agent's own member row, required by LockAndRevalidateAgentPrincipalManager.
		if _, err := tx.Exec(ctx, `INSERT INTO members (id, workspace_id, type, display_name, agent_display_name_key, status, created_at, updated_at) VALUES ('mem_agent', 'wsp_channel_mgmt', 'agent', 'agent', 'agent', 'active', NOW(), NOW())`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO machines (id, workspace_id, created_by_user_id, name, api_key, status, created_at, updated_at) VALUES ('mac_owner', 'wsp_channel_mgmt', $1, 'mac', 'key', 'active', NOW(), NOW())`, ownerUserID); err != nil {
			return err
		}
		_, err := tx.Exec(ctx, `INSERT INTO agents (id, workspace_id, machine_id, member_id, name, runtime, model, status, description, activity_detail, created_at, updated_at) VALUES ('agt_owner', 'wsp_channel_mgmt', 'mac_owner', 'mem_agent', 'agent', 'local', 'm', 'active', '', '', NOW(), NOW())`)
		return err
	}))
	store := &fakeChannelStore{}
	realtime := &recordingRealtime{}
	service := New(Dependencies{
		DB:         db,
		Channels:   store,
		Apps:       app.NewChannelManagementPort(app.New(db, app.Options{})),
		Owners:     fakeOwnerResolver{userID: ownerUserID, memberID: ownerMemberID},
		Workspaces: fakeWorkspaceResolver{workspaceID: "wsp_channel_mgmt", memberID: ownerMemberID, userID: ownerUserID},
		Realtime:   realtime,
	})
	return service, store, realtime, db
}

func mustExec(t *testing.T, ctx context.Context, db coredb.QueryExecer, query string, args ...any) {
	t.Helper()
	_, err := db.Exec(ctx, query, args...)
	require.NoError(t, err)
}

// TestCreateReservesReceiptBeforeMutation is the RED hinge for #t2347: the
// receipt must be reserved (INSERT) before the mutation runs, so a concurrent
// duplicate collides on the receipt primary key instead of performing the
// mutation. On the old mutation-then-receipt order, the fake observes zero
// receipt rows when CreateChannelInTx runs, so this fails.
func TestCreateReservesReceiptBeforeMutation(t *testing.T) {
	ctx := context.Background()
	service, store, _, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	_, err := service.Create(ctx, agent, CreateInput{RequestID: "req-1", DisplayName: "release-room"})
	require.NoError(t, err)

	store.mu.Lock()
	sawReceipt := store.receiptReservedBeforeMutation
	calls := store.createCalls
	store.mu.Unlock()
	require.Equal(t, 1, calls, "first create must run the mutation exactly once")
	require.True(t, sawReceipt, "receipt must be reserved before the mutation runs (receipt-first)")
}

// TestCreateSequentialReplay verifies that replaying the same idempotency key
// returns the cached result without re-running the mutation.
func TestCreateSequentialReplay(t *testing.T) {
	ctx := context.Background()
	service, store, _, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	first, err := service.Create(ctx, agent, CreateInput{RequestID: "req-replay", DisplayName: "release-room"})
	require.NoError(t, err)
	replayed, err := service.Create(ctx, agent, CreateInput{RequestID: "req-replay", DisplayName: "release-room"})
	require.NoError(t, err)
	require.Equal(t, first.Channel.ChannelID, replayed.Channel.ChannelID, "replay must return the same channel")

	store.mu.Lock()
	calls := store.createCalls
	store.mu.Unlock()
	require.Equal(t, 1, calls, "replay must not re-run the mutation")
}

// TestCreateReplayEmitsNoRealtime verifies the replay path does not emit
// realtime (only the initial mutation does).
func TestCreateReplayEmitsNoRealtime(t *testing.T) {
	ctx := context.Background()
	service, _, realtime, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	_, err := service.Create(ctx, agent, CreateInput{RequestID: "req-nort", DisplayName: "release-room"})
	require.NoError(t, err)
	realtime.mu.Lock()
	firstEmit := realtime.channelUpdated
	firstSubscriptions := append([]string(nil), realtime.subscribedRooms...)
	realtime.mu.Unlock()
	require.Equal(t, 1, firstEmit, "initial mutation emits once")
	require.Len(t, firstSubscriptions, 1, "initial create subscribes the owner to the new channel")

	_, err = service.Create(ctx, agent, CreateInput{RequestID: "req-nort", DisplayName: "release-room"})
	require.NoError(t, err)
	realtime.mu.Lock()
	afterReplay := realtime.channelUpdated
	afterReplaySubscriptions := append([]string(nil), realtime.subscribedRooms...)
	realtime.mu.Unlock()
	require.Equal(t, 1, afterReplay, "replay must not emit realtime")
	require.Equal(t, firstSubscriptions, afterReplaySubscriptions, "replay must not subscribe the owner twice")
}

// TestCreateCrossActionConflict verifies that reusing an idempotency key with a
// different action returns a stable channel_idempotency_conflict.
func TestCreateCrossActionConflict(t *testing.T) {
	ctx := context.Background()
	service, _, _, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	_, err := service.Create(ctx, agent, CreateInput{RequestID: "req-conflict", DisplayName: "release-room"})
	require.NoError(t, err)

	// Reuse the same key for an update (different action). updateChannelInTx is
	// never reached because the receipt lookup detects the action mismatch first.
	_, err = service.Update(ctx, agent, UpdateInput{RequestID: "req-conflict", ChannelID: "chn_fake_release-room", ExpectedVersion: 1, Description: strPtr("notes")})
	require.ErrorIs(t, err, ErrRequestIdConflict)
}

// TestCreateConcurrentSameKeyReplaysOnce exercises REAL TiDB concurrency: two
// goroutines create with the same idempotency key + digest. The receipt
// primary-key lock serializes them — exactly one reserves the key and mutates;
// the loser's reservation blocks until the winner commits, then collides,
// rolls back, and replays the winner's committed result. On the old
// mutation-then-receipt order both goroutines mutate before colliding on the
// receipt key, so createCalls would be 2 (RED); receipt-first holds it to 1.
func TestCreateConcurrentSameKeyReplaysOnce(t *testing.T) {
	ctx := context.Background()
	service, store, _, _ := buildChannelManagementService(t, ctx, newPooledTestDB(t, ctx), "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	input := CreateInput{RequestID: "req-concurrent", DisplayName: "concurrent-room"}

	results := make([]CreateResult, 2)
	errs := make([]error, 2)
	var wg sync.WaitGroup
	for i := range results {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			results[i], errs[i] = service.Create(ctx, agent, input)
		}(i)
	}
	wg.Wait()

	require.NoError(t, errs[0])
	require.NoError(t, errs[1])
	require.NotEmpty(t, results[0].Channel.ChannelID)
	require.Equal(t, results[0].Channel.ChannelID, results[1].Channel.ChannelID,
		"concurrent same-key create must return the same channel")

	store.mu.Lock()
	calls := store.createCalls
	store.mu.Unlock()
	require.Equal(t, 1, calls, "concurrent same-key create must mutate exactly once")
}

// TestCreateBusinessUniqueNotClassifiedAsReceiptConflict verifies that a
// business unique violation raised by the mutation (e.g. a slug collision) is
// NOT misclassified as a receipt/idempotency conflict. The receipt-first order
// makes this structurally sound: the receipt reservation succeeds, the mutation
// then fails with its canonical error, and the whole transaction (including the
// reserved placeholder) rolls back — the caller sees the business error, never
// errReceiptPKConflict/ErrRequestIdConflict.
func TestCreateClassifiesBusinessConflictAndQuotaWithoutReceiptConflict(t *testing.T) {
	ctx := context.Background()
	service, store, _, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	store.mu.Lock()
	store.mutationErr = channel.ChannelSlugConflictError{}
	store.mu.Unlock()
	_, err := service.Create(ctx, agent, CreateInput{RequestID: "req-slug-conflict", DisplayName: "dup-room"})
	require.ErrorIs(t, err, ErrSlugConflict)
	require.NotErrorIs(t, err, ErrVersionConflict, "create slug conflict must not masquerade as update CAS conflict")
	require.NotErrorIs(t, err, ErrRequestIdConflict, "business unique must not be misclassified as receipt conflict")

	store.mu.Lock()
	store.mutationErr = plan.QuotaExceededError{Metric: plan.MetricChannels, Limit: 10, Current: 10, Delta: 1}
	store.mu.Unlock()
	_, err = service.Create(ctx, agent, CreateInput{RequestID: "req-quota", DisplayName: "over-quota"})
	require.ErrorIs(t, err, ErrQuotaExceeded, "typed quota errors do not unwrap to storeerr and need explicit classification")
	require.NotErrorIs(t, err, ErrRequestIdConflict)
}

// seedOwnedChannel inserts a real plain-IM channel owned by ownerMemberID
// (resolved from ownerUserID) at version 1, with the owner joined, so the
// service update path (access.LoadHumanChannelAccess + CAS) can run.
func seedOwnedChannel(t *testing.T, ctx context.Context, db coredb.QueryExecer, channelID, ownerMemberID, ownerUserID string) {
	t.Helper()
	mustExec(t, ctx, db, `INSERT INTO scopes (id, workspace_id, type, identity_key, status, created_at, updated_at) VALUES ($1, 'wsp_channel_mgmt', 'channel', $2, 'active', NOW(), NOW())`, channelID, "channel:"+channelID)
	mustExec(t, ctx, db, `INSERT INTO scope_channel_profiles (scope_id, workspace_id, slug, display_name, description, visibility, subsystem, owner_member_id, is_all_channel, version, created_at, updated_at) VALUES ($1, 'wsp_channel_mgmt', $2, $2, '', 'private', 'im', $3, FALSE, 1, NOW(), NOW())`, channelID, channelID+"-slug", ownerMemberID)
	mustExec(t, ctx, db, `INSERT INTO scope_memberships (scope_id, member_id, visibility, owner_member_id, last_read_seq, last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq, effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at) VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, NOW(), NOW(), NULL, NULL)`, channelID, ownerMemberID)
}

// TestUpdateConcurrentSameKeyReplaysOnce exercises REAL TiDB concurrency on the
// update path. Two goroutines update the same channel with the same
// idempotency key + digest. Receipt-first makes exactly one perform the CAS
// (version 1 -> 2); the loser collides on the receipt primary key, rolls back,
// and replays the winner's committed result. On the old mutation-then-receipt
// order the loser would instead hit the CAS version fence and surface a version
// conflict rather than the winner's result.
func TestUpdateConcurrentSameKeyReplaysOnce(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_upd", "mem_owner", "usr_owner")

	input := UpdateInput{RequestID: "req-upd-concurrent", ChannelID: "chn_upd", ExpectedVersion: 1, Description: strPtr("updated")}
	results := make([]UpdateResult, 2)
	errs := make([]error, 2)
	var wg sync.WaitGroup
	for i := range results {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			results[i], errs[i] = service.Update(ctx, agent, input)
		}(i)
	}
	wg.Wait()

	require.NoError(t, errs[0])
	require.NoError(t, errs[1])
	require.Equal(t, int64(2), results[0].Channel.Version, "winner CAS result is version 2")
	require.Equal(t, results[0].Channel.Version, results[1].Channel.Version, "concurrent same-key update must return the same version")

	var finalVersion int64
	require.NoError(t, db.QueryRow(ctx, `SELECT version FROM scope_channel_profiles WHERE scope_id = $1`, "chn_upd").Scan(&finalVersion))
	require.Equal(t, int64(2), finalVersion, "concurrent same-key update must apply the CAS mutation exactly once (version 1 -> 2)")
}

// TestUpdateCASLostUpdateRejected validates the optimistic-concurrency fence:
// after a successful update (version 1 -> 2), a second update carrying the stale
// expectedVersion=1 is rejected with ErrVersionConflict instead of a lost update.
func TestUpdateCASLostUpdateRejected(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_cas", "mem_owner", "usr_owner")

	first, err := service.Update(ctx, agent, UpdateInput{RequestID: "req-cas-1", ChannelID: "chn_cas", ExpectedVersion: 1, Description: strPtr("v2")})
	require.NoError(t, err)
	require.Equal(t, int64(2), first.Channel.Version)

	_, err = service.Update(ctx, agent, UpdateInput{RequestID: "req-cas-2", ChannelID: "chn_cas", ExpectedVersion: 1, Description: strPtr("v3")})
	require.ErrorIs(t, err, ErrVersionConflict, "stale expectedVersion must be rejected as a lost update")
}

// TestUpdateRejectsNonIMCarrier validates the ManagedIMExec policy's post-lock
// carrier revalidation: a channel whose carrier is not plain IM (here flipped to
// kb) is rejected even though ownership is otherwise valid.
func TestUpdateRejectsNonIMCarrier(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_kb", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'kb' WHERE scope_id = $1`, "chn_kb")

	_, err := service.Update(ctx, agent, UpdateInput{RequestID: "req-kb", ChannelID: "chn_kb", ExpectedVersion: 1, Description: strPtr("x")})
	require.ErrorIs(t, err, ErrNotFound, "ManagedIMExec policy must reject non-IM carriers after the lock")
}

// TestUpdateRejectsOwnerTransferRace proves the post-lock owner revalidation
// fences an owner transfer that happens after the caller's precheck. tx-A locks
// the profile row and holds it; the Update goroutine completes its precheck
// (owner still mem_owner) and blocks on the profile FOR UPDATE; tx-A then
// transfers ownership and commits; the Update acquires the lock, revalidates the
// owner against the locked row, and must reject without writing the profile.
func TestUpdateRejectsOwnerTransferRace(t *testing.T) {
	// Serialized: uses package-global app.PreProfileLockHookForTest.
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_xfer", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, status, created_at, updated_at) VALUES ('mem_other', 'wsp_channel_mgmt', 'human', 'other', 'usr_other', 'member', 'active', NOW(), NOW())`)

	// Deterministic handshake: the canonical primitive fires the seam AFTER the
	// channel-access precheck and BEFORE the profile FOR UPDATE, proving B has a
	// stale precheck snapshot when A transfers ownership. No timing guess.
	bPastPrecheck := make(chan struct{})
	bRelease := make(chan struct{})
	app.PreProfileLockHookForTest = func(context.Context) {
		select {
		case <-bPastPrecheck:
		default:
			close(bPastPrecheck)
		}
		<-bRelease
	}
	defer func() { app.PreProfileLockHookForTest = nil }()

	ready := make(chan struct{})
	release := make(chan struct{})
	transferErr := make(chan error, 1)
	go func() {
		txA, err := db.BeginTx(ctx, coredb.TxOptions{})
		if err != nil {
			transferErr <- err
			return
		}
		var lockedID string
		if err := txA.QueryRow(ctx, `SELECT scope_id FROM scope_channel_profiles WHERE workspace_id = $1 AND scope_id = $2 FOR UPDATE`, "wsp_channel_mgmt", "chn_xfer").Scan(&lockedID); err != nil {
			_ = txA.Rollback(ctx)
			transferErr <- err
			return
		}
		close(ready)
		<-release
		if _, err := txA.Exec(ctx, `UPDATE scope_channel_profiles SET owner_member_id = 'mem_other' WHERE scope_id = $1`, "chn_xfer"); err != nil {
			_ = txA.Rollback(ctx)
			transferErr <- err
			return
		}
		transferErr <- txA.Commit(ctx)
	}()

	select {
	case <-ready:
	case err := <-transferErr:
		t.Fatalf("tx-A failed before locking the profile: %v", err)
	case <-time.After(30 * time.Second):
		t.Fatal("tx-A did not lock the profile in time")
	}

	updateErr := make(chan error, 1)
	go func() {
		_, err := service.Update(ctx, agent, UpdateInput{RequestID: "req-xfer", ChannelID: "chn_xfer", ExpectedVersion: 1, Description: strPtr("changed")})
		updateErr <- err
	}()

	select {
	case <-bPastPrecheck:
	case err := <-updateErr:
		t.Fatalf("update finished before reaching the profile lock (race not exercised): %v", err)
	case <-time.After(30 * time.Second):
		t.Fatal("update never reached the profile-lock precheck")
	}

	close(release) // tx-A transfers ownership and commits.
	require.NoError(t, <-transferErr)
	close(bRelease) // B proceeds to the profile FOR UPDATE on the transferred row.

	select {
	case err := <-updateErr:
		require.ErrorIs(t, err, ErrForbidden, "owner transfer after precheck must be rejected by post-lock revalidation")
	case <-time.After(15 * time.Second):
		t.Fatal("update did not complete after owner transfer")
	}
	var description string
	require.NoError(t, db.QueryRow(ctx, `SELECT description FROM scope_channel_profiles WHERE scope_id = $1`, "chn_xfer").Scan(&description))
	require.Equal(t, "", description, "rejected update must not write the profile")
}

// TestMutationReservesReceiptBeforePrincipalLock proves the receipt-first fence
// ordering (#t2347): the idempotency receipt placeholder is reserved on the
// caller's tx BEFORE resolveOwnerInTx takes the Agent principal lock, so the
// receipt primary key — not the principal lock — is the first concurrency fence.
// The pre-principal-lock seam counts receipts on the tx: under the old
// (resolveOwnerInTx-before-receipt) order the count was 0; the receipt-first
// reorder makes it 1.
func TestMutationReservesReceiptBeforePrincipalLock(t *testing.T) {
	ctx := context.Background()
	service, _, _, _ := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	var receiptCountAtPrincipalLock int
	prePrincipalLockHookForTest = func(_ context.Context, tx coredb.Tx) {
		var count int
		if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM channel_operation_receipts WHERE workspace_id = $1 AND applicant_agent_id = $2`, "wsp_channel_mgmt", "agt_owner").Scan(&count); err != nil {
			t.Errorf("count receipts at principal lock: %v", err)
			return
		}
		receiptCountAtPrincipalLock = count
	}
	defer func() { prePrincipalLockHookForTest = nil }()

	_, err := service.Create(ctx, agent, CreateInput{RequestID: "req_lock_order", DisplayName: "lock-order", Visibility: "private"})
	require.NoError(t, err)
	require.Equal(t, 1, receiptCountAtPrincipalLock, "receipt placeholder must be reserved before the Agent principal lock (receipt-first fence)")
}

// TestClassifyCommittedReceiptMissIsNotConflict proves the shared classifier
// never maps a not-found receipt to a 409 conflict: only a FOUND receipt with a
// mismatched action/digest is a conflict. A miss (including the loser-miss
// visibility window) must stay internal so the client retries the same request
// id rather than risk a duplicate mutation.
func TestClassifyCommittedReceiptMissIsNotConflict(t *testing.T) {
	outcome, err := classifyCommittedReceipt(receiptstore.Receipt{}, false, nil, string(ActionChannelCreate), digestCreate(CreateInput{RequestID: "req", DisplayName: "n", Visibility: "private"}))
	require.NoError(t, err)
	require.Equal(t, receiptReplayMiss, outcome, "not-found receipt must classify as Miss, not Conflict")
}

// TestCreateReplayAfterOwnerOffboard proves the receipt-first pre-lookup serves
// an exact replay WITHOUT entering the mutation tx or reaching the principal
// lock, so the stored result survives a principal change that would otherwise
// fail authority (#t2347). A first Create commits the receipt; the owner is then
// offboarded (status=inactive) so any tx that re-entered resolveOwnerInTx would
// be rejected; the same-digest replay must return the original result unchanged
// and must NOT fire the principal-lock hook.
func TestCreateReplayAfterOwnerOffboard(t *testing.T) {
	ctx := context.Background()
	service, _, _, db := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	first, err := service.Create(ctx, agent, CreateInput{RequestID: "req_replay_offboard", DisplayName: "replay-channel", Visibility: "private"})
	require.NoError(t, err)
	require.Equal(t, "replay-channel", first.Channel.DisplayName)

	// Offboard the owner: a tx that re-entered resolveOwnerInTx would now fail
	// authority (LoadHumanWorkspaceMember rejects inactive members), proving the
	// replay must be served pre-tx.
	_, err = db.Exec(ctx, `UPDATE members SET status = 'inactive', updated_at = NOW() WHERE id = 'mem_owner'`)
	require.NoError(t, err)

	var hookFired int
	prePrincipalLockHookForTest = func(context.Context, coredb.Tx) { hookFired++ }
	defer func() { prePrincipalLockHookForTest = nil }()

	replayed, err := service.Create(ctx, agent, CreateInput{RequestID: "req_replay_offboard", DisplayName: "replay-channel", Visibility: "private"})
	require.NoError(t, err)
	require.Equal(t, first.Channel, replayed.Channel, "exact replay must return the committed result unchanged")
	require.Equal(t, 0, hookFired, "exact replay must be served by the pre-lookup without reaching the principal lock")
}

// TestUpdateRejectsOwnerRemovalRace proves the locked Human-manager-member
// revalidation fences member removal/inactivation that happens after the caller's
// precheck. tx-A locks the owner member row and holds it; the Update goroutine's
// resolveOwnerInTx fires the principal-lock seam after owner resolution and
// blocks; tx-A then marks the member inactive and commits; the Update's
// LockAndRevalidateAgentPrincipalManager must reject.
func TestUpdateRejectsOwnerRemovalRace(t *testing.T) {
	// Serialized: uses package-global prePrincipalLockHookForTest.
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_remove", "mem_owner", "usr_owner")

	bPastResolve := make(chan struct{})
	bRelease := make(chan struct{})
	prePrincipalLockHookForTest = func(context.Context, coredb.Tx) {
		select {
		case <-bPastResolve:
		default:
			close(bPastResolve)
		}
		<-bRelease
	}
	defer func() { prePrincipalLockHookForTest = nil }()

	ready := make(chan struct{})
	release := make(chan struct{})
	removalErr := make(chan error, 1)
	go func() {
		txA, err := db.BeginTx(ctx, coredb.TxOptions{})
		if err != nil {
			removalErr <- err
			return
		}
		var lockedMemberID string
		if err := txA.QueryRow(ctx, `SELECT id FROM members WHERE workspace_id = $1 AND id = $2 FOR UPDATE`, "wsp_channel_mgmt", "mem_owner").Scan(&lockedMemberID); err != nil {
			_ = txA.Rollback(ctx)
			removalErr <- err
			return
		}
		close(ready)
		<-release
		if _, err := txA.Exec(ctx, `UPDATE members SET status = 'inactive' WHERE id = $1`, "mem_owner"); err != nil {
			_ = txA.Rollback(ctx)
			removalErr <- err
			return
		}
		removalErr <- txA.Commit(ctx)
	}()

	select {
	case <-ready:
	case err := <-removalErr:
		t.Fatalf("tx-A failed before locking the owner member: %v", err)
	case <-time.After(30 * time.Second):
		t.Fatal("tx-A did not lock the owner member in time")
	}

	updateErr := make(chan error, 1)
	go func() {
		_, err := service.Update(ctx, agent, UpdateInput{RequestID: "req-remove", ChannelID: "chn_remove", ExpectedVersion: 1, Description: strPtr("changed")})
		updateErr <- err
	}()

	select {
	case <-bPastResolve:
	case err := <-updateErr:
		t.Fatalf("update finished before reaching the principal lock (race not exercised): %v", err)
	case <-time.After(30 * time.Second):
		t.Fatal("update never reached the principal-lock precheck")
	}

	close(release) // tx-A deactivates the owner member and commits.
	require.NoError(t, <-removalErr)
	close(bRelease) // B proceeds to LockAndRevalidate on the inactive member.

	select {
	case err := <-updateErr:
		require.ErrorIs(t, err, ErrForbidden, "owner removal after precheck must be rejected by locked member revalidation")
	case <-time.After(15 * time.Second):
		t.Fatal("update did not complete after owner removal")
	}
}

// TestUpdateManagedIMRejectsAllCarriers is the table-driven carrier AC for the
// ManagedIMExec policy: every non-plain-IM carrier is rejected after the lock.
func TestUpdateManagedIMRejectsAllCarriers(t *testing.T) {
	ctx := context.Background()

	cases := []struct {
		name      string
		setup     func(t *testing.T, db coredb.DB, channelID string)
		wantError error
	}{
		{
			name: "all-channel",
			setup: func(t *testing.T, db coredb.DB, channelID string) {
				mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET slug = 'all', subsystem = 'im' WHERE scope_id = $1`, channelID)
			},
			wantError: ErrValidation,
		},
		{
			name: "active-team",
			setup: func(t *testing.T, db coredb.DB, channelID string) {
				mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'im' WHERE scope_id = $1`, channelID)
				mustExec(t, ctx, db, `INSERT INTO workspace_teams (id, workspace_id, handle, display_name, description, status, channel_id, created_by_member_id, created_at, updated_at) VALUES ('team_test', 'wsp_channel_mgmt', 'test', 'test', '', 'active', $1, 'mem_owner', NOW(), NOW())`, channelID)
			},
			wantError: ErrNotFound,
		},
		{name: "kb", setup: func(t *testing.T, db coredb.DB, ch string) {
			mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'kb' WHERE scope_id = $1`, ch)
		}, wantError: ErrNotFound},
		{name: "project", setup: func(t *testing.T, db coredb.DB, ch string) {
			mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'project' WHERE scope_id = $1`, ch)
		}, wantError: ErrNotFound},
		{name: "tutorial", setup: func(t *testing.T, db coredb.DB, ch string) {
			mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'tutorial' WHERE scope_id = $1`, ch)
		}, wantError: ErrNotFound},
	}

	for _, tc := range cases {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			db := newPooledTestDB(t, ctx)
			service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
			agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
			channelID := "chn_" + tc.name
			seedOwnedChannel(t, ctx, db, channelID, "mem_owner", "usr_owner")
			tc.setup(t, db, channelID)
			_, err := service.Update(ctx, agent, UpdateInput{RequestID: "req-" + tc.name, ChannelID: channelID, ExpectedVersion: 1, Description: strPtr("x")})
			require.ErrorIs(t, err, tc.wantError, "carrier %s must be rejected", tc.name)
		})
	}
}

// TestBrowserUpdateAllChannelRenameIsNoOp is the I1 regression: under the browser
// policy, renaming #all to its own slug (or current name) is a no-op that keeps
// the original display name rather than renaming.
func TestBrowserUpdateAllChannelRenameIsNoOp(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	_ = agent
	channelID := "chn_all_browser"
	seedOwnedChannel(t, ctx, db, channelID, "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET slug = 'all', display_name = 'All', subsystem = 'im' WHERE scope_id = $1`, channelID)

	appStore := app.New(db, app.Options{})
	slug := strPtr("all") // input == slug
	var summary message.ChannelSummary
	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		updated, _, updateErr := app.NewChannelManagementPort(appStore).UpdateChannelProfileInTx(ctx, tx, "wsp_channel_mgmt", channelID, "usr_owner", app.BrowserChannelUpdate, nil, slug, nil, nil)
		summary = updated
		return updateErr
	})
	require.NoError(t, err)
	require.Equal(t, "All", summary.DisplayName, "#all rename to its own slug must keep the original display name (no-op)")
}

// TestGetRejectsUnjoinedPrivateChannel validates the C1 unauthorized-get gate: an
// owner cannot read a private IM channel they have not joined.
func TestGetRejectsUnjoinedPrivateChannel(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_private_other", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, status, created_at, updated_at) VALUES ('mem_other', 'wsp_channel_mgmt', 'human', 'other', 'usr_other', 'member', 'active', NOW(), NOW())`)
	// Re-point the channel to a different owner and remove mem_owner's membership.
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET owner_member_id = 'mem_other', visibility = 'private' WHERE scope_id = $1`, "chn_private_other")
	mustExec(t, ctx, db, `DELETE FROM scope_memberships WHERE scope_id = $1 AND member_id = $2`, "chn_private_other", "mem_owner")

	_, err := service.Get(ctx, agent, "chn_private_other")
	require.ErrorIs(t, err, ErrNotFound, "owner must not read a private channel they have not joined")
	_, err = service.Members(ctx, agent, "chn_private_other", PageInput{})
	require.ErrorIs(t, err, ErrNotFound, "owner must not list members of a private channel they have not joined")
}

// TestCreateWithRealStorePersistsDescriptionAndVersion is the concrete
// channel.Store end-to-end create test (no fake mutation): it verifies
// description is persisted, profile version=1, the returned view carries
// version=1, and sequential replay returns the same channel with only one
// scope/profile row.
func TestCreateWithRealStorePersistsDescriptionAndVersion(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	seedWorkspaceAndPrincipal(t, ctx, db)
	realStore := channel.New(db, channel.Options{
		RequireWorkspaceAccess: func(_ context.Context, _, _ string) (string, error) { return "mem_owner", nil },
		RequireChannelReadAccess: func(_ context.Context, _, _, _ string) (channel.HumanChannelAccess, error) {
			return channel.HumanChannelAccess{}, nil
		},
		EnsureChannelScopeParticipantProjection: func(ctx context.Context, execer coredb.QueryExecer, channelID, memberID, deliveryPolicy string, createdAt time.Time) error {
			return message.EnsureScopeMembership(ctx, execer, channelID, memberID, "public", "", deliveryPolicy, 0, 0, createdAt)
		},
		MarkScopeParticipantProjectionRemoved: func(_ context.Context, _ coredb.QueryExecer, _, _ string, _ time.Time) error { return nil },
		CheckWorkspaceQuotaInTx:               func(_ context.Context, _ coredb.Tx, _ string, _ plan.MetricKey, _ int64) error { return nil },
	})
	service := New(Dependencies{
		DB: db, Channels: realStore, Apps: app.NewChannelManagementPort(app.New(db, app.Options{})),
		Owners:     fakeOwnerResolver{userID: "usr_owner", memberID: "mem_owner"},
		Workspaces: fakeWorkspaceResolver{workspaceID: "wsp_channel_mgmt", memberID: "mem_owner", userID: "usr_owner"},
		Realtime:   &recordingRealtime{},
	})
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	desc := "release coordination room"
	first, err := service.Create(ctx, agent, CreateInput{RequestID: "req-real-1", DisplayName: "release-room", Description: &desc})
	require.NoError(t, err)
	require.Equal(t, int64(1), first.Channel.Version, "created channel must return version=1")
	require.Equal(t, desc, first.Channel.Description, "created channel must carry the description")

	var dbDescription string
	var dbVersion int64
	require.NoError(t, db.QueryRow(ctx, `SELECT description, version FROM scope_channel_profiles WHERE scope_id = $1`, first.Channel.ChannelID).Scan(&dbDescription, &dbVersion))
	require.Equal(t, desc, dbDescription, "description must be persisted")
	require.Equal(t, int64(1), dbVersion, "profile version must be 1 after create")

	// Sequential replay: same channel, same result, only one row.
	replayed, err := service.Create(ctx, agent, CreateInput{RequestID: "req-real-1", DisplayName: "release-room", Description: &desc})
	require.NoError(t, err)
	require.Equal(t, first.Channel.ChannelID, replayed.Channel.ChannelID)

	var scopeCount int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM scopes WHERE workspace_id = $1 AND type = 'channel'`, "wsp_channel_mgmt").Scan(&scopeCount))
	require.Equal(t, 1, scopeCount, "sequential replay must not create a second channel")
}

// TestReadMatrixFiltersCarriersAndAllowsPlainIM is the read DB matrix: it
// directly guards the List carrier filter (Tutorial || TeamID != "" || Subsystem
// != "im") + Get/Members carrier rejection, using real channel.Store + app.Store
// (no fake mutation). Plain IM and #all are listed/read; team, tutorial, KB and
// project are excluded from List and rejected by Get/Members.
func TestReadMatrixFiltersCarriersAndAllowsPlainIM(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	seedWorkspaceAndPrincipal(t, ctx, db)
	realStore := channel.New(db, channel.Options{
		RequireWorkspaceAccess: func(_ context.Context, _, _ string) (string, error) { return "mem_owner", nil },
		RequireChannelReadAccess: func(_ context.Context, _, _, _ string) (channel.HumanChannelAccess, error) {
			return channel.HumanChannelAccess{}, nil
		},
		EnsureChannelScopeParticipantProjection: func(ctx context.Context, execer coredb.QueryExecer, channelID, memberID, deliveryPolicy string, createdAt time.Time) error {
			return message.EnsureScopeMembership(ctx, execer, channelID, memberID, "public", "", deliveryPolicy, 0, 0, createdAt)
		},
		MarkScopeParticipantProjectionRemoved: func(_ context.Context, _ coredb.QueryExecer, _, _ string, _ time.Time) error { return nil },
		CheckWorkspaceQuotaInTx:               func(_ context.Context, _ coredb.Tx, _ string, _ plan.MetricKey, _ int64) error { return nil },
	})
	service := New(Dependencies{
		DB: db, Channels: realStore, Apps: app.NewChannelManagementPort(app.New(db, app.Options{})),
		Owners:     fakeOwnerResolver{userID: "usr_owner", memberID: "mem_owner"},
		Workspaces: fakeWorkspaceResolver{workspaceID: "wsp_channel_mgmt", memberID: "mem_owner", userID: "usr_owner"},
		Realtime:   &recordingRealtime{},
	})
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}

	// Seed five channels the owner has joined, each with a different carrier.
	seedOwnedChannel(t, ctx, db, "chn_im", "mem_owner", "usr_owner")
	seedOwnedChannel(t, ctx, db, "chn_all", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET slug = 'all', subsystem = 'im' WHERE scope_id = $1`, "chn_all")
	seedOwnedChannel(t, ctx, db, "chn_team", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'team' WHERE scope_id = $1`, "chn_team")
	mustExec(t, ctx, db, `INSERT INTO workspace_teams (id, workspace_id, handle, display_name, description, status, channel_id, created_by_member_id, created_at, updated_at) VALUES ('team_rmx', 'wsp_channel_mgmt', 'rmx', 'Team RMX', '', 'active', 'chn_team', 'mem_owner', NOW(), NOW())`)
	seedOwnedChannel(t, ctx, db, "chn_tutorial", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'tutorial' WHERE scope_id = $1`, "chn_tutorial")
	seedOwnedChannel(t, ctx, db, "chn_kb", "mem_owner", "usr_owner")
	mustExec(t, ctx, db, `UPDATE scope_channel_profiles SET subsystem = 'kb' WHERE scope_id = $1`, "chn_kb")

	// List must include only plain IM + #all; team/tutorial/KB excluded.
	list, err := service.List(ctx, agent, PageInput{})
	require.NoError(t, err)
	ids := make([]string, 0, len(list.Channels))
	for _, ch := range list.Channels {
		ids = append(ids, ch.ChannelID)
	}
	require.ElementsMatch(t, []string{"chn_im", "chn_all"}, ids, "List must include only plain IM and #all")

	// Plain IM: Get returns version; Members succeeds and contains the owner.
	imGet, err := service.Get(ctx, agent, "chn_im")
	require.NoError(t, err)
	require.Equal(t, int64(1), imGet.Channel.Version)
	imMembers, err := service.Members(ctx, agent, "chn_im", PageInput{})
	require.NoError(t, err)
	require.NotEmpty(t, imMembers.Members, "Members must include the owner")

	// #all: Get succeeds (allowAll=true contract).
	_, err = service.Get(ctx, agent, "chn_all")
	require.NoError(t, err, "#all read must succeed under allowAll=true")

	// Team/tutorial/KB: Get and Members must reject (NotFound).
	for _, ch := range []string{"chn_team", "chn_tutorial", "chn_kb"} {
		_, gerr := service.Get(ctx, agent, ch)
		require.ErrorIs(t, gerr, ErrNotFound, "Get must reject carrier %s", ch)
		_, merr := service.Members(ctx, agent, ch, PageInput{})
		require.ErrorIs(t, merr, ErrNotFound, "Members must reject carrier %s", ch)
	}
}

func TestMembersPaginationContinuesPastOneHundred(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	const channelID = "chn_members_page"
	seedOwnedChannel(t, ctx, db, channelID, "mem_owner", "usr_owner")

	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		if _, err := tx.Exec(ctx, `SET FOREIGN_KEY_CHECKS = 0`); err != nil {
			return err
		}
		for index := 0; index < 101; index++ {
			memberID := fmt.Sprintf("mem_page_%03d", index)
			userID := fmt.Sprintf("usr_page_%03d", index)
			if _, err := tx.Exec(ctx, `
INSERT INTO members (
  id, workspace_id, type, display_name, user_id, workspace_role,
  status, created_at, updated_at
) VALUES ($1, 'wsp_channel_mgmt', 'human', $1, $2, 'member', 'active', NOW(), NOW())
`, memberID, userID); err != nil {
				return err
			}
			if _, err := tx.Exec(ctx, `
INSERT INTO scope_memberships (
  scope_id, member_id, visibility, owner_member_id,
  last_read_seq, last_read_global_seq, last_activity_global_seq,
  owner_context_dismissed_through_global_seq, effective_from_seq,
  created_at, updated_at
) VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NOW(), NOW())
`, channelID, memberID); err != nil {
				return err
			}
		}
		return nil
	}))

	first, err := service.Members(ctx, agent, channelID, PageInput{Limit: 100})
	require.NoError(t, err)
	require.Len(t, first.Members, 100)
	require.True(t, first.HasMore)
	require.Equal(t, "100", first.NextCursor)
	require.Equal(t, 102, first.TotalCount)

	second, err := service.Members(ctx, agent, channelID, PageInput{Limit: 100, Cursor: first.NextCursor})
	require.NoError(t, err)
	require.Len(t, second.Members, 2)
	require.False(t, second.HasMore)
	require.Empty(t, second.NextCursor)
	require.Equal(t, first.TotalCount, second.TotalCount)

	seen := make(map[string]struct{}, first.TotalCount)
	for _, member := range append(append([]MemberView(nil), first.Members...), second.Members...) {
		seen[member.MemberID] = struct{}{}
	}
	require.Len(t, seen, first.TotalCount, "continuation must expose every member exactly once")
}

// TestAgentAddPersistsAgentAuthor verifies the system message written by
// AgentAdd carries the Agent member as author (author_member_id=mem_agent,
// author_type=agent), proving the actor/authority separation. Human members are
// not addable by an Agent, so the target is an Agent (here the acting Agent
// itself, which is a legitimate explicit target).
func TestAgentAddPersistsAgentAuthor(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_member_add", "mem_owner", "usr_owner")

	_, err := service.AgentAdd(ctx, agent, "chn_member_add", "req-member-add-1", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)

	var authorMemberID, authorType string
	require.NoError(t, db.QueryRow(ctx, `SELECT author_member_id, author_type FROM scope_messages WHERE scope_id = $1 ORDER BY seq DESC LIMIT 1`, "chn_member_add").Scan(&authorMemberID, &authorType))
	require.Equal(t, "mem_agent", authorMemberID, "system message author must be the Agent member")
	require.Equal(t, "agent", authorType, "system message author_type must be agent")
}

// TestAgentAddSequentialReplay verifies that replaying the same requestId
// returns the same result without re-executing the mutation or emitting realtime.
func TestAgentAddSequentialReplay(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, realtime, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_member_replay", "mem_owner", "usr_owner")

	first, err := service.AgentAdd(ctx, agent, "chn_member_replay", "req-member-replay", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)
	require.Len(t, first.Members, 1, "first response must preserve the complete DirectMembers projection")
	require.Equal(t, "mem_agent", first.Members[0].MemberID)
	require.Equal(t, "agent", first.Members[0].Kind)
	require.NotEmpty(t, first.Members[0].DisplayName)

	realtime.mu.Lock()
	firstEmit := realtime.membersUpdated
	realtime.mu.Unlock()
	require.Equal(t, 1, firstEmit, "initial mutation emits once")

	replayed, err := service.AgentAdd(ctx, agent, "chn_member_replay", "req-member-replay", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)
	require.Equal(t, first, replayed, "receipt replay must preserve memberId/kind/displayName exactly")
	realtime.mu.Lock()
	afterReplay := realtime.membersUpdated
	realtime.mu.Unlock()
	require.Equal(t, 1, afterReplay, "replay must not emit realtime")
}

func TestAgentMutationRejectsOneHundredAndOneBeforeReceipt(t *testing.T) {
	ctx := context.Background()
	service, _, _, db := newChannelManagementTestService(t, ctx, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	agents := make([]AgentReference, 0, 101)
	for index := 0; index < 101; index++ {
		agents = append(agents, AgentReference{AgentID: fmt.Sprintf("agt_batch_%03d", index)})
	}

	_, err := service.AgentAdd(ctx, agent, "chn_batch_limit", "req-batch-limit", agents)
	require.ErrorIs(t, err, ErrValidation)

	var receiptCount int
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM channel_operation_receipts
WHERE workspace_id = $1 AND applicant_agent_id = $2 AND request_id = $3
`, agent.WorkspaceID, agent.AgentID, "req-batch-limit").Scan(&receiptCount))
	require.Zero(t, receiptCount, "oversized batch must be rejected before reserving an idempotency receipt")
}

// TestAgentMutationRejectsEmptyAgentReference verifies that a reference with an
// empty agentId is rejected.
func TestAgentMutationRejectsEmptyAgentReference(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_member_service", "mem_owner", "usr_owner")

	_, err := service.AgentAdd(ctx, agent, "chn_member_service", "req-service-reject", []AgentReference{{}})
	require.ErrorIs(t, err, ErrValidation, "empty agent reference must be rejected")
}

// TestAgentRemovePersistsAgentAuthor verifies that the remove system message
// also carries the Agent member as author.
func TestAgentRemovePersistsAgentAuthor(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, _, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_member_rm", "mem_owner", "usr_owner")
	// First add the agent so they are a member.
	_, err := service.AgentAdd(ctx, agent, "chn_member_rm", "req-rm-add", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)
	// Then remove them.
	_, err = service.AgentRemove(ctx, agent, "chn_member_rm", "req-rm-remove", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)

	var authorMemberID, authorType string
	require.NoError(t, db.QueryRow(ctx, `SELECT author_member_id, author_type FROM scope_messages WHERE scope_id = $1 AND JSON_UNQUOTE(JSON_EXTRACT(metadata, '$.channelMemberEvent.action')) = 'removed' ORDER BY seq DESC LIMIT 1`, "chn_member_rm").Scan(&authorMemberID, &authorType))
	require.Equal(t, "mem_agent", authorMemberID, "remove system message author must be the Agent member")
	require.Equal(t, "agent", authorType, "remove system message author_type must be agent")
}

// TestAgentAddNoOpDoesNotEmit verifies that adding an already-member agent is a
// no-op: no realtime emit, no system message written.
func TestAgentAddNoOpDoesNotEmit(t *testing.T) {
	ctx := context.Background()
	db := newPooledTestDB(t, ctx)
	service, _, realtime, _ := buildChannelManagementService(t, ctx, db, "usr_owner", "mem_owner")
	agent := agent.AuthenticatedAgent{AgentID: "agt_owner", WorkspaceID: "wsp_channel_mgmt", MemberID: "mem_agent", MachineID: "mac_owner"}
	seedOwnedChannel(t, ctx, db, "chn_noop", "mem_owner", "usr_owner")
	// First add (changed).
	_, err := service.AgentAdd(ctx, agent, "chn_noop", "req-noop-1", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)
	realtime.mu.Lock()
	firstEmit := realtime.membersUpdated
	realtime.mu.Unlock()
	require.Equal(t, 1, firstEmit, "first add must emit once")
	// Second add of the same agent (no-op: already added).
	result2, err := service.AgentAdd(ctx, agent, "chn_noop", "req-noop-2", []AgentReference{{AgentID: "agt_owner"}})
	require.NoError(t, err)
	require.Empty(t, result2.Members, "no-op add must return empty members")
	realtime.mu.Lock()
	afterNoOp := realtime.membersUpdated
	realtime.mu.Unlock()
	require.Equal(t, 1, afterNoOp, "no-op add must not emit realtime")
}

// seedWorkspaceAndPrincipal seeds the workspace + human owner member + machine +
// agent + agent member for real-store tests that exercise resolveOwnerInTx.
func seedWorkspaceAndPrincipal(t *testing.T, ctx context.Context, db coredb.DB) {
	t.Helper()
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		if _, err := tx.Exec(ctx, `SET FOREIGN_KEY_CHECKS = 0`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO workspaces (id, organization_id, slug, name, created_at, updated_at) VALUES ('wsp_channel_mgmt', 'org_test', 'wsp-channel-mgmt', 'test', NOW(), NOW())`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, status, created_at, updated_at) VALUES ('mem_owner', 'wsp_channel_mgmt', 'human', 'owner', 'usr_owner', 'owner', 'active', NOW(), NOW())`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO members (id, workspace_id, type, display_name, agent_display_name_key, status, created_at, updated_at) VALUES ('mem_agent', 'wsp_channel_mgmt', 'agent', 'agent', 'agent', 'active', NOW(), NOW())`); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO machines (id, workspace_id, created_by_user_id, name, api_key, status, created_at, updated_at) VALUES ('mac_owner', 'wsp_channel_mgmt', 'usr_owner', 'mac', 'key', 'active', NOW(), NOW())`); err != nil {
			return err
		}
		_, err := tx.Exec(ctx, `INSERT INTO agents (id, workspace_id, machine_id, member_id, name, runtime, model, status, description, activity_detail, created_at, updated_at) VALUES ('agt_owner', 'wsp_channel_mgmt', 'mac_owner', 'mem_agent', 'agent', 'local', 'm', 'active', '', '', NOW(), NOW())`)
		return err
	}))
}