package app

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/service/appruntime"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/app/registry"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/serviceactor"
)

// afterOuterAppManagerReadDB replaces the active manager after the selected
// non-locking authorization read but before the final mutation transaction. It
// also wraps that transaction so the test can prove the manager assignment was
// re-read with FOR UPDATE.
type afterOuterAppManagerReadDB struct {
	coredb.DB
	once             sync.Once
	mu               sync.Mutex
	outerReads       int
	triggerAfterRead int
	beganTx          bool
	sawLockedRead    bool
	afterRead        func()
}

func (db *afterOuterAppManagerReadDB) QueryRow(ctx context.Context, query string, args ...any) coredb.Row {
	row := db.DB.QueryRow(ctx, query, args...)
	if !strings.Contains(query, "FROM extension_definition_managers") || strings.Contains(query, "FOR UPDATE") {
		return row
	}
	return afterOuterMembershipReadRow{Row: row, afterRead: func() {
		db.mu.Lock()
		db.outerReads++
		trigger := db.outerReads == db.triggerAfterRead
		db.mu.Unlock()
		if trigger {
			db.once.Do(db.afterRead)
		}
	}}
}

func (db *afterOuterAppManagerReadDB) BeginTx(ctx context.Context, options coredb.TxOptions) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, options)
	if err != nil {
		return nil, err
	}
	db.mu.Lock()
	db.beganTx = true
	db.mu.Unlock()
	return &appManagerTrackingTx{Tx: tx, db: db}, nil
}

func (db *afterOuterAppManagerReadDB) snapshot() (outerReads int, beganTx bool, sawLockedRead bool) {
	db.mu.Lock()
	defer db.mu.Unlock()
	return db.outerReads, db.beganTx, db.sawLockedRead
}

type appManagerTrackingTx struct {
	coredb.Tx
	db *afterOuterAppManagerReadDB
}

func (tx *appManagerTrackingTx) QueryRow(ctx context.Context, query string, args ...any) coredb.Row {
	if strings.Contains(query, "FROM extension_definition_managers") && strings.Contains(query, "FOR UPDATE") {
		tx.db.mu.Lock()
		tx.db.sawLockedRead = true
		tx.db.mu.Unlock()
	}
	return tx.Tx.QueryRow(ctx, query, args...)
}

func TestAppRunnerCredentialHeartbeatClaimAndLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runner_runtime", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "runner_runtime", now)
	channelID := "chn_bot_runner_runtime"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-runner-runtime", "App runner runtime", "private", fixture.MemberID, fixture.MemberID, now)

	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)

	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{
		Name:     "staging runner",
		Metadata: map[string]any{"environment": "staging"},
	})
	require.NoError(t, err)
	require.NotEmpty(t, created.Token)
	require.Equal(t, serviceactor.RuntimeCredentialKindRunnerStreamToken, created.Credential.CredentialKind)
	require.Equal(t, serviceactor.RuntimeTransportKindRunnerStream, created.Credential.TransportKind)
	require.Equal(t, serviceactor.RuntimeCredentialStatusActive, created.Credential.Status)
	require.Equal(t, installation.ID, created.Credential.ServiceInstallationID)
	require.NotEmpty(t, created.Credential.ServiceMemberID)
	require.Equal(t, "staging", created.Credential.Metadata["environment"])
	require.Equal(t, serviceactor.RuntimeTransportKindRunnerStream, created.Credential.Metadata["transport"])

	listedCredentials, err := appStore.ListAppRunnerCredentialsForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID)
	require.NoError(t, err)
	require.Len(t, listedCredentials, 1)
	require.Equal(t, created.Credential.ID, listedCredentials[0].ID)
	require.Equal(t, "staging runner", listedCredentials[0].Name)
	require.Equal(t, serviceactor.RuntimeCredentialStatusActive, listedCredentials[0].Status)

	auth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	require.Equal(t, fixture.WorkspaceID, auth.WorkspaceID)
	require.Equal(t, installation.ID, auth.ServiceInstallationID)
	require.Equal(t, created.Credential.ID, auth.ServiceRuntimeCredentialID)

	session, err := newRunnerServiceForTest(appStore).Heartbeat(ctx, auth, appruntime.RunnerHeartbeatParams{
		RunnerID:      "runner-staging-1",
		RunnerVersion: "1.0.0",
		Capabilities:  map[string]any{"maxConcurrency": float64(2)},
		Metadata:      map[string]any{"zone": "staging-a"},
	})
	require.NoError(t, err)
	require.Equal(t, serviceactor.RuntimeSessionStatusOnline, session.Status)
	require.Equal(t, "runner-staging-1", session.RunnerID)
	require.Equal(t, "1.0.0", session.RunnerVersion)
	require.NotNil(t, session.LastHeartbeatAt)
	require.Equal(t, float64(2), session.Capabilities["maxConcurrency"])
	require.Equal(t, "staging-a", session.Metadata["zone"])
	require.Equal(t, serviceactor.RuntimeTransportKindRunnerStream, session.Metadata["transport"])

	content := "@deploy-app-target-runner-runtime upgrade --version 1.2.3"
	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, content, nil)
	require.NoError(t, err)
	record := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, sourceMessage.ID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, record.SubscriptionKind)
	require.Equal(t, sourceMessage.ID, record.EventID)

	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{
		RunnerID:      "runner-staging-1",
		RunnerVersion: "1.0.1",
		Limit:         1,
		LeaseDuration: time.Minute,
	})
	require.NoError(t, err)
	require.Equal(t, session.ID, claimed.Session.ID)
	require.NotNil(t, claimed.Session.LastClaimAt)
	require.Len(t, claimed.Envelopes, 1)
	require.Equal(t, sourceMessage.ID, claimed.Envelopes[0].Delivery.EventID)
	require.Equal(t, serviceactor.RuntimeTransportKindRunnerStream, claimed.Envelopes[0].Attempt.TransportKind)
	require.Equal(t, created.Credential.ID, claimed.Envelopes[0].Attempt.ServiceRuntimeCredentialID)
	require.Equal(t, session.ID, claimed.Envelopes[0].Attempt.ServiceRuntimeSessionID)
	require.NotEmpty(t, claimed.Envelopes[0].DeliveryToken)

	offline, err := newRunnerServiceForTest(appStore).MarkSessionOffline(ctx, auth, session.ID)
	require.NoError(t, err)
	require.Equal(t, serviceactor.RuntimeSessionStatusOffline, offline.Status)
	require.Equal(t, session.ID, offline.ID)

	disabled, err := appStore.DisableAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, created.Credential.ID)
	require.NoError(t, err)
	require.Equal(t, serviceactor.RuntimeCredentialStatusDisabled, disabled.Status)
	disabledSession, err := appStore.serviceActorStore().LoadRuntimeSession(ctx, fixture.WorkspaceID, session.ID)
	require.NoError(t, err)
	require.Equal(t, serviceactor.RuntimeSessionStatusDisabled, disabledSession.Status)
	listedCredentials, err = appStore.ListAppRunnerCredentialsForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID)
	require.NoError(t, err)
	require.Empty(t, listedCredentials)
	_, err = newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.ErrorIs(t, err, ErrUnauthorized)
	_, err = newRunnerServiceForTest(appStore).Heartbeat(ctx, auth, appruntime.RunnerHeartbeatParams{RunnerID: "runner-staging-1"})
	require.ErrorIs(t, err, ErrUnauthorized)
}

func TestCreateAppRunnerCredentialRevalidatesCurrentAdminInMutationTransaction(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_runner_create_admin_demoted", now)
	creator := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_runner_create_creator", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "runner_create_admin_demoted", now)
	mustExec(t, pool, `
UPDATE extension_definition_managers
SET member_id = $3,
    granted_by_member_id = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND service_definition_id = $2
  AND status = 'active'
  AND ended_at IS NULL
`, admin.WorkspaceID, installation.AppID, creator.MemberID, now.Add(time.Second))

	guardedStore := New(&afterOuterMembershipReadDB{
		DB: pool,
		afterRead: func() {
			mustExec(t, pool, `UPDATE members SET workspace_role = 'member', updated_at = $2 WHERE id = $1`, admin.MemberID, time.Now().UTC())
		},
	}, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})

	_, err := guardedStore.CreateAppRunnerCredentialForUser(ctx, admin.UserID, admin.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{Name: "must not be created"})
	require.ErrorIs(t, err, ErrNotFound)
	var credentialCount int64
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_runtime_credentials
WHERE workspace_id = $1
  AND service_installation_id = $2
`, admin.WorkspaceID, installation.ID).Scan(&credentialCount))
	require.Zero(t, credentialCount)
}

func TestCreateAppRunnerCredentialRevalidatesManagerAssignmentInMutationTransaction(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_runner_creator_changed", now)
	creator := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_runner_creator_changed_creator", now)
	peer := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_runner_creator_changed_peer", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "runner_creator_changed", now)
	mustExec(t, pool, `
UPDATE extension_definition_managers
SET member_id = $3,
    granted_by_member_id = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND service_definition_id = $2
  AND status = 'active'
  AND ended_at IS NULL
`, admin.WorkspaceID, installation.AppID, creator.MemberID, now.Add(time.Second))

	guardedDB := &afterOuterAppManagerReadDB{
		DB:               pool,
		triggerAfterRead: 1,
		afterRead: func() {
			mustExec(t, pool, `
UPDATE extension_definition_managers
SET member_id = $3,
    granted_by_member_id = $3,
    version = version + 1,
    updated_at = $4
WHERE workspace_id = $1
  AND service_definition_id = $2
  AND status = 'active'
  AND ended_at IS NULL
`, admin.WorkspaceID, installation.AppID, peer.MemberID, time.Now().UTC())
		},
	}
	guardedStore := New(guardedDB, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})

	_, err := guardedStore.CreateAppRunnerCredentialForUser(ctx, creator.UserID, admin.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{Name: "must roll back"})
	require.ErrorIs(t, err, ErrNotFound)
	outerReads, beganTx, sawLockedRead := guardedDB.snapshot()
	require.Equal(t, 1, outerReads, "manager must change only after the consolidated outer authorization read")
	require.True(t, beganTx, "the mutation must reach its transaction before rejecting the replaced manager")
	require.True(t, sawLockedRead, "transactional revalidation must lock the authoritative manager assignment")
	var credentialCount int64
	require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM service_runtime_credentials WHERE workspace_id = $1 AND service_installation_id = $2`, admin.WorkspaceID, installation.ID).Scan(&credentialCount))
	require.Zero(t, credentialCount)
}

func TestDisableAppRunnerCredentialRevalidatesCurrentAdminInMutationTransaction(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_runner_disable_admin_removed", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "runner_disable_admin_removed", now)
	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, admin.UserID, admin.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{Name: "stay active"})
	require.NoError(t, err)

	guardedStore := New(&afterOuterMembershipReadDB{
		DB: pool,
		afterRead: func() {
			mustExec(t, pool, `UPDATE members SET status = 'inactive', updated_at = $2 WHERE id = $1`, admin.MemberID, time.Now().UTC())
		},
	}, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})

	_, err = guardedStore.DisableAppRunnerCredentialForUser(ctx, admin.UserID, admin.WorkspaceID, created.Credential.ID)
	require.ErrorIs(t, err, ErrForbidden)
	persisted, err := appStore.serviceActorStore().LoadRuntimeCredential(ctx, admin.WorkspaceID, created.Credential.ID)
	require.NoError(t, err)
	require.Equal(t, serviceactor.RuntimeCredentialStatusActive, persisted.Status)
	require.Nil(t, persisted.DisabledAt)
}

func TestRotateAppCallbackSecretRevalidatesCurrentAdminInMutationTransaction(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	admin := seedAppAdminFixture(t, pool, "bot_callback_secret_admin_demoted", now)
	creator := seedWorkspaceMemberFixture(t, pool, admin.WorkspaceID, "bot_callback_secret_creator", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, admin, "callback_secret_admin_demoted", now)
	mustExec(t, pool, `
UPDATE extension_definition_managers
SET member_id = $3,
    granted_by_member_id = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND service_definition_id = $2
  AND status = 'active'
  AND ended_at IS NULL
`, admin.WorkspaceID, installation.AppID, creator.MemberID, now.Add(time.Second))
	var originalSecret string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COALESCE(JSON_UNQUOTE(JSON_EXTRACT(COALESCE(app.config, JSON_OBJECT()), '$.callbackSigningSecret')), '')
FROM service_definitions app
JOIN service_installations installation ON installation.service_definition_id = app.id
WHERE installation.id = $1
`, installation.ID).Scan(&originalSecret))
	rotated, err := appStore.RotateAppCallbackSecretForUser(ctx, admin.UserID, admin.WorkspaceID, installation.ID)
	require.NoError(t, err)
	require.NotEmpty(t, rotated.CallbackSigningSecret)
	require.NotEqual(t, originalSecret, rotated.CallbackSigningSecret)
	originalSecret = rotated.CallbackSigningSecret

	guardedStore := New(&afterOuterMembershipReadDB{
		DB: pool,
		afterRead: func() {
			mustExec(t, pool, `UPDATE members SET workspace_role = 'member', updated_at = $2 WHERE id = $1`, admin.MemberID, time.Now().UTC())
		},
	}, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})

	_, err = guardedStore.RotateAppCallbackSecretForUser(ctx, admin.UserID, admin.WorkspaceID, installation.ID)
	require.ErrorIs(t, err, ErrNotFound)
	var persistedSecret string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COALESCE(JSON_UNQUOTE(JSON_EXTRACT(COALESCE(app.config, JSON_OBJECT()), '$.callbackSigningSecret')), '')
FROM service_definitions app
JOIN service_installations installation ON installation.service_definition_id = app.id
WHERE installation.id = $1
`, installation.ID).Scan(&persistedSecret))
	require.Equal(t, originalSecret, persistedSecret)
}

func TestAppRuntimeReplyAckAndFailLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	fanout := &recordingCommittedMessageFanout{query: func(callbackCtx context.Context, delivery message.MessageDelivery) {
		var effectCount int
		require.NoError(t, pool.QueryRow(callbackCtx, `
SELECT COUNT(*)
FROM service_event_delivery_effects
WHERE message_id = $1
`, delivery.MessageID).Scan(&effectCount))
		require.Equal(t, 1, effectCount, "App reply fanout must run after its effect transaction commits")
	}}
	appStore.SetCommittedMessageFanout(fanout)
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runtime_reply_ack_fail", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "runtime_reply", now)
	channelID := "chn_bot_runtime_reply_ack_fail"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-runtime-reply", "App runtime reply", "private", fixture.MemberID, fixture.MemberID, now)
	const (
		runtimeObserverAgentID     = "agt_bot_runtime_reply_observer"
		runtimeObserverMemberID    = "mem_bot_runtime_reply_observer"
		runtimeObserverAgentHandle = "runtime-observer"
	)
	seedAppAgentParticipant(
		t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, channelID,
		"mach_bot_runtime_reply_observer", runtimeObserverAgentID, runtimeObserverMemberID,
		runtimeObserverAgentHandle, now,
	)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	auth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)

	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "@deploy-app-target-runtime-reply upgrade --version 1.2.3", nil)
	require.NoError(t, err)
	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-runtime", RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	require.Equal(t, sourceMessage.ID, envelope.Delivery.EventID)

	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	replyBody := "升级开始，目标版本 1.2.3 @" + runtimeObserverAgentHandle
	reply, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        replyBody,
		IdempotencyKey: "progress-1",
		Metadata:       map[string]any{"phase": "start"},
	})
	require.NoError(t, err)
	require.Equal(t, serviceactor.DeliveryAttemptStatusDelivered, reply.Attempt.Status)
	require.Equal(t, "message", reply.Effect.EffectKind)
	require.NotEmpty(t, reply.MessageID)
	require.Len(t, fanout.deliveries, 1)
	committed := fanout.deliveries[0]
	require.Equal(t, reply.MessageID, committed.MessageID)
	require.NotNil(t, committed.AppMessage)
	require.Equal(t, channelID, committed.AppMessage.ChannelID)
	require.True(t, committed.RecoverableStreamProjection)
	require.Contains(t, agentDeliveryReasonsByMember(committed.Recipients), runtimeObserverMemberID)
	require.Empty(t, committed.WakeRecipients, "silent App replies must not wake Agents")
	durable := requireScopeFanoutPayloadForMessage(t, pool, reply.MessageID)
	require.True(t, durable.Delivery.RecoverableStreamProjection)
	require.Contains(t, agentDeliveryReasonsByMember(durable.Delivery.Recipients), runtimeObserverMemberID)
	require.Empty(t, durable.Delivery.WakeRecipients)

	replayed, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        replyBody,
		IdempotencyKey: "progress-1",
		Metadata:       map[string]any{"phase": "start"},
	})
	require.NoError(t, err)
	require.Equal(t, reply.Effect.ID, replayed.Effect.ID)
	require.Equal(t, reply.MessageID, replayed.MessageID)
	require.Len(t, fanout.deliveries, 1, "idempotent App reply replay must not fan out again")
	requireScopeMessageAndSingleFanout(t, pool, reply.MessageID, fixture.WorkspaceID, channelID)

	_, err = appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "升级开始，目标版本 1.2.4",
		IdempotencyKey: "progress-1",
		Metadata:       map[string]any{"phase": "start"},
	})
	require.ErrorIs(t, err, ErrConflict)

	var replyContent, replyAuthorType, replyAuthorMember string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT content, author_type, author_member_id
FROM scope_messages
WHERE id = $1
`, reply.MessageID).Scan(&replyContent, &replyAuthorType, &replyAuthorMember))
	require.Equal(t, replyBody, replyContent)
	require.Equal(t, "service", replyAuthorType)
	require.Equal(t, installation.MemberID, replyAuthorMember)

	var effectCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_delivery_effects
WHERE service_event_delivery_id = $1
`, envelope.Delivery.ID).Scan(&effectCount))
	require.Equal(t, 1, effectCount)

	runnerDeliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, created.Token, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	acked, err := appStore.AckAppRuntimeDelivery(ctx, runnerDeliveryAuth, AppRuntimeAckParams{AttemptID: envelope.Attempt.ID, Result: map[string]any{"version": "1.2.3"}})
	require.NoError(t, err)
	require.Equal(t, "succeeded", acked.Delivery.Status)
	require.Equal(t, serviceactor.DeliveryAttemptStatusAcked, acked.Attempt.Status)
	require.Equal(t, reply.MessageID, acked.Delivery.MessageID)

	ackedAgain, err := appStore.AckAppRuntimeDelivery(ctx, runnerDeliveryAuth, AppRuntimeAckParams{AttemptID: envelope.Attempt.ID, Result: map[string]any{"version": "1.2.3"}})
	require.NoError(t, err)
	require.Equal(t, "succeeded", ackedAgain.Delivery.Status)
	require.Equal(t, serviceactor.DeliveryAttemptStatusAcked, ackedAgain.Attempt.Status)
	_, err = appStore.AckAppRuntimeDelivery(ctx, runnerDeliveryAuth, AppRuntimeAckParams{AttemptID: envelope.Attempt.ID, Result: map[string]any{"version": "2.0.0"}})
	require.ErrorIs(t, err, ErrConflict)

	_, err = appStore.FailAppRuntimeDelivery(ctx, runnerDeliveryAuth, AppRuntimeFailParams{AttemptID: envelope.Attempt.ID, Code: "should_not_flip", Message: "should not flip"})
	require.Error(t, err)
	require.Equal(t, "succeeded", loadAppRuntimeDeliveryStatusForTest(t, ctx, pool, envelope.Delivery.ID))

	retrySource, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "@deploy-app-target-runtime-reply upgrade --version 2.0.0", nil)
	require.NoError(t, err)
	auth, err = newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	retryClaim, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-runtime-retry", RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, retryClaim.Envelopes, 1)
	require.Equal(t, retrySource.ID, retryClaim.Envelopes[0].Delivery.EventID)
	retryResult, err := appStore.serviceActorStore().ScheduleDeliveryAttemptRetry(ctx, serviceactor.DeliveryAttemptRetrySchedule{
		WorkspaceID:  fixture.WorkspaceID,
		AttemptID:    retryClaim.Envelopes[0].Attempt.ID,
		ErrorCode:    "transport_failed",
		ErrorDetail:  "network down",
		BaseBackoff:  time.Minute,
		MaxBackoff:   time.Minute,
		ResponseBody: "retry later",
		Now:          now.Add(2 * time.Minute),
	})
	require.NoError(t, err)
	require.Equal(t, "retry_waiting", retryResult.Delivery.Status)
	require.Equal(t, serviceactor.DeliveryAttemptStatusFailed, retryResult.Attempt.Status)
	_, err = appStore.AuthenticateAppRuntimeDelivery(ctx, created.Token, retryClaim.Envelopes[0].Delivery.ID, retryClaim.Envelopes[0].Attempt.ID)
	require.Error(t, err)

	failureSource, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "@deploy-app-target-runtime-reply upgrade --version 9.9.9", nil)
	require.NoError(t, err)
	auth, err = newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	failedClaim, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-runtime-fail", RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, failedClaim.Envelopes, 1)
	require.Equal(t, failureSource.ID, failedClaim.Envelopes[0].Delivery.EventID)
	failureAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, failedClaim.Envelopes[0].DeliveryToken, failedClaim.Envelopes[0].Delivery.ID, failedClaim.Envelopes[0].Attempt.ID)
	require.NoError(t, err)
	failed, err := appStore.FailAppRuntimeDelivery(ctx, failureAuth, AppRuntimeFailParams{
		AttemptID: failedClaim.Envelopes[0].Attempt.ID,
		Code:      "upgrade_failed",
		Message:   "版本不存在",
	})
	require.NoError(t, err)
	require.Equal(t, "failed", failed.Delivery.Status)
	require.Equal(t, serviceactor.DeliveryAttemptStatusFailed, failed.Attempt.Status)
	require.Equal(t, "upgrade_failed", failed.Delivery.ErrorCode)
	require.Equal(t, "版本不存在", failed.Delivery.ErrorDetail)

	failedAgain, err := appStore.FailAppRuntimeDelivery(ctx, failureAuth, AppRuntimeFailParams{
		AttemptID: failedClaim.Envelopes[0].Attempt.ID,
		Code:      "upgrade_failed",
		Message:   "版本不存在",
	})
	require.NoError(t, err)
	require.Equal(t, "failed", failedAgain.Delivery.Status)
	require.Equal(t, serviceactor.DeliveryAttemptStatusFailed, failedAgain.Attempt.Status)
	_, err = appStore.FailAppRuntimeDelivery(ctx, failureAuth, AppRuntimeFailParams{
		AttemptID: failedClaim.Envelopes[0].Attempt.ID,
		Code:      "another_failure",
		Message:   "different terminal outcome",
	})
	require.ErrorIs(t, err, ErrConflict)

	_, err = appStore.AckAppRuntimeDelivery(ctx, failureAuth, AppRuntimeAckParams{AttemptID: failedClaim.Envelopes[0].Attempt.ID})
	require.Error(t, err)
	require.Equal(t, "failed", loadAppRuntimeDeliveryStatusForTest(t, ctx, pool, failedClaim.Envelopes[0].Delivery.ID))

	require.True(t, cancelAppRuntimeDeliveryForTests(t, ctx, appStore, fixture.WorkspaceID, installation.ID, retryClaim.Envelopes[0].Delivery.ID, "maintenance"))
	require.False(t, cancelAppRuntimeDeliveryForTests(t, ctx, appStore, fixture.WorkspaceID, installation.ID, retryClaim.Envelopes[0].Delivery.ID, "maintenance"))
	require.Equal(t, "canceled", loadAppRuntimeDeliveryStatusForTest(t, ctx, pool, retryClaim.Envelopes[0].Delivery.ID))
	var canceledErrorCode, canceledErrorDetail, canceledLeaseOwner string
	var canceledLeaseExpiresAt, canceledNextRetryAt *time.Time
	require.NoError(t, pool.QueryRow(ctx, `
SELECT error_code, error_detail, lease_owner, lease_expires_at, next_retry_at
FROM service_event_deliveries
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, retryClaim.Envelopes[0].Delivery.ID).Scan(&canceledErrorCode, &canceledErrorDetail, &canceledLeaseOwner, &canceledLeaseExpiresAt, &canceledNextRetryAt))
	require.Equal(t, appRuntimeDeliveryCanceledCode, canceledErrorCode)
	require.Equal(t, "maintenance", canceledErrorDetail)
	require.Empty(t, canceledLeaseOwner)
	require.Nil(t, canceledLeaseExpiresAt)
	require.Nil(t, canceledNextRetryAt)
}

func TestAppNotificationActionRuntimeAckRequiresMutationEffect(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_notification_action_runtime", now)
	installation := installAppNotificationActionAppForTargetBindingTestsWithTransports(t, ctx, pool, appStore, fixture, "notification_action_runtime", now, []string{serviceactor.RuntimeTransportKindRunnerStream, serviceactor.RuntimeTransportKindCallback})
	channelID := "chn_bot_notification_action_runtime"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-notification-action-runtime", "App notification action runtime", "private", fixture.MemberID, fixture.MemberID, now)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)

	mentionSubscriptions, err := appStore.serviceActorStore().ListEventSubscriptions(ctx, serviceactor.EventSubscriptionListQuery{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: installation.ID,
		SubscriptionKind: serviceactor.SubscriptionKindAppMention,
		TargetKind:       "channel", TargetRef: channelID,
	})
	require.NoError(t, err)
	require.Len(t, mentionSubscriptions, 1)
	mentionSubscription := mentionSubscriptions[0]

	var actionSubscription serviceactor.EventSubscription
	var actionDelivery serviceactor.EventDelivery
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		txStore := appStore.serviceActorStore().InTx(tx)
		var err error
		actionSubscription, _, err = txStore.UpsertEventSubscription(ctx, serviceactor.EventSubscriptionUpsert{
			ID: "svcesub_bot_notification_action_runtime", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID:   mentionSubscription.ServiceDefinitionID,
			ServiceInstallationID: installation.ID, ServiceMemberID: mentionSubscription.ServiceMemberID,
			ServiceConnectionRef: "app_runtime", ServiceTargetBindingID: mentionSubscription.ServiceTargetBindingID,
			ServiceKind: serviceactor.KindApp, SubscriptionKind: serviceactor.SubscriptionKindAppNotificationAction,
			TargetKind: mentionSubscription.TargetKind, TargetRef: mentionSubscription.TargetRef,
			DeliveryScopeID: mentionSubscription.DeliveryScopeID, PermissionCheckScopeID: mentionSubscription.PermissionCheckScopeID,
			OwnerMemberID: fixture.MemberID, AuthorizerMemberID: fixture.MemberID,
			CreatedByActorType: "human", CreatedByActorID: fixture.UserID, CreatedVia: "test",
			ExternalResourceType: "app_notification_action", ExternalResourceKey: "action:btn_runtime:approve",
			ExternalResourceRef: "btn_runtime", EventFilterKey: "agentteam.app.notification.action",
			EventFilters: []byte(`{"events":["agentteam.app.notification.action"]}`), DeliveryPolicy: "app_runtime_delivery",
			Status:                serviceactor.EventSubscriptionStatusActive,
			ActiveSubscriptionKey: "app-notification-action-runtime:" + installation.ID,
			Now:                   now,
		})
		if err != nil {
			return err
		}
		actionDelivery, _, err = txStore.ReserveEventDelivery(ctx, serviceactor.EventDeliveryReserve{
			ID: "svced_bot_notification_action_runtime", WorkspaceID: fixture.WorkspaceID,
			EventSubscriptionID: actionSubscription.ID, EventType: "agentteam.app.notification.action",
			EventAction: "approve", EventID: "bnaction_runtime", EventKey: "btn_runtime:1:approve",
			EventPayload:         []byte(`{"notificationId":"btn_runtime","actionKey":"approve"}`),
			ExternalResourceType: appNotificationExternalResource, ExternalResourceKey: "btn_runtime",
			ExternalResourceRef: "btn_runtime", IdempotencyKey: "app_notification_action:bnaction_runtime",
			Now: now,
		})
		return err
	}))
	require.Equal(t, serviceactor.SubscriptionKindAppNotificationAction, actionDelivery.SubscriptionKind)

	visibleMessageID := "msg_bot_notification_action_runtime"
	var visibleScopeSeq, visibleWorkspaceSeq int64
	require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(MAX(seq), 0) + 1 FROM scope_messages WHERE scope_id = $1`, channelID).Scan(&visibleScopeSeq))
	require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(MAX(workspace_global_seq), 0) + 1 FROM scope_messages WHERE workspace_id = $1`, fixture.WorkspaceID).Scan(&visibleWorkspaceSeq))
	mustExec(t, pool, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, client_message_id, message_type,
    content, metadata, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, 'service', $7, 'app_interactive_card', $8, JSON_OBJECT(), $9, $9)
`, visibleMessageID, fixture.WorkspaceID, channelID, visibleScopeSeq, visibleWorkspaceSeq, installation.MemberID,
		"app-notification:btn_runtime", "approval notification card", now)
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		if err := appStore.InsertAppNotificationInTx(ctx, tx, AppNotificationRecord{
			ID: "btn_runtime", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID:   mentionSubscription.ServiceDefinitionID,
			ServiceInstallationID: installation.ID, ServiceMemberID: mentionSubscription.ServiceMemberID,
			ServiceTargetBindingID: mentionSubscription.ServiceTargetBindingID,
			TargetKind:             AppNotificationTargetScope, TargetRef: channelID, ScopeID: channelID,
			MessageID: visibleMessageID, ContentKind: AppNotificationContentInteractiveCard,
			ClientNotificationID: "runtime-action-notification", CreatePayloadDigest: strings.Repeat("a", 64),
			Status: AppNotificationStatusActive, Revision: 1,
			CausedByActorType: "human", CausedByActorID: fixture.MemberID,
			ExecutionActorType: "app_runtime", ExecutionActorID: installation.ID,
			CorrelationID: "cor_runtime_action", CreatedAt: now, UpdatedAt: now,
		}); err != nil {
			return err
		}
		if err := appStore.InsertAppNotificationCardInTx(ctx, tx, AppNotificationCardRecord{
			ScopeID: channelID, MessageID: visibleMessageID, AppNotificationID: "btn_runtime",
			SchemaVersion: "1", CardRevision: 1,
			RenderPayloadJSON: json.RawMessage(`{"schemaVersion":"1","title":"Approval"}`),
			ActionSchemaJSON:  json.RawMessage(`{"actions":[{"actionKey":"approve","type":"callback","label":"Approve"},{"actionKey":"reject","type":"callback","label":"Reject"}]}`),
			State:             AppNotificationCardStateActive, CreatedAt: now, UpdatedAt: now,
		}); err != nil {
			return err
		}
		return appStore.InsertAppNotificationActionInTx(ctx, tx, AppNotificationActionRecord{
			ID: "bnaction_runtime", WorkspaceID: fixture.WorkspaceID, AppNotificationID: "btn_runtime",
			MessageID: visibleMessageID, CardRevision: 1, ActionKey: "approve",
			ActorMemberID: fixture.MemberID, ClientActionID: "approve-runtime",
			RequestDigest: strings.Repeat("b", 64), InputDigest: strings.Repeat("c", 64), InputJSON: json.RawMessage(`{}`),
			ServiceEventDeliveryID: actionDelivery.ID, State: AppNotificationActionStateAccepted,
			CorrelationID: "cor_runtime_action", AcceptedAt: now, CreatedAt: now, UpdatedAt: now,
		})
	}))
	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	runnerAuth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, runnerAuth, appruntime.RunnerClaimParams{RunnerID: "runner-notification-action", Limit: 10, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	require.Equal(t, serviceactor.SubscriptionKindAppNotificationAction, envelope.Delivery.SubscriptionKind)

	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	_, err = appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{AttemptID: envelope.Attempt.ID, Content: "must not become a reply"})
	require.ErrorIs(t, err, ErrForbidden)
	_, err = appStore.AckAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeAckParams{AttemptID: envelope.Attempt.ID})
	require.Error(t, err)
	require.Contains(t, err.Error(), "notification mutation effect")

	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		_, err := appStore.serviceActorStore().InTx(tx).AppendDeliveryEffect(ctx, serviceactor.DeliveryEffectAppend{
			WorkspaceID: fixture.WorkspaceID, DeliveryAttemptID: envelope.Attempt.ID,
			EffectKind: string(runeffect.EffectKindNotificationResolved), EffectRef: "btn_runtime",
			TargetKind: envelope.Delivery.TargetKind, TargetRef: envelope.Delivery.TargetRef,
			MessageID: visibleMessageID, Status: serviceactor.DeliveryEffectStatusCreated,
			Metadata: []byte(`{"mutationId":"bnmut_runtime","kind":"resolve","resultRevision":2}`), Now: now.Add(time.Second),
		})
		return err
	}))
	acked, err := appStore.AckAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeAckParams{AttemptID: envelope.Attempt.ID})
	require.NoError(t, err)
	require.Equal(t, string(runeffect.StatusSucceeded), acked.Delivery.Status)
	require.Equal(t, string(runeffect.EffectKindNotificationResolved), acked.Delivery.EffectKind)
	require.Equal(t, "btn_runtime", acked.Delivery.EffectRef)
	require.Equal(t, visibleMessageID, acked.Delivery.MessageID)

	var mutationEffectCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_delivery_effects
WHERE workspace_id = $1
  AND service_event_delivery_id = $2
  AND effect_kind = $3
  AND effect_ref = $4
  AND message_id = $5
`, fixture.WorkspaceID, envelope.Delivery.ID, string(runeffect.EffectKindNotificationResolved), "btn_runtime", visibleMessageID).Scan(&mutationEffectCount))
	require.Equal(t, 1, mutationEffectCount)

	var pendingActionDelivery serviceactor.EventDelivery
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		var err error
		pendingActionDelivery, _, err = appStore.serviceActorStore().InTx(tx).ReserveEventDelivery(ctx, serviceactor.EventDeliveryReserve{
			ID: "svced_bot_notification_action_pending", WorkspaceID: fixture.WorkspaceID,
			EventSubscriptionID: actionSubscription.ID, EventType: "agentteam.app.notification.action",
			EventAction: "reject", EventID: "bnaction_pending", EventKey: "btn_runtime:1:reject",
			EventPayload:         []byte(`{"notificationId":"btn_runtime","actionKey":"reject"}`),
			ExternalResourceType: appNotificationExternalResource, ExternalResourceKey: "btn_runtime",
			ExternalResourceRef: "btn_runtime", IdempotencyKey: "app_notification_action:bnaction_pending",
			Now: now.Add(2 * time.Second),
		})
		return err
	}))
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		return appStore.InsertAppNotificationActionInTx(ctx, tx, AppNotificationActionRecord{
			ID: "bnaction_pending", WorkspaceID: fixture.WorkspaceID, AppNotificationID: "btn_runtime",
			MessageID: visibleMessageID, CardRevision: 1, ActionKey: "reject",
			ActorMemberID: fixture.MemberID, ClientActionID: "reject-runtime",
			RequestDigest: strings.Repeat("d", 64), InputDigest: strings.Repeat("e", 64), InputJSON: json.RawMessage(`{}`),
			ServiceEventDeliveryID: pendingActionDelivery.ID, State: AppNotificationActionStateAccepted,
			CorrelationID: "cor_runtime_action_cancel", AcceptedAt: now.Add(2 * time.Second), CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second),
		})
	}))
	pendingClaim, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, runnerAuth, appruntime.RunnerClaimParams{RunnerID: "runner-notification-action", Limit: 10, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, pendingClaim.Envelopes, 1)
	require.Equal(t, pendingActionDelivery.ID, pendingClaim.Envelopes[0].Delivery.ID)
	require.True(t, cancelAppRuntimeDeliveryForTests(t, ctx, appStore, fixture.WorkspaceID, installation.ID, pendingActionDelivery.ID, "approval expired"))
	require.Equal(t, string(runeffect.StatusCanceled), loadAppRuntimeDeliveryStatusForTest(t, ctx, pool, pendingActionDelivery.ID))
	canceledAttempt, err := appStore.serviceActorStore().LoadDeliveryAttempt(ctx, fixture.WorkspaceID, pendingClaim.Envelopes[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, serviceactor.DeliveryAttemptStatusCanceled, canceledAttempt.Status)
	canceledAction, err := appStore.LoadAppNotificationActionInTx(ctx, pool, fixture.WorkspaceID, "bnaction_pending", false)
	require.NoError(t, err)
	require.Equal(t, appNotificationActionStateCancelled, canceledAction.State)
	require.Equal(t, appRuntimeDeliveryCanceledCode, canceledAction.ResultCode)
	require.NotNil(t, canceledAction.CompletedAt)
	var activeAttemptCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_delivery_attempts
WHERE workspace_id = $1
  AND service_event_delivery_id = $2
  AND status IN ('leased', 'delivered')
`, fixture.WorkspaceID, pendingActionDelivery.ID).Scan(&activeAttemptCount))
	require.Zero(t, activeAttemptCount)
}

func TestAppNotificationCallbackAttemptTransitionsActionProjection(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_notification_callback_action", now)
	installation := installAppNotificationActionAppForTargetBindingTestsWithTransports(t, ctx, pool, appStore, fixture, "notification_callback_action", now, []string{serviceactor.RuntimeTransportKindCallback})
	channelID := "chn_bot_notification_callback_action"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-notification-callback-action", "App notification callback action", "private", fixture.MemberID, fixture.MemberID, now)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	visibleMessageID := "msg_bot_notification_callback_action"
	var visibleScopeSeq, visibleWorkspaceSeq int64
	require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(MAX(seq), 0) + 1 FROM scope_messages WHERE scope_id = $1`, channelID).Scan(&visibleScopeSeq))
	require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(MAX(workspace_global_seq), 0) + 1 FROM scope_messages WHERE workspace_id = $1`, fixture.WorkspaceID).Scan(&visibleWorkspaceSeq))
	mustExec(t, pool, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, client_message_id, message_type,
    content, metadata, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, 'service', $7, 'app_interactive_card', $8, JSON_OBJECT(), $9, $9)
`, visibleMessageID, fixture.WorkspaceID, channelID, visibleScopeSeq, visibleWorkspaceSeq,
		installation.MemberID, "app-notification:btn_callback_action", "callback action projection", now)

	subscriptions, err := appStore.serviceActorStore().ListEventSubscriptions(ctx, serviceactor.EventSubscriptionListQuery{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: installation.ID,
		SubscriptionKind: serviceactor.SubscriptionKindAppMention,
		TargetKind:       "channel", TargetRef: channelID,
	})
	require.NoError(t, err)
	require.Len(t, subscriptions, 1)
	mentionSubscription := subscriptions[0]
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		return appStore.InsertAppNotificationInTx(ctx, tx, AppNotificationRecord{
			ID: "btn_callback_action", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID:   mentionSubscription.ServiceDefinitionID,
			ServiceInstallationID: installation.ID, ServiceMemberID: mentionSubscription.ServiceMemberID,
			ServiceTargetBindingID: mentionSubscription.ServiceTargetBindingID,
			TargetKind:             AppNotificationTargetScope, TargetRef: channelID, ScopeID: channelID,
			MessageID: visibleMessageID, ContentKind: AppNotificationContentInteractiveCard,
			ClientNotificationID: "callback-action-notification", CreatePayloadDigest: strings.Repeat("c", 64),
			Status: AppNotificationStatusActive, Revision: 1,
			CausedByActorType: "human", CausedByActorID: fixture.MemberID,
			ExecutionActorType: appNotificationExecutionActorAppRuntime, ExecutionActorID: installation.ID,
			CorrelationID: "cor_callback_action", CreatedAt: now, UpdatedAt: now,
		})
	}))

	var delivery serviceactor.EventDelivery
	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		txStore := appStore.serviceActorStore().InTx(tx)
		subscription, _, err := txStore.UpsertEventSubscription(ctx, serviceactor.EventSubscriptionUpsert{
			ID: "svcesub_bot_notification_callback_action", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID:   mentionSubscription.ServiceDefinitionID,
			ServiceInstallationID: installation.ID, ServiceMemberID: mentionSubscription.ServiceMemberID,
			ServiceConnectionRef:   serviceactor.AppRuntimeConnectionRefExternal,
			ServiceTargetBindingID: mentionSubscription.ServiceTargetBindingID,
			ServiceKind:            serviceactor.KindApp, SubscriptionKind: serviceactor.SubscriptionKindAppNotificationAction,
			TargetKind: mentionSubscription.TargetKind, TargetRef: mentionSubscription.TargetRef,
			DeliveryScopeID: mentionSubscription.DeliveryScopeID, PermissionCheckScopeID: mentionSubscription.PermissionCheckScopeID,
			OwnerMemberID: fixture.MemberID, AuthorizerMemberID: fixture.MemberID,
			CreatedByActorType: "human", CreatedByActorID: fixture.UserID, CreatedVia: "test",
			ExternalResourceType: appNotificationExternalResource, ExternalResourceKey: "btn_callback_action",
			ExternalResourceRef: "btn_callback_action", EventFilterKey: "agentteam.app.notification.action",
			EventFilters:   []byte(`{"events":["agentteam.app.notification.action"]}`),
			DeliveryPolicy: serviceactor.AppRuntimeDeliveryPolicyExternal,
			Status:         serviceactor.EventSubscriptionStatusActive, ActiveSubscriptionKey: "callback-action:" + installation.ID,
			Now: now,
		})
		if err != nil {
			return err
		}
		delivery, _, err = txStore.ReserveEventDelivery(ctx, serviceactor.EventDeliveryReserve{
			ID: "svced_bot_notification_callback_action", WorkspaceID: fixture.WorkspaceID,
			EventSubscriptionID: subscription.ID, EventType: "agentteam.app.notification.action",
			EventAction: "approve", EventID: "bnaction_callback", EventKey: "btn_callback_action:1:approve",
			EventPayload:         []byte(`{"notificationId":"btn_callback_action","actionKey":"approve"}`),
			ExternalResourceType: appNotificationExternalResource, ExternalResourceKey: "btn_callback_action",
			ExternalResourceRef: "btn_callback_action", IdempotencyKey: "app_notification_action:bnaction_callback",
			Now: now,
		})
		if err != nil {
			return err
		}
		return nil
	}))
	mustExec(t, pool, `
INSERT INTO service_notification_actions (
    id, workspace_id, service_notification_id, message_id, card_revision,
    action_key, actor_member_id, client_action_id, request_digest,
    input_digest, input_json, service_event_delivery_id, state,
    result_code, result_digest, correlation_id, accepted_at,
    created_at, updated_at
) VALUES (
    $1, $2, $3, $4, 1,
    'approve', $5, 'approve-callback', $6,
    $7, JSON_OBJECT(), $8, 'accepted',
    '', '', 'cor_callback_action', $9,
    $9, $9
)
`, "bnaction_callback", fixture.WorkspaceID, "btn_callback_action", visibleMessageID,
		fixture.MemberID, strings.Repeat("a", 64), strings.Repeat("b", 64), delivery.ID, now)

	envelopes, err := appStore.ClaimDeliveryEnvelopes(ctx, serviceactor.DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: installation.ID,
		SubscriptionKind: serviceactor.SubscriptionKindAppNotificationAction,
		DeliveryPolicy:   serviceactor.AppRuntimeDeliveryPolicyExternal, ServiceConnectionRef: serviceactor.AppRuntimeConnectionRefExternal,
		TransportKind: serviceactor.RuntimeTransportKindCallback, LeaseOwner: "callback:" + installation.ID,
		LeaseDuration: time.Minute, Limit: 1, Now: now.Add(time.Second),
		CallbackURLSnapshot: "https://app.example.test/events",
	})
	require.NoError(t, err)
	require.Len(t, envelopes, 1)
	envelope := envelopes[0]

	_, err = appStore.MarkDeliveryAttemptDelivered(ctx, serviceactor.DeliveryAttemptDeliveredMark{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID, Now: now.Add(2 * time.Second),
	})
	require.NoError(t, err)
	action, err := appStore.LoadAppNotificationActionInTx(ctx, pool, fixture.WorkspaceID, "bnaction_callback", false)
	require.NoError(t, err)
	require.Equal(t, AppNotificationActionStateDelivered, action.State)

	retry, err := appStore.ScheduleDeliveryAttemptRetry(ctx, serviceactor.DeliveryAttemptRetrySchedule{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID,
		ErrorCode: "callback_failed", ErrorDetail: "temporary", Now: now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, serviceactor.DeliveryAttemptStatusFailed, retry.Attempt.Status)
	action, err = appStore.LoadAppNotificationActionInTx(ctx, pool, fixture.WorkspaceID, "bnaction_callback", false)
	require.NoError(t, err)
	require.Equal(t, AppNotificationActionStateDelivered, action.State)
	require.Nil(t, action.CompletedAt)
}

func TestAppRuntimeThreadMentionDeliveryRepairsMembershipAndRepliesInThread(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	fanout := &recordingCommittedMessageFanout{}
	appStore.SetCommittedMessageFanout(fanout)
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runtime_thread_mention", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "thread_mention", now)
	channelID := "chn_bot_runtime_thread_mention"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-runtime-thread-mention", "App runtime thread mention", "private", fixture.MemberID, fixture.MemberID, now)

	parentMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "thread parent for App mention", nil)
	require.NoError(t, err)
	replyThread, err := message.OpenReplyThreadForMember(ctx, pool, fixture.MemberID, fixture.WorkspaceID, channelID, parentMessage.ID)
	require.NoError(t, err)
	threadID := replyThread.ThreadChannelID
	require.NotEmpty(t, threadID)

	_, err = appStore.bindAppInstallationTargetForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, threadID, AppMentionCreatedViaTarget)
	require.NoError(t, err)
	requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)

	require.NoError(t, coredb.WithTx(ctx, pool, func(tx coredb.Tx) error {
		return message.ReplaceReplyThreadParticipants(ctx, tx, threadID, []string{fixture.UserID})
	}))
	requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)

	staleAt := now.Add(time.Minute)
	mustExec(t, pool, `
UPDATE scope_memberships
SET removed_at = $3,
    archived_at = $3,
    updated_at = $3
WHERE scope_id = $1
  AND member_id = $2
`, threadID, installation.MemberID, staleAt)
	requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)

	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	auth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)

	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, threadID, "@deploy-app-target-thread-mention inspect this thread", nil)
	require.NoError(t, err)
	requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)

	record := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, sourceMessage.ID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, record.SubscriptionKind)
	require.Equal(t, "thread", record.TargetKind)
	require.Equal(t, threadID, record.TargetRef)
	require.Equal(t, threadID, record.DeliveryScopeID)
	require.Equal(t, channelID, record.PermissionCheckScopeID)

	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-thread-mention", RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	require.Equal(t, sourceMessage.ID, envelope.Delivery.EventID)

	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	reply, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "thread-scoped App reply",
		IdempotencyKey: "thread-progress-1",
	})
	require.NoError(t, err)
	require.NotEmpty(t, reply.MessageID)
	require.Len(t, fanout.deliveries, 1)
	committed := fanout.deliveries[0]
	require.Equal(t, reply.MessageID, committed.MessageID)
	require.NotNil(t, committed.AppMessage)
	require.Equal(t, threadID, committed.AppMessage.ChannelID)
	require.True(t, committed.RecoverableStreamProjection)
	require.Empty(t, committed.WakeRecipients)
	require.True(t, requireScopeFanoutPayloadForMessage(t, pool, reply.MessageID).Delivery.RecoverableStreamProjection)

	replayed, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "thread-scoped App reply",
		IdempotencyKey: "thread-progress-1",
	})
	require.NoError(t, err)
	require.Equal(t, reply.MessageID, replayed.MessageID)
	require.Len(t, fanout.deliveries, 1, "idempotent thread reply must not fan out again")
	requireScopeMessageAndSingleFanout(t, pool, reply.MessageID, fixture.WorkspaceID, threadID)

	var replyScopeID, replyContent, replyAuthorType, replyAuthorMember string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT scope_id, content, author_type, author_member_id
FROM scope_messages
WHERE id = $1
`, reply.MessageID).Scan(&replyScopeID, &replyContent, &replyAuthorType, &replyAuthorMember))
	require.Equal(t, threadID, replyScopeID)
	require.NotEqual(t, channelID, replyScopeID)
	require.Equal(t, "thread-scoped App reply", replyContent)
	require.Equal(t, "service", replyAuthorType)
	require.Equal(t, installation.MemberID, replyAuthorMember)
}

func TestAppRuntimeThreadMentionDeliveryCreatesThreadTargetForChannelBoundApp(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runtime_thread_channel_bound", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "thread_channel_bound", now)
	channelID := "chn_bot_runtime_thread_channel_bound"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-runtime-thread-channel-bound", "App runtime thread channel-bound", "private", fixture.MemberID, fixture.MemberID, now)

	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	assertActiveAppTargetBinding(t, ctx, pool, fixture.WorkspaceID, installation.ID, installation.MemberID, channelID, "channel", channelID, channelID)

	parentMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "thread parent for channel-bound App mention", nil)
	require.NoError(t, err)
	replyThread, err := message.OpenReplyThreadForMember(ctx, pool, fixture.MemberID, fixture.WorkspaceID, channelID, parentMessage.ID)
	require.NoError(t, err)
	threadID := replyThread.ThreadChannelID
	require.NotEmpty(t, threadID)
	requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
	requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)

	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	auth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)

	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, threadID, "@deploy-app-target-thread-channel-bound inspect this thread", nil)
	require.NoError(t, err)
	assertActiveAppTargetBinding(t, ctx, pool, fixture.WorkspaceID, installation.ID, installation.MemberID, threadID, "thread", threadID, channelID)

	record := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, sourceMessage.ID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, record.SubscriptionKind)
	require.Equal(t, "thread", record.TargetKind)
	require.Equal(t, threadID, record.TargetRef)
	require.Equal(t, threadID, record.DeliveryScopeID)
	require.Equal(t, channelID, record.PermissionCheckScopeID)

	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-thread-channel-bound", RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	require.Equal(t, sourceMessage.ID, envelope.Delivery.EventID)

	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	reply, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "channel-bound App reply in thread",
		IdempotencyKey: "thread-channel-bound-progress-1",
	})
	require.NoError(t, err)
	require.NotEmpty(t, reply.MessageID)

	var replyScopeID, replyContent, replyAuthorType, replyAuthorMember string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT scope_id, content, author_type, author_member_id
FROM scope_messages
WHERE id = $1
`, reply.MessageID).Scan(&replyScopeID, &replyContent, &replyAuthorType, &replyAuthorMember))
	require.Equal(t, threadID, replyScopeID)
	require.NotEqual(t, channelID, replyScopeID)
	require.Equal(t, "channel-bound App reply in thread", replyContent)
	require.Equal(t, "service", replyAuthorType)
	require.Equal(t, installation.MemberID, replyAuthorMember)
}

func TestAppRuntimeThreadMentionDeliveryRequiresUnexpiredGrantsBeforeMaterializingTarget(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		name   string
		suffix string
		scope  string
	}{
		{name: "expired mention read grant", suffix: "expired_mention_read", scope: AppMentionScopeRead},
		{name: "expired message write grant", suffix: "expired_message_write", scope: AppRuntimeScopeMessageWrite},
	}

	for _, testCase := range testCases {
		testCase := testCase
		t.Run(testCase.name, func(t *testing.T) {
			ctx := context.Background()
			pool := startAppRollbackDB(t, ctx)
			appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
			now := time.Now().UTC()
			fixture := seedAppAdminFixture(t, pool, "bot_runtime_thread_"+testCase.suffix, now)
			installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "thread_"+testCase.suffix, now)
			channelID := "chn_bot_runtime_thread_" + testCase.suffix
			channelSlug := "app-runtime-thread-" + strings.ReplaceAll(testCase.suffix, "_", "-")
			seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, channelSlug, "App runtime expired grant", "private", fixture.MemberID, fixture.MemberID, now)

			_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
			require.NoError(t, err)

			parentMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, channelID, "thread parent for expired App grant", nil)
			require.NoError(t, err)
			replyThread, err := message.OpenReplyThreadForMember(ctx, pool, fixture.MemberID, fixture.WorkspaceID, channelID, parentMessage.ID)
			require.NoError(t, err)
			threadID := replyThread.ThreadChannelID
			require.NotEmpty(t, threadID)
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)

			mustExec(t, pool, `
UPDATE service_grants
SET expires_at = CURRENT_TIMESTAMP - INTERVAL 1 HOUR,
    updated_at = CURRENT_TIMESTAMP
WHERE service_installation_id = $1
  AND scope = $2
  AND revoked_at IS NULL
`, installation.ID, testCase.scope)

			appHandle := "@deploy-app-target-thread-" + strings.ReplaceAll(testCase.suffix, "_", "-")
			expiredMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" must not materialize", nil)
			require.NoError(t, err)
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredMessage.ID, installation.ID, 0)

			mustExec(t, pool, `
UPDATE service_grants
SET expires_at = CURRENT_TIMESTAMP + INTERVAL 1 HOUR,
    updated_at = CURRENT_TIMESTAMP
WHERE service_installation_id = $1
  AND scope = $2
  AND revoked_at IS NULL
`, installation.ID, testCase.scope)
			// Renewing a grant is not a replay trigger: the old message must not
			// retroactively materialize target state or create a delivery.
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredMessage.ID, installation.ID, 0)

			renewedMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" materialize after renewal", nil)
			require.NoError(t, err)
			requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			assertActiveAppTargetBinding(t, ctx, pool, fixture.WorkspaceID, installation.ID, installation.MemberID, threadID, "thread", threadID, channelID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, renewedMessage.ID, installation.ID)

			followupMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" reuse renewed thread target", nil)
			require.NoError(t, err)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, followupMessage.ID, installation.ID)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredMessage.ID, installation.ID, 0)

			mustExec(t, pool, `
UPDATE service_grants
SET expires_at = CURRENT_TIMESTAMP - INTERVAL 1 HOUR,
    updated_at = CURRENT_TIMESTAMP
WHERE service_installation_id = $1
  AND scope = $2
  AND revoked_at IS NULL
`, installation.ID, testCase.scope)
			// An already-materialized target must not bypass grant expiry when
			// loading delivery candidates.
			expiredBoundMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" must not deliver through an existing target", nil)
			require.NoError(t, err)
			requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredBoundMessage.ID, installation.ID, 0)

			mustExec(t, pool, `
UPDATE scope_memberships
SET removed_at = CURRENT_TIMESTAMP,
    archived_at = CURRENT_TIMESTAMP,
    updated_at = CURRENT_TIMESTAMP
WHERE scope_id = $1
  AND member_id = $2
`, threadID, installation.MemberID)
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			// The same expiry fence must prevent the existing target's membership
			// repair path from rematerializing the App in the thread.
			expiredRepairMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" must not repair membership", nil)
			require.NoError(t, err)
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredRepairMessage.ID, installation.ID, 0)

			mustExec(t, pool, `
UPDATE service_grants
SET expires_at = CURRENT_TIMESTAMP + INTERVAL 1 HOUR,
    updated_at = CURRENT_TIMESTAMP
WHERE service_installation_id = $1
  AND scope = $2
  AND revoked_at IS NULL
`, installation.ID, testCase.scope)
			requireInactiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredBoundMessage.ID, installation.ID, 0)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, expiredRepairMessage.ID, installation.ID, 0)

			postRenewalMessage, _, err := appStore.SendMessageForUser(ctx, model.User{
				ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
			}, fixture.WorkspaceID, threadID, appHandle+" repair through the reused target after renewal", nil)
			require.NoError(t, err)
			requireActiveAppThreadMembershipForTest(t, ctx, pool, threadID, installation.MemberID)
			requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
			loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, postRenewalMessage.ID, installation.ID)
			requireAppMentionDeliveryCountForTest(t, ctx, pool, fixture.WorkspaceID, "", installation.ID, 3)
		})
	}
}

func TestAppRuntimeThreadMentionDeliveryReusesExistingThreadTargetForHumanAndAgentFollowups(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runtime_thread_agent_followup", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "thread_agent_followup", now)
	channelID := "chn_bot_runtime_thread_agent_followup"
	channelSlug := "app-runtime-thread-agent-followup"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, channelSlug, "App runtime thread agent follow-up", "private", fixture.MemberID, fixture.MemberID, now)

	authorAgentID := "agt_brt_agent_followup"
	authorAgentMemberID := "mem_brt_agent_followup"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, channelID, "mac_brt_agent_followup", authorAgentID, authorAgentMemberID, "app-runtime-thread-agent-followup", now)

	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)

	parentMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "thread parent for agent App mention follow-up", nil)
	require.NoError(t, err)
	replyThread, err := message.OpenReplyThreadForMember(ctx, pool, fixture.MemberID, fixture.WorkspaceID, channelID, parentMessage.ID)
	require.NoError(t, err)
	threadID := replyThread.ThreadChannelID
	require.NotEmpty(t, threadID)
	requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 0)

	sourceMessage, _, err := appStore.SendMessageForUser(
		ctx,
		model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"},
		fixture.WorkspaceID,
		threadID,
		fmt.Sprintf("[@display-token-does-not-match-handle](/mentions/service/%s) inspect this thread", installation.MemberID),
		nil,
	)
	require.NoError(t, err)
	assertActiveAppTargetBinding(t, ctx, pool, fixture.WorkspaceID, installation.ID, installation.MemberID, threadID, "thread", threadID, channelID)
	requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
	record := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, sourceMessage.ID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, record.SubscriptionKind)
	require.Equal(t, "thread", record.TargetKind)
	require.Equal(t, threadID, record.TargetRef)
	requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)

	// The same human mentioning the same App in the same thread again must
	// reuse the thread binding/subscription. This is the original regression:
	// attempting to materialize a second thread target used to revalidate the
	// parent and fail with "requires an active definition of the same kind".
	humanFollowup, _, err := appStore.SendMessageForUser(ctx, model.User{
		ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator",
	}, fixture.WorkspaceID, threadID, "@deploy-app-target-thread-agent-followup inspect this thread again", nil)
	require.NoError(t, err)
	requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
	requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
	humanFollowupRecord := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, humanFollowup.ID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, humanFollowupRecord.SubscriptionKind)
	require.Equal(t, "thread", humanFollowupRecord.TargetKind)
	require.Equal(t, threadID, humanFollowupRecord.TargetRef)
	require.Equal(t, threadID, humanFollowupRecord.DeliveryScopeID)
	require.Equal(t, channelID, humanFollowupRecord.PermissionCheckScopeID)

	followup, err := appStore.SendAgentThreadMessage(ctx, agent.AuthenticatedAgent{
		AgentID:     authorAgentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    authorAgentMemberID,
	}, message.SendMessageParams{
		TargetRef: "#" + channelSlug + ":" + parentMessage.ID[:8],
		Content:   "@deploy-app-target-thread-agent-followup inspect this thread again",
	})
	require.NoError(t, err)
	require.NotEmpty(t, followup.MessageID)
	requireAppMentionTargetBindingCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
	requireAppMentionSubscriptionCountForTest(t, ctx, pool, fixture.WorkspaceID, installation.ID, "thread", threadID, 1)
	followupRecord := loadAppMentionDeliveryForTest(t, ctx, pool, fixture.WorkspaceID, followup.MessageID, installation.ID)
	require.Equal(t, serviceactor.SubscriptionKindAppMention, followupRecord.SubscriptionKind)
	require.Equal(t, "thread", followupRecord.TargetKind)
	require.Equal(t, threadID, followupRecord.TargetRef)
	require.Equal(t, threadID, followupRecord.DeliveryScopeID)
	require.Equal(t, channelID, followupRecord.PermissionCheckScopeID)
}

func TestAppRuntimeContextReadUsesScopeVisibilityAndTargetBinding(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	setup := setupAppRuntimeDeliveryForContextTest(t, ctx, "runtime_context_binding")
	pool := setup.Pool
	appStore := setup.Store
	envelope := setup.Envelope
	deliveryAuth := setup.DeliveryAuth
	sourceMessageID := setup.SourceMessageID

	newerMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: setup.Fixture.UserID, Name: setup.Fixture.UserID, DisplayName: "App Operator"}, setup.Fixture.WorkspaceID, setup.ChannelID, "context after the app mention", nil)
	require.NoError(t, err)

	page, err := appStore.ReadAppRuntimeDeliveryContext(ctx, deliveryAuth, AppRuntimeContextParams{AttemptID: envelope.Attempt.ID, Limit: 1})
	require.NoError(t, err)
	require.GreaterOrEqual(t, len(page.Messages), 2)
	require.Equal(t, sourceMessageID, page.Messages[0].ID)
	listed := requireListedAppMessage(t, page.Messages, sourceMessageID)
	require.Contains(t, listed.Content, "upgrade --version 3.0.0")
	requireListedAppMessage(t, page.Messages, newerMessage.ID)

	mustExec(t, pool, `
UPDATE service_target_bindings
SET status = 'disabled',
    disabled_reason = 'test disabled binding',
    active_binding_key = NULL,
    updated_at = $2
WHERE id = $1
`, envelope.Delivery.ServiceTargetBindingID, time.Now().UTC().Add(time.Minute))

	_, err = appStore.ReadAppRuntimeDeliveryContext(ctx, deliveryAuth, AppRuntimeContextParams{AttemptID: envelope.Attempt.ID, Limit: 10})
	require.Error(t, err)
	_, err = appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "binding disabled",
		IdempotencyKey: "binding-disabled",
	})
	require.Error(t, err)
}

func TestAppRuntimeContextReadRequiresRuntimeReadGrants(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	setup := setupAppRuntimeDeliveryForContextTest(t, ctx, "runtime_context_grants")
	pool := setup.Pool
	appStore := setup.Store
	installation := setup.Installation
	envelope := setup.Envelope
	deliveryAuth := setup.DeliveryAuth

	_, err := appStore.ReadAppRuntimeDeliveryContext(ctx, deliveryAuth, AppRuntimeContextParams{AttemptID: envelope.Attempt.ID, Limit: 10})
	require.NoError(t, err)

	mustExec(t, pool, `
UPDATE service_grants
SET revoked_at = $2,
    updated_at = $2
WHERE service_installation_id = $1
  AND scope = $3
  AND revoked_at IS NULL
`, installation.ID, time.Now().UTC().Add(time.Minute), AppRuntimeScopeMessageRead)

	_, err = appStore.ReadAppRuntimeDeliveryContext(ctx, deliveryAuth, AppRuntimeContextParams{AttemptID: envelope.Attempt.ID, Limit: 10})
	require.Error(t, err)
}

func TestAppRuntimeContextReadAcceptsCallbackDeliveryTokenWithoutRunnerStream(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runtime_context_callback", now)
	installation := installAppAppForTargetBindingTestsWithTransports(t, ctx, pool, appStore, fixture, "runtime_context_callback", now, []string{serviceactor.RuntimeTransportKindCallback})
	channelID := "chn_bot_runtime_context_callback"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-runtime-context-callback", "App runtime context callback", "private", fixture.MemberID, fixture.MemberID, now)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)

	_, err = appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.Error(t, err)

	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "@deploy-app-target-runtime-context-callback upgrade --version 4.0.0", nil)
	require.NoError(t, err)
	envelopes, err := appStore.serviceActorStore().ClaimDeliveryEnvelopes(ctx, serviceactor.DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: installation.ID,
		SubscriptionKind:      serviceactor.SubscriptionKindAppMention,
		TransportKind:         serviceactor.RuntimeTransportKindCallback,
		LeaseOwner:            "callback:" + installation.ID,
		LeaseDuration:         time.Minute,
		Limit:                 1,
		Now:                   now.Add(time.Minute),
		CallbackURLSnapshot:   "https://app.example.test/events",
	})
	require.NoError(t, err)
	require.Len(t, envelopes, 1)
	envelope := envelopes[0]
	require.Equal(t, sourceMessage.ID, envelope.Delivery.EventID)
	require.Equal(t, serviceactor.RuntimeTransportKindCallback, envelope.Attempt.TransportKind)
	require.Empty(t, envelope.Attempt.ServiceRuntimeCredentialID)
	require.NotEmpty(t, envelope.DeliveryToken)

	statusCode := 202
	deliveredAttempt, err := appStore.serviceActorStore().MarkDeliveryAttemptDelivered(ctx, serviceactor.DeliveryAttemptDeliveredMark{
		WorkspaceID:        fixture.WorkspaceID,
		AttemptID:          envelope.Attempt.ID,
		ResponseStatusCode: &statusCode,
		ResponseExcerpt:    "accepted",
		Now:                now.Add(90 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, serviceactor.DeliveryAttemptStatusDelivered, deliveredAttempt.Status)

	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, deliveredAttempt.ID)
	require.NoError(t, err)
	require.Empty(t, deliveryAuth.ServiceRuntimeCredentialID)
	page, err := appStore.ReadAppRuntimeDeliveryContext(ctx, deliveryAuth, AppRuntimeContextParams{AttemptID: deliveredAttempt.ID, Limit: 1})
	require.NoError(t, err)
	listed := requireListedAppMessage(t, page.Messages, sourceMessage.ID)
	require.Contains(t, listed.Content, "upgrade --version 4.0.0")

	reply, err := appStore.ReplyAppRuntimeDelivery(ctx, deliveryAuth, AppRuntimeReplyParams{
		AttemptID:      deliveredAttempt.ID,
		Content:        "callback runtime accepted 4.0.0",
		IdempotencyKey: "callback-progress-1",
	})
	require.NoError(t, err)
	require.NotEmpty(t, reply.MessageID)
}

func loadAppRuntimeDeliveryStatusForTest(t *testing.T, ctx context.Context, queryer coredb.QueryRower, deliveryID string) string {
	t.Helper()
	var status string
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT status
FROM service_event_deliveries
WHERE id = $1
`, deliveryID).Scan(&status))
	return status
}

func requireActiveAppThreadMembershipForTest(t *testing.T, ctx context.Context, queryer coredb.QueryRower, threadID string, memberID string) {
	t.Helper()
	var count int
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND visibility = 'public'
  AND removed_at IS NULL
  AND archived_at IS NULL
  AND effective_to_seq IS NULL
`, threadID, memberID).Scan(&count))
	require.Equal(t, 1, count)
}

func requireInactiveAppThreadMembershipForTest(t *testing.T, ctx context.Context, queryer coredb.QueryRower, threadID string, memberID string) {
	t.Helper()
	var count int
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND archived_at IS NULL
  AND effective_to_seq IS NULL
`, threadID, memberID).Scan(&count))
	require.Zero(t, count)
}

func requireAppMentionTargetBindingCountForTest(t *testing.T, ctx context.Context, queryer coredb.QueryRower, workspaceID string, installationID string, targetKind string, targetRef string, want int) {
	t.Helper()
	var count int
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_target_bindings
WHERE workspace_id = $1
  AND service_installation_id = $2
  AND target_kind = $3
  AND target_ref = $4
  AND status = 'active'
  AND removed_at IS NULL
`, workspaceID, installationID, targetKind, targetRef).Scan(&count))
	require.Equal(t, want, count)
}

func requireAppMentionSubscriptionCountForTest(t *testing.T, ctx context.Context, queryer coredb.QueryRower, workspaceID string, installationID string, targetKind string, targetRef string, want int) {
	t.Helper()
	var count int
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE workspace_id = $1
  AND service_installation_id = $2
  AND subscription_kind = $3
  AND event_filter_key = $4
  AND target_kind = $5
  AND target_ref = $6
  AND status = 'active'
  AND active_subscription_key IS NOT NULL
  AND removed_at IS NULL
`, workspaceID, installationID, serviceactor.SubscriptionKindAppMention, registry.AppEventMessageMention, targetKind, targetRef).Scan(&count))
	require.Equal(t, want, count)
}

type appRuntimeDeliveryContextTestSetup struct {
	Pool            coredb.DB
	Store           *Store
	Fixture         appAdminFixture
	Installation    appTargetInstallationFixture
	Envelope        appruntime.RunnerDeliveryEnvelopeView
	DeliveryAuth    AppRuntimeDeliveryAuth
	ChannelID       string
	SourceMessageID string
}

func setupAppRuntimeDeliveryForContextTest(t *testing.T, ctx context.Context, suffix string) appRuntimeDeliveryContextTestSetup {
	t.Helper()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_"+suffix, now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, suffix, now)
	channelID := "chn_bot_" + suffix
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, "app-"+strings.ReplaceAll(suffix, "_", "-"), "App "+suffix, "private", fixture.MemberID, fixture.MemberID, now)
	_, err := appStore.AddChannelServiceForUserWithEvent(ctx, fixture.UserID, fixture.WorkspaceID, channelID, installation.MemberID)
	require.NoError(t, err)
	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	auth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	slugSuffix := strings.ReplaceAll(suffix, "_", "-")
	sourceMessage, _, err := appStore.SendMessageForUser(ctx, model.User{ID: fixture.UserID, Name: fixture.UserID, DisplayName: "App Operator"}, fixture.WorkspaceID, channelID, "@deploy-app-target-"+slugSuffix+" upgrade --version 3.0.0", nil)
	require.NoError(t, err)
	claimed, err := newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, auth, appruntime.RunnerClaimParams{RunnerID: "runner-" + slugSuffix, RunnerVersion: "1.0.0", Limit: 1, LeaseDuration: time.Minute})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	require.Equal(t, sourceMessage.ID, envelope.Delivery.EventID)
	deliveryAuth, err := appStore.AuthenticateAppRuntimeDelivery(ctx, envelope.DeliveryToken, envelope.Delivery.ID, envelope.Attempt.ID)
	require.NoError(t, err)
	return appRuntimeDeliveryContextTestSetup{
		Pool:            pool,
		Store:           appStore,
		Fixture:         fixture,
		Installation:    installation,
		Envelope:        envelope,
		DeliveryAuth:    deliveryAuth,
		ChannelID:       channelID,
		SourceMessageID: sourceMessage.ID,
	}
}

func TestAppRunnerRuntimeRequiresCurrentInstallationCapabilityAndGrants(t *testing.T) {
	t.Parallel()

	for name, mutate := range map[string]func(*testing.T, context.Context, coredb.QueryExecer, string, time.Time){
		"disabled installation": func(t *testing.T, ctx context.Context, pool coredb.QueryExecer, installationID string, now time.Time) {
			mustExec(t, pool, `
UPDATE service_installations
SET status = $2,
    updated_at = $3
WHERE id = $1
`, installationID, serviceactor.StatusDisabled, now)
		},
		"revoked app mention grant": func(t *testing.T, ctx context.Context, pool coredb.QueryExecer, installationID string, now time.Time) {
			mustExec(t, pool, `
UPDATE service_grants
SET revoked_at = $2,
    updated_at = $2
WHERE service_installation_id = $1
  AND scope = $3
  AND revoked_at IS NULL
`, installationID, now, AppMentionScopeRead)
		},
		"pinned app capability removed": func(t *testing.T, ctx context.Context, pool coredb.QueryExecer, installationID string, now time.Time) {
			mustExec(t, pool, `
UPDATE service_installations
SET metadata = JSON_REMOVE(metadata, '$.manifest.capabilities.runtime'),
    updated_at = $2
WHERE id = $1
`, installationID, now)
		},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			ctx := context.Background()
			pool := startAppRollbackDB(t, ctx)
			appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
			now := time.Now().UTC()
			fixture := seedAppAdminFixture(t, pool, "bot_runner_current_"+strings.ReplaceAll(name, " ", "_"), now)
			installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "runner_current_"+strings.ReplaceAll(name, " ", "_"), now)

			created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
			require.NoError(t, err)
			staleAuth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
			require.NoError(t, err)

			mutate(t, ctx, pool, installation.ID, now.Add(time.Minute))

			_, err = newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
			require.Error(t, err)
			_, err = newRunnerServiceForTest(appStore).Heartbeat(ctx, staleAuth, appruntime.RunnerHeartbeatParams{RunnerID: "runner-current"})
			require.Error(t, err)
			_, err = newRunnerServiceForTest(appStore).ClaimDeliveries(ctx, staleAuth, appruntime.RunnerClaimParams{RunnerID: "runner-current", Limit: 1})
			require.Error(t, err)

			var onlineCount int
			require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_runtime_sessions
WHERE service_installation_id = $1
  AND status = $2
`, installation.ID, serviceactor.RuntimeSessionStatusOnline).Scan(&onlineCount))
			require.Zero(t, onlineCount)
		})
	}
}

func TestAppRunnerRuntimeIgnoresMutableDefinitionRuntimeCapabilityChanges(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "bot_runner_pinned_definition_change", now)
	installation := installAppAppForTargetBindingTests(t, ctx, pool, appStore, fixture, "runner_pinned_definition_change", now)

	created, err := appStore.CreateAppRunnerCredentialForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, AppRunnerCredentialCreateParams{})
	require.NoError(t, err)
	staleAuth, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)

	mustExec(t, pool, `
UPDATE service_capabilities
SET default_enabled = FALSE,
    updated_at = $2
WHERE service_definition_id = (
    SELECT service_definition_id
    FROM service_installations
    WHERE id = $1
)
  AND capability_key = $3
`, installation.ID, now.Add(time.Minute), serviceactor.CapabilityApp)

	refreshed, err := newRunnerServiceForTest(appStore).AuthenticateRunner(ctx, created.Token)
	require.NoError(t, err)
	require.Equal(t, installation.ID, refreshed.ServiceInstallationID)
	_, err = newRunnerServiceForTest(appStore).Heartbeat(ctx, staleAuth, appruntime.RunnerHeartbeatParams{RunnerID: "runner-pinned-definition-change"})
	require.NoError(t, err)
}