package serviceactor

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

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	fixture := seedServiceActorFixture(t, db, "runtime_credential", time.Now().UTC())
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-credential")

	created, err := store.CreateRuntimeCredentialToken(ctx, RuntimeCredentialTokenCreate{
		RuntimeCredentialCreate: RuntimeCredentialCreate{
			WorkspaceID:           fixture.WorkspaceID,
			ServiceDefinitionID:   seed.DefinitionID,
			ServiceInstallationID: seed.InstallationID,
			ServiceMemberID:       seed.ServiceMemberID,
			ServiceKind:           "app",
			CredentialKind:        RuntimeCredentialKindRunnerStreamToken,
			TransportKind:         RuntimeTransportKindRunnerStream,
			Name:                  "primary runner",
			CreatedByMemberID:     fixture.HumanMemberID,
			Metadata:              []byte(`{"env":"staging"}`),
		},
	})
	require.NoError(t, err)
	require.NotEmpty(t, created.Token)
	require.NotContains(t, created.Credential.TokenHash, created.Token)
	require.Equal(t, secrets.HashToken(created.Token), created.Credential.TokenHash)
	require.Equal(t, RuntimeCredentialStatusActive, created.Credential.Status)
	require.JSONEq(t, `{"env":"staging"}`, string(created.Credential.Metadata))

	authenticated, err := store.AuthenticateRuntimeCredential(ctx, created.Token)
	require.NoError(t, err)
	require.Equal(t, created.Credential.ID, authenticated.ID)
	require.NotNil(t, authenticated.LastUsedAt)

	disabled, err := store.DisableRuntimeCredential(ctx, fixture.WorkspaceID, created.Credential.ID, time.Now().UTC())
	require.NoError(t, err)
	require.Equal(t, RuntimeCredentialStatusDisabled, disabled.Status)
	require.NotNil(t, disabled.DisabledAt)

	_, err = store.AuthenticateRuntimeCredential(ctx, created.Token)
	require.ErrorIs(t, err, ErrNotFound)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	fixture := seedServiceActorFixture(t, db, "runtime_session", time.Now().UTC())
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-session")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-session-token")

	firstHeartbeat := time.Date(2026, 6, 1, 1, 0, 0, 0, time.UTC)
	first, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-host-1",
		RunnerVersion:              "1.0.0",
		CapabilitiesJSON:           []byte(`["agentteam.message.mention"]`),
		LastHeartbeatAt:            &firstHeartbeat,
		Now:                        firstHeartbeat,
	})
	require.NoError(t, err)
	require.Equal(t, RuntimeSessionStatusOnline, first.Status)
	require.Equal(t, "runner-host-1", first.RunnerID)
	require.Equal(t, firstHeartbeat, *first.LastHeartbeatAt)

	secondHeartbeat := firstHeartbeat.Add(5 * time.Minute)
	second, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-host-1",
		RunnerVersion:              "1.1.0",
		CapabilitiesJSON:           []byte(`["agentteam.message.mention","files.write"]`),
		LastHeartbeatAt:            &secondHeartbeat,
		Now:                        secondHeartbeat,
	})
	require.NoError(t, err)
	require.Equal(t, first.ID, second.ID)
	require.Equal(t, "1.1.0", second.RunnerVersion)
	require.Equal(t, secondHeartbeat, *second.LastHeartbeatAt)
	require.JSONEq(t, `["agentteam.message.mention","files.write"]`, string(second.CapabilitiesJSON))

	offline, err := store.SetRuntimeSessionStatus(ctx, fixture.WorkspaceID, second.ID, RuntimeSessionStatusOffline, secondHeartbeat.Add(time.Minute))
	require.NoError(t, err)
	require.Equal(t, RuntimeSessionStatusOffline, offline.Status)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 2, 0, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_attempt", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-attempt")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-attempt-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-attempt",
		RunnerVersion:              "1.0.0",
		Now:                        now,
	})
	require.NoError(t, err)
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	candidates, err := store.ListDeliveryAttemptCandidates(ctx, DeliveryAttemptCandidateQuery{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      "app_mention",
		Now:                   now,
	})
	require.NoError(t, err)
	require.Len(t, candidates, 1)
	require.Equal(t, delivery.ID, candidates[0].ID)

	leaseExpiresAt := now.Add(2 * time.Minute)
	attempt, err := store.CreateDeliveryAttempt(ctx, DeliveryAttemptCreate{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceEventDeliveryID:     delivery.ID,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		LeaseOwner:                 session.ID,
		LeaseExpiresAt:             &leaseExpiresAt,
		Now:                        now,
	})
	require.NoError(t, err)
	require.Equal(t, 1, attempt.AttemptNumber)
	require.Equal(t, DeliveryAttemptStatusLeased, attempt.Status)
	require.Empty(t, attempt.DeliveryTokenHash)
	require.NotNil(t, attempt.ClaimedAt)

	candidates, err = store.ListDeliveryAttemptCandidates(ctx, DeliveryAttemptCandidateQuery{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      "app_mention",
		Now:                   now.Add(time.Second),
	})
	require.NoError(t, err)
	require.Empty(t, candidates)

	secondAttempt, err := store.CreateDeliveryAttempt(ctx, DeliveryAttemptCreate{
		WorkspaceID:            fixture.WorkspaceID,
		ServiceEventDeliveryID: delivery.ID,
		TransportKind:          RuntimeTransportKindRunnerStream,
		LeaseOwner:             session.ID,
		LeaseExpiresAt:         &leaseExpiresAt,
		Now:                    now.Add(time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, 2, secondAttempt.AttemptNumber)
	require.Empty(t, secondAttempt.DeliveryTokenHash)

	statusCode := 202
	delivered, err := store.SetDeliveryAttemptStatus(ctx, DeliveryAttemptStatusUpdate{
		WorkspaceID:        fixture.WorkspaceID,
		ID:                 attempt.ID,
		Status:             DeliveryAttemptStatusDelivered,
		ResponseStatusCode: &statusCode,
		ResponseExcerpt:    "accepted",
		Now:                now.Add(2 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusDelivered, delivered.Status)
	require.NotNil(t, delivered.DeliveredAt)
	require.Equal(t, 202, *delivered.ResponseStatusCode)

	seedRuntimeEffectMessage(t, db, fixture, seed, "msg_runtime_effect_1", now.Add(3*time.Second))
	effect, err := store.AppendDeliveryEffect(ctx, DeliveryEffectAppend{
		WorkspaceID:       fixture.WorkspaceID,
		DeliveryAttemptID: attempt.ID,
		EffectKind:        string(runeffect.EffectKindMessage),
		EffectRef:         "msg_runtime_effect_1",
		TargetKind:        "channel",
		TargetRef:         fixture.ScopeID,
		MessageID:         "msg_runtime_effect_1",
		Metadata:          []byte(`{"phase":"progress"}`),
		Now:               now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, delivery.ID, effect.ServiceEventDeliveryID)
	require.Equal(t, attempt.ID, effect.DeliveryAttemptID)
	require.JSONEq(t, `{"phase":"progress"}`, string(effect.Metadata))

	acked, err := store.SetDeliveryAttemptStatus(ctx, DeliveryAttemptStatusUpdate{
		WorkspaceID: fixture.WorkspaceID,
		ID:          attempt.ID,
		Status:      DeliveryAttemptStatusAcked,
		ResultJSON:  []byte(`{"status":"success"}`),
		Now:         now.Add(4 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, acked.Status)
	require.NotNil(t, acked.AckedAt)
	require.NotNil(t, acked.ResponseStatusCode)
	require.Equal(t, 202, *acked.ResponseStatusCode)
	require.Equal(t, "accepted", acked.ResponseExcerpt)
	require.JSONEq(t, `{"status":"success"}`, string(acked.ResultJSON))

	effects, err := store.ListDeliveryEffectsByDelivery(ctx, DeliveryEffectListQuery{
		WorkspaceID:            fixture.WorkspaceID,
		ServiceEventDeliveryID: delivery.ID,
	})
	require.NoError(t, err)
	require.Len(t, effects, 1)
	require.Equal(t, effect.ID, effects[0].ID)

	var summaryKind, summaryRef, summaryMessage string
	require.NoError(t, db.QueryRow(ctx, `
SELECT effect_kind, effect_ref, COALESCE(message_id, '')
FROM service_event_deliveries
WHERE id = $1
`, delivery.ID).Scan(&summaryKind, &summaryRef, &summaryMessage))
	require.Equal(t, string(runeffect.EffectKindMessage), summaryKind)
	require.Equal(t, "msg_runtime_effect_1", summaryRef)
	require.Equal(t, "msg_runtime_effect_1", summaryMessage)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 2, 30, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_candidate", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-candidate")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	expiredLease := now.Add(-time.Minute)
	_, err := store.CreateDeliveryAttempt(ctx, DeliveryAttemptCreate{
		WorkspaceID:            fixture.WorkspaceID,
		ServiceEventDeliveryID: delivery.ID,
		TransportKind:          RuntimeTransportKindRunnerStream,
		LeaseOwner:             "runner-expired",
		LeaseExpiresAt:         &expiredLease,
		Now:                    now.Add(-2 * time.Minute),
	})
	require.NoError(t, err)

	candidates, err := store.ListDeliveryAttemptCandidates(ctx, DeliveryAttemptCandidateQuery{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      "app_mention",
		Now:                   now,
	})
	require.NoError(t, err)
	require.Len(t, candidates, 1)
	require.Equal(t, delivery.ID, candidates[0].ID)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 4, 0, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_claim", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-claim")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-dispatch-claim-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-dispatch-claim",
		RunnerVersion:              "1.0.0",
		Now:                        now,
	})
	require.NoError(t, err)
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	envelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      seed.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		LeaseDuration:              2 * time.Minute,
		Now:                        now.Add(time.Minute),
	})
	require.NoError(t, err)
	require.Len(t, envelopes, 1)
	envelope := envelopes[0]
	require.Equal(t, envelope.Attempt.ID, envelope.EnvelopeID)
	require.Equal(t, delivery.ID, envelope.Delivery.ID)
	require.Equal(t, "agentteam.message.mention", envelope.EventType)
	require.JSONEq(t, `{"message":{"id":"msg_runtime_source"}}`, string(envelope.EventPayload))
	require.NotEmpty(t, envelope.DeliveryToken)
	require.Equal(t, RuntimeTransportKindRunnerStream, envelope.Attempt.TransportKind)
	require.Equal(t, credential.ID, envelope.Attempt.ServiceRuntimeCredentialID)
	require.Equal(t, session.ID, envelope.Attempt.ServiceRuntimeSessionID)
	require.Equal(t, session.ID, envelope.Attempt.LeaseOwner)
	require.NotNil(t, envelope.Attempt.LeaseExpiresAt)
	require.Equal(t, secrets.HashToken(envelope.DeliveryToken), envelope.Attempt.DeliveryTokenHash)
	require.NotNil(t, envelope.Attempt.DeliveryTokenExpiresAt)
	require.Equal(t, envelope.Attempt.LeaseExpiresAt.UTC(), envelope.Attempt.DeliveryTokenExpiresAt.UTC())
	require.Equal(t, session.ID, envelope.Delivery.LeaseOwner)
	require.NotNil(t, envelope.Delivery.LeaseExpiresAt)

	claimedSession, err := store.LoadRuntimeSession(ctx, fixture.WorkspaceID, session.ID)
	require.NoError(t, err)
	require.NotNil(t, claimedSession.LastClaimAt)
	require.Equal(t, now.Add(time.Minute), claimedSession.LastClaimAt.UTC())

	again, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      seed.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		Now:                        now.Add(90 * time.Second),
	})
	require.NoError(t, err)
	require.Empty(t, again)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 4, 5, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_bot_allowlist", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-app-allowlist")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-app-allowlist-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID: fixture.WorkspaceID, ServiceDefinitionID: seed.DefinitionID,
		ServiceInstallationID: seed.InstallationID, ServiceMemberID: seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID, RunnerID: "runner-app-allowlist", Now: now,
	})
	require.NoError(t, err)

	mention := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	mentionSubscription, err := store.LoadEventSubscription(ctx, fixture.WorkspaceID, mention.EventSubscriptionID)
	require.NoError(t, err)
	var action EventDelivery
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		subscription, _, err := txStore.UpsertEventSubscription(ctx, EventSubscriptionUpsert{
			ID: "svcesub_runtime_action_allowlist", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID: seed.DefinitionID, ServiceInstallationID: seed.InstallationID,
			ServiceMemberID: seed.ServiceMemberID, ServiceConnectionRef: mentionSubscription.ServiceConnectionRef,
			ServiceTargetBindingID: mentionSubscription.ServiceTargetBindingID, ServiceKind: KindApp,
			SubscriptionKind: SubscriptionKindAppNotificationAction,
			TargetKind:       mentionSubscription.TargetKind, TargetRef: mentionSubscription.TargetRef,
			DeliveryScopeID: mentionSubscription.DeliveryScopeID, PermissionCheckScopeID: mentionSubscription.PermissionCheckScopeID,
			OwnerMemberID: fixture.HumanMemberID, AuthorizerMemberID: fixture.HumanMemberID,
			CreatedByActorType: "human", CreatedByActorID: fixture.UserID, CreatedVia: "test",
			ExternalResourceType: "app_notification_action", ExternalResourceKey: "bnaction_allowlist",
			ExternalResourceRef: "btn_allowlist", EventFilterKey: "agentteam.app.notification.action",
			EventFilters: []byte(`{"events":["agentteam.app.notification.action"]}`), DeliveryPolicy: AppRuntimeDeliveryPolicyExternal,
			ActiveSubscriptionKey: "svc-runtime-action-allowlist", Now: now.Add(time.Second),
		})
		if err != nil {
			return err
		}
		action, _, err = txStore.ReserveEventDelivery(ctx, EventDeliveryReserve{
			ID: "svced_runtime_action_allowlist", WorkspaceID: fixture.WorkspaceID,
			EventSubscriptionID: subscription.ID, EventType: "agentteam.app.notification.action",
			EventAction: "approve", EventID: "bnaction_allowlist", EventKey: "btn_allowlist:approve",
			EventPayload:   []byte(`{"notificationId":"btn_allowlist","actionKey":"approve"}`),
			IdempotencyKey: "app_notification_action:bnaction_allowlist", Now: now.Add(time.Second),
		})
		return err
	}))
	seedRuntimeAppAuthority(
		t,
		ctx,
		db,
		store,
		fixture,
		seed,
		"",
		[]string{RuntimeTransportKindRunnerStream},
		[]string{AppRuntimeEventMessageMention, AppRuntimeEventNotificationAction},
		[]string{"app.mentions.read", "messages.write", "app.notification_actions.read"},
		now.Add(2*time.Second),
	)

	envelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKinds: AppRuntimeSubscriptionKinds(), TransportKind: RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID, ServiceRuntimeSessionID: session.ID,
		LeaseDuration: time.Minute, Limit: 10, Now: now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Len(t, envelopes, 2)
	kinds := map[string]bool{}
	ids := map[string]bool{}
	for _, envelope := range envelopes {
		kinds[envelope.Delivery.SubscriptionKind] = true
		ids[envelope.Delivery.ID] = true
	}
	require.True(t, kinds[SubscriptionKindAppMention])
	require.True(t, kinds[SubscriptionKindAppNotificationAction])
	require.True(t, ids[mention.ID])
	require.True(t, ids[action.ID])
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 4, 10, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_same_claim_ts", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-same-claim-ts")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-same-claim-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-same-claim-ts",
		RunnerVersion:              "1.0.0",
		Now:                        now,
	})
	require.NoError(t, err)

	require.NoError(t, markRuntimeSessionClaimed(ctx, db, fixture.WorkspaceID, session.ID, credential.ID, seed.InstallationID, now))
	require.NoError(t, markRuntimeSessionClaimed(ctx, db, fixture.WorkspaceID, session.ID, credential.ID, seed.InstallationID, now))
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 4, 30, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_expire", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-expire")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-dispatch-expire-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-dispatch-expire",
		Now:                        now,
	})
	require.NoError(t, err)
	seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	first, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      seed.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		LeaseDuration:              time.Minute,
		Now:                        now,
	})
	require.NoError(t, err)
	require.Len(t, first, 1)

	second, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      seed.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		LeaseDuration:              time.Minute,
		Now:                        now.Add(2 * time.Minute),
	})
	require.NoError(t, err)
	require.Len(t, second, 1)
	require.Equal(t, 2, second[0].Attempt.AttemptNumber)
	require.NotEqual(t, first[0].Attempt.ID, second[0].Attempt.ID)

	expired, err := store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, first[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusExpired, expired.Status)
	require.NotNil(t, expired.ExpiredAt)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 4, 45, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_stale_lease", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-stale-lease")
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-dispatch-stale-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        seed.DefinitionID,
		ServiceInstallationID:      seed.InstallationID,
		ServiceMemberID:            seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID,
		RunnerID:                   "runner-dispatch-stale",
		Now:                        now,
	})
	require.NoError(t, err)
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	for index := 0; index < 3; index++ {
		expiredLease := now.Add(time.Duration(-3+index) * time.Minute)
		_, err := store.CreateDeliveryAttempt(ctx, DeliveryAttemptCreate{
			WorkspaceID:            fixture.WorkspaceID,
			ServiceEventDeliveryID: delivery.ID,
			TransportKind:          RuntimeTransportKindRunnerStream,
			LeaseOwner:             fmt.Sprintf("stale-runner-%d", index),
			LeaseExpiresAt:         &expiredLease,
			Now:                    expiredLease.Add(-time.Minute),
		})
		require.NoError(t, err)
	}

	claimed, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      seed.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID,
		ServiceRuntimeSessionID:    session.ID,
		LeaseDuration:              5 * time.Minute,
		Limit:                      1,
		Now:                        now,
	})
	require.NoError(t, err)
	require.Len(t, claimed, 1)
	require.Equal(t, 4, claimed[0].Attempt.AttemptNumber)
	require.Equal(t, session.ID, claimed[0].Delivery.LeaseOwner)

	changed, err := store.ExpireDeliveryAttemptLeases(ctx, DeliveryAttemptLeaseExpire{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		Now:                   now.Add(30 * time.Second),
		Limit:                 10,
	})
	require.NoError(t, err)
	require.Zero(t, changed)
	leased, err := loadEventDelivery(ctx, db, delivery.ID)
	require.NoError(t, err)
	require.Equal(t, session.ID, leased.LeaseOwner)
	require.NotNil(t, leased.LeaseExpiresAt)
	require.Equal(t, claimed[0].Attempt.LeaseExpiresAt.UTC(), leased.LeaseExpiresAt.UTC())
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 0, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_retry", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-retry")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	claimed, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		TransportKind:         RuntimeTransportKindCallback,
		LeaseOwner:            "callback-dispatcher",
		LeaseDuration:         2 * time.Minute,
		Now:                   now,
		CallbackURLSnapshot:   "https://app.example.test/events",
	})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	statusCode := 503
	retry, err := store.ScheduleDeliveryAttemptRetry(ctx, DeliveryAttemptRetrySchedule{
		WorkspaceID:    fixture.WorkspaceID,
		AttemptID:      claimed[0].Attempt.ID,
		ErrorCode:      "callback_unavailable",
		ErrorDetail:    "upstream unavailable",
		ResponseStatus: &statusCode,
		ResponseBody:   "service unavailable",
		BaseBackoff:    time.Minute,
		MaxBackoff:     5 * time.Minute,
		Now:            now.Add(10 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusFailed, retry.Attempt.Status)
	require.Equal(t, "callback_unavailable", retry.Attempt.ErrorCode)
	require.Equal(t, now.Add(70*time.Second), retry.NextRetryAt)
	require.Equal(t, "retry_waiting", retry.Delivery.Status)
	require.NotNil(t, retry.Delivery.NextRetryAt)
	require.Equal(t, retry.NextRetryAt, retry.Delivery.NextRetryAt.UTC())

	blocked, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		TransportKind:         RuntimeTransportKindCallback,
		LeaseOwner:            "callback-dispatcher",
		Now:                   now.Add(30 * time.Second),
	})
	require.NoError(t, err)
	require.Empty(t, blocked)

	due, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		TransportKind:         RuntimeTransportKindCallback,
		LeaseOwner:            "callback-dispatcher",
		Now:                   retry.NextRetryAt.Add(time.Second),
	})
	require.NoError(t, err)
	require.Len(t, due, 1)
	require.Equal(t, delivery.ID, due[0].Delivery.ID)
	require.Equal(t, 2, due[0].Attempt.AttemptNumber)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 15, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_retry_terminal", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-retry-terminal")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	claimed, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		TransportKind:         RuntimeTransportKindCallback,
		LeaseOwner:            "callback-dispatcher",
		LeaseDuration:         2 * time.Minute,
		Now:                   now,
		CallbackURLSnapshot:   "https://app.example.test/events",
	})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	acked, err := store.SetDeliveryAttemptStatus(ctx, DeliveryAttemptStatusUpdate{
		WorkspaceID: fixture.WorkspaceID,
		ID:          claimed[0].Attempt.ID,
		Status:      DeliveryAttemptStatusAcked,
		Now:         now.Add(10 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, acked.Status)

	_, err = store.ScheduleDeliveryAttemptRetry(ctx, DeliveryAttemptRetrySchedule{
		WorkspaceID: fixture.WorkspaceID,
		AttemptID:   claimed[0].Attempt.ID,
		ErrorCode:   "late_failure",
		Now:         now.Add(20 * time.Second),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "not retryable")

	currentAttempt, err := store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, claimed[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, currentAttempt.Status)
	currentDelivery, err := loadEventDelivery(ctx, db, delivery.ID)
	require.NoError(t, err)
	require.NotEqual(t, "retry_waiting", currentDelivery.Status)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 45, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_delivered_current", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-delivered-current")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)

	claimed, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceInstallationID: seed.InstallationID,
		SubscriptionKind:      SubscriptionKindAppMention,
		TransportKind:         RuntimeTransportKindCallback,
		LeaseOwner:            "callback-dispatcher",
		LeaseDuration:         2 * time.Minute,
		Now:                   now,
		CallbackURLSnapshot:   "https://app.example.test/events",
	})
	require.NoError(t, err)
	require.Len(t, claimed, 1)

	mustExec(t, db, `
UPDATE service_event_deliveries
SET lease_owner = 'another-callback-dispatcher'
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, delivery.ID)
	_, err = store.MarkDeliveryAttemptDelivered(ctx, DeliveryAttemptDeliveredMark{
		WorkspaceID: fixture.WorkspaceID,
		AttemptID:   claimed[0].Attempt.ID,
		Now:         now.Add(5 * time.Second),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "no longer current")
	currentAttempt, err := store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, claimed[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusLeased, currentAttempt.Status)

	require.NotNil(t, claimed[0].Attempt.LeaseExpiresAt)
	mustExec(t, db, `
UPDATE service_event_deliveries
SET lease_owner = $3, lease_expires_at = $4
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, delivery.ID, claimed[0].Attempt.LeaseOwner, *claimed[0].Attempt.LeaseExpiresAt)
	statusCode := 202
	delivered, err := store.MarkDeliveryAttemptDelivered(ctx, DeliveryAttemptDeliveredMark{
		WorkspaceID:        fixture.WorkspaceID,
		AttemptID:          claimed[0].Attempt.ID,
		ResponseStatusCode: &statusCode,
		ResponseExcerpt:    "accepted",
		Now:                now.Add(10 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusDelivered, delivered.Status)
	require.NotNil(t, delivered.DeliveredAt)
	require.Equal(t, 202, *delivered.ResponseStatusCode)
	require.Equal(t, "accepted", delivered.ResponseExcerpt)

	acked, err := store.SetDeliveryAttemptStatus(ctx, DeliveryAttemptStatusUpdate{
		WorkspaceID: fixture.WorkspaceID,
		ID:          claimed[0].Attempt.ID,
		Status:      DeliveryAttemptStatusAcked,
		Now:         now.Add(20 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, acked.Status)
	_, err = store.MarkDeliveryAttemptDelivered(ctx, DeliveryAttemptDeliveredMark{
		WorkspaceID: fixture.WorkspaceID,
		AttemptID:   claimed[0].Attempt.ID,
		Now:         now.Add(30 * time.Second),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "not leased")
	currentAttempt, err = store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, claimed[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, currentAttempt.Status)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 55, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_callback_config", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-callback-config")
	mustExec(t, db, `
UPDATE service_definitions
SET config = $2
WHERE id = $1
`, seed.DefinitionID, []byte(`{"callbackSigningSecret":"secret"}`))
	mustExec(t, db, `
UPDATE service_installations
SET metadata = $2
WHERE id = $1
`, seed.InstallationID, []byte(`{"callbackURL":"https://app.example.com/callback"}`))

	_, err := store.LoadRuntimeCallbackConfig(ctx, fixture.WorkspaceID, seed.InstallationID)
	require.Error(t, err)
	require.Contains(t, err.Error(), "not found")

	_, err = store.UpsertCapability(ctx, CapabilityUpsert{
		ServiceDefinitionID: seed.DefinitionID,
		ServiceKind:         KindApp,
		CapabilityKey:       CapabilityApp,
		Config:              []byte(`{"runtime":{"transports":["callback","runner_stream"]},"events":["agentteam.message.mention"]}`),
		DefaultEnabled:      true,
		Now:                 now,
	})
	require.NoError(t, err)
	_, err = store.LoadRuntimeCallbackConfig(ctx, fixture.WorkspaceID, seed.InstallationID)
	require.Error(t, err)
	require.Contains(t, err.Error(), "not found", "latest definition capability must not upgrade the pinned installation")

	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "https://app.example.com/callback", []string{RuntimeTransportKindRunnerStream}, []string{"agentteam.message.mention"})
	_, err = store.LoadRuntimeCallbackConfig(ctx, fixture.WorkspaceID, seed.InstallationID)
	require.Error(t, err)
	require.Contains(t, err.Error(), "callback transport")

	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "https://app.example.com/callback", []string{RuntimeTransportKindCallback, RuntimeTransportKindRunnerStream}, []string{"agentteam.message.mention"})
	config, err := store.LoadRuntimeCallbackConfig(ctx, fixture.WorkspaceID, seed.InstallationID)
	require.NoError(t, err)
	require.Equal(t, "https://app.example.com/callback", config.CallbackURL)
	require.Equal(t, "secret", config.CallbackSigningSecret)
	require.Equal(t, []string{RuntimeTransportKindCallback, RuntimeTransportKindRunnerStream}, config.AppRuntimeTransports)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 58, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_callback_grants", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-callback-grants")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	mustExec(t, db, `UPDATE service_definitions SET config = $2 WHERE id = $1`, seed.DefinitionID, []byte(`{"callbackSigningSecret":"secret"}`))
	seedRuntimeAppAuthority(
		t,
		ctx,
		db,
		store,
		fixture,
		seed,
		"https://app.example.com/callback",
		[]string{RuntimeTransportKindCallback},
		[]string{AppRuntimeEventMessageMention, AppRuntimeEventNotificationAction},
		nil,
		now,
	)

	listTargets := func(kind string, at time.Time) []RuntimeCallbackDispatchTarget {
		targets, err := store.ListRuntimeCallbackDispatchTargets(ctx, RuntimeCallbackDispatchTargetList{
			ServiceKind: KindApp, SubscriptionKind: kind, Now: at, Limit: 10,
		})
		require.NoError(t, err)
		return targets
	}
	require.Empty(t, listTargets(SubscriptionKindAppMention, now.Add(time.Second)))

	setGrants := func(scopes ...string) {
		require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
			txStore := store.InTx(tx)
			installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
			if err != nil {
				return err
			}
			return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, scopes, now.Add(2*time.Second))
		}))
	}
	setGrants("app.mentions.read")
	require.Empty(t, listTargets(SubscriptionKindAppMention, now.Add(3*time.Second)))
	setGrants("app.mentions.read", "messages.write")
	require.Len(t, listTargets(SubscriptionKindAppMention, now.Add(4*time.Second)), 1)

	mustExec(t, db, `
UPDATE service_event_subscriptions SET subscription_kind = $2, updated_at = $3 WHERE id = $1
`, delivery.EventSubscriptionID, SubscriptionKindAppNotificationAction, now.Add(5*time.Second))
	mustExec(t, db, `
UPDATE service_event_deliveries SET subscription_kind = $2, event_type = 'agentteam.app.notification.action', updated_at = $3 WHERE id = $1
`, delivery.ID, SubscriptionKindAppNotificationAction, now.Add(5*time.Second))
	require.Empty(t, listTargets(SubscriptionKindAppNotificationAction, now.Add(6*time.Second)))
	setGrants("app.notification_actions.read")
	require.Len(t, listTargets(SubscriptionKindAppNotificationAction, now.Add(7*time.Second)), 1)

	mustExec(t, db, `
UPDATE service_grants SET expires_at = $2, updated_at = $3
WHERE service_installation_id = $1 AND scope = 'app.notification_actions.read' AND revoked_at IS NULL
`, seed.InstallationID, now.Add(7*time.Second), now.Add(8*time.Second))
	require.Empty(t, listTargets(SubscriptionKindAppNotificationAction, now.Add(9*time.Second)))
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 59, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_pinned_event_authority", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-pinned-event-authority")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	credential := seedRuntimeCredential(t, ctx, store, fixture, seed, "runtime-pinned-event-authority-token")
	session, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID: fixture.WorkspaceID, ServiceDefinitionID: seed.DefinitionID,
		ServiceInstallationID: seed.InstallationID, ServiceMemberID: seed.ServiceMemberID,
		ServiceRuntimeCredentialID: credential.ID, RunnerID: "runner-pinned-event-authority",
		RunnerVersion: "1.0.0", Now: now,
	})
	require.NoError(t, err)
	mustExec(t, db, `UPDATE service_definitions SET config = $2 WHERE id = $1`, seed.DefinitionID, []byte(`{"callbackSigningSecret":"secret"}`))
	// Legacy rows deliberately have the relevant grants and active substrate,
	// but the immutable installation manifest declares neither App event.
	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "https://app.example.com/callback", []string{RuntimeTransportKindCallback, RuntimeTransportKindRunnerStream}, []string{})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
		if err != nil {
			return err
		}
		return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, []string{
			"app.mentions.read", "messages.write", "app.notification_actions.read",
		}, now.Add(time.Second))
	}))
	// Make the forged legacy delivery explicitly ready for every dispatcher.
	// The only reason it must remain unclaimable is missing pinned event
	// authority, not status, retry timing, or a pre-existing lease.
	mustExec(t, db, `
UPDATE service_event_deliveries
SET status = 'pending',
    next_retry_at = NULL,
    lease_owner = '',
    lease_expires_at = NULL,
    updated_at = $2
WHERE id = $1
`, delivery.ID, now.Add(time.Second))
	requireServiceActorSingleInt(t, ctx, db, 1, `
SELECT COUNT(*)
FROM service_event_deliveries
WHERE id = $1
  AND status = 'pending'
  AND next_retry_at IS NULL
  AND COALESCE(lease_owner, '') = ''
  AND lease_expires_at IS NULL
`, delivery.ID)
	requireServiceActorSingleInt(t, ctx, db, 0, `SELECT COUNT(*) FROM service_event_delivery_attempts WHERE service_event_delivery_id = $1`, delivery.ID)

	callbackTargets, err := store.ListRuntimeCallbackDispatchTargets(ctx, RuntimeCallbackDispatchTargetList{
		ServiceKind: KindApp, SubscriptionKind: SubscriptionKindAppMention,
		Now: now.Add(2 * time.Second), Limit: 10,
	})
	require.NoError(t, err)
	require.Empty(t, callbackTargets, "callback scan must require the pinned mention event")
	callbackEnvelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppMention, TransportKind: RuntimeTransportKindCallback,
		LeaseOwner: "callback:test:pinned-event-authority", Now: now.Add(2 * time.Second), Limit: 10,
	})
	require.NoError(t, err)
	require.Empty(t, callbackEnvelopes, "callback final claim must require the pinned mention event")
	runnerEnvelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppMention, TransportKind: RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: credential.ID, ServiceRuntimeSessionID: session.ID,
		Now: now.Add(2 * time.Second), Limit: 10,
	})
	require.NoError(t, err)
	require.Empty(t, runnerEnvelopes, "runner final claim must require the pinned mention event")

	handlerRef, err := InternalHandlerConnectionRef("agentteam.approval")
	require.NoError(t, err)
	mustExec(t, db, `
UPDATE service_event_subscriptions
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4,
    event_filter_key = $5, updated_at = $6
WHERE id = $1
`, delivery.EventSubscriptionID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, AppRuntimeEventNotificationAction, now.Add(3*time.Second))
	mustExec(t, db, `
UPDATE service_event_deliveries
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4,
    event_type = $5, updated_at = $6
WHERE id = $1
`, delivery.ID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, AppRuntimeEventNotificationAction, now.Add(3*time.Second))

	internalTargets, err := store.ListRuntimeInternalDispatchTargets(ctx, RuntimeInternalDispatchTargetList{Now: now.Add(4 * time.Second), Limit: 10})
	require.NoError(t, err)
	require.Empty(t, internalTargets, "internal-handler scan must require the pinned notification-action event")
	internalEnvelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppNotificationAction,
		DeliveryPolicy:   AppRuntimeDeliveryPolicyInternalHandler, ServiceConnectionRef: handlerRef,
		TransportKind: RuntimeTransportKindInternalHandler, LeaseOwner: "internal:test:agentteam.approval",
		Now: now.Add(4 * time.Second), Limit: 10,
	})
	require.NoError(t, err)
	require.Empty(t, internalEnvelopes, "internal-handler final claim must require the pinned notification-action event")
	requireServiceActorSingleInt(t, ctx, db, 0, `SELECT COUNT(*) FROM service_event_delivery_attempts WHERE service_event_delivery_id = $1`, delivery.ID)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 6, 0, 0, 0, time.UTC)
	store := New(db)

	seedTarget := func(suffix string, slug string, workspaceGroupID string, deliveryAt time.Time) (serviceActorFixture, runtimeServiceActorSeed) {
		fixture := seedServiceActorFixture(t, db, suffix, deliveryAt)
		seed := seedRuntimeServiceActor(t, ctx, store, fixture, slug)
		seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, deliveryAt)
		mustExec(t, db, `
INSERT INTO workspace_placements (workspace_id, workspace_group_id, state, created_at, updated_at)
VALUES ($1, $2, 'active', $3, $3)
`, fixture.WorkspaceID, workspaceGroupID, deliveryAt)
		mustExec(t, db, `UPDATE service_definitions SET config = $2 WHERE id = $1`, seed.DefinitionID, []byte(`{"callbackSigningSecret":"secret"}`))
		seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "https://app.example.com/callback", []string{RuntimeTransportKindCallback}, []string{"agentteam.message.mention"})
		require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
			txStore := store.InTx(tx)
			installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
			if err != nil {
				return err
			}
			return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, []string{"app.mentions.read", "messages.write"}, deliveryAt)
		}))
		return fixture, seed
	}

	olderFixture, olderSeed := seedTarget("runtime_callback_group_a", "runtime-callback-group-a", "callback-group-a", now)
	newerFixture, newerSeed := seedTarget("runtime_callback_group_b", "runtime-callback-group-b", "callback-group-b", now.Add(time.Second))

	listTargets := func(workspaceGroupID string) []RuntimeCallbackDispatchTarget {
		targets, err := store.ListRuntimeCallbackDispatchTargets(ctx, RuntimeCallbackDispatchTargetList{
			WorkspaceGroupID: workspaceGroupID,
			ServiceKind:      KindApp,
			SubscriptionKind: SubscriptionKindAppMention,
			Now:              now.Add(time.Minute),
			Limit:            1,
		})
		require.NoError(t, err)
		return targets
	}

	require.Equal(t, []RuntimeCallbackDispatchTarget{{
		WorkspaceID:           olderFixture.WorkspaceID,
		ServiceInstallationID: olderSeed.InstallationID,
	}}, listTargets(""))
	require.Equal(t, []RuntimeCallbackDispatchTarget{{
		WorkspaceID:           newerFixture.WorkspaceID,
		ServiceInstallationID: newerSeed.InstallationID,
	}}, listTargets(" callback-group-b "))
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 5, 30, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_dispatch_identity", now)
	store := New(db)
	first := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-dispatch-identity-a")
	secondFixture := fixture
	secondFixture.ServiceMemberID = "mem_svcactor_service_runtime_dispatch_identity_b"
	mustExec(t, db, `
INSERT INTO members (id, workspace_id, type, display_name, workspace_role, status, created_at, updated_at)
VALUES ($1, $2, 'service', 'Service B', NULL, 'active', $3, $3)
`, secondFixture.ServiceMemberID, secondFixture.WorkspaceID, now)
	second := seedRuntimeServiceActor(t, ctx, store, secondFixture, "runtime-dispatch-identity-b")
	secondCredential := seedRuntimeCredential(t, ctx, store, secondFixture, second, "runtime-identity-token-b")
	secondSession, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        second.DefinitionID,
		ServiceInstallationID:      second.InstallationID,
		ServiceMemberID:            second.ServiceMemberID,
		ServiceRuntimeCredentialID: secondCredential.ID,
		RunnerID:                   "runner-identity-b",
		Now:                        now,
	})
	require.NoError(t, err)
	seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, first, now)

	_, err = store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      first.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: secondCredential.ID,
		ServiceRuntimeSessionID:    secondSession.ID,
		Now:                        now.Add(time.Minute),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "credential installation")

	firstCredential := seedRuntimeCredential(t, ctx, store, fixture, first, "runtime-identity-token-a")
	firstSession, err := store.UpsertRuntimeSession(ctx, RuntimeSessionUpsert{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceDefinitionID:        first.DefinitionID,
		ServiceInstallationID:      first.InstallationID,
		ServiceMemberID:            first.ServiceMemberID,
		ServiceRuntimeCredentialID: firstCredential.ID,
		RunnerID:                   "runner-identity-a",
		Now:                        now,
	})
	require.NoError(t, err)
	_, err = store.SetRuntimeSessionStatus(ctx, fixture.WorkspaceID, firstSession.ID, RuntimeSessionStatusOffline, now.Add(2*time.Minute))
	require.NoError(t, err)
	_, err = store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID:                fixture.WorkspaceID,
		ServiceInstallationID:      first.InstallationID,
		SubscriptionKind:           SubscriptionKindAppMention,
		TransportKind:              RuntimeTransportKindRunnerStream,
		ServiceRuntimeCredentialID: firstCredential.ID,
		ServiceRuntimeSessionID:    firstSession.ID,
		Now:                        now.Add(3 * time.Minute),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "session must be online")
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 6, 1, 3, 0, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_retention", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-retention")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	attempt, err := store.CreateDeliveryAttempt(ctx, DeliveryAttemptCreate{
		WorkspaceID:            fixture.WorkspaceID,
		ServiceEventDeliveryID: delivery.ID,
		TransportKind:          RuntimeTransportKindCallback,
		AttemptNumber:          1,
		Status:                 DeliveryAttemptStatusFailed,
		Now:                    now.Add(-2 * time.Hour),
	})
	require.NoError(t, err)
	failedAt := now.Add(-time.Hour)
	_, err = store.SetDeliveryAttemptStatus(ctx, DeliveryAttemptStatusUpdate{
		WorkspaceID: fixture.WorkspaceID,
		ID:          attempt.ID,
		Status:      DeliveryAttemptStatusFailed,
		ErrorCode:   "callback_failed",
		ErrorDetail: "callback failed",
		Now:         failedAt,
	})
	require.NoError(t, err)
	effect, err := store.AppendDeliveryEffect(ctx, DeliveryEffectAppend{
		WorkspaceID:       fixture.WorkspaceID,
		DeliveryAttemptID: attempt.ID,
		EffectKind:        string(runeffect.EffectKindTaskDiscussion),
		EffectRef:         "task_entry_runtime_retention",
		TargetKind:        "task",
		TargetRef:         "tsk_runtime_retention",
		Status:            DeliveryEffectStatusFailed,
		Now:               failedAt,
	})
	require.NoError(t, err)
	_, err = db.Exec(ctx, `UPDATE service_event_delivery_effects SET updated_at = $2 WHERE id = $1`, effect.ID, failedAt)
	require.NoError(t, err)

	deletedEffects, err := store.DeleteDeliveryEffectsBefore(ctx, RuntimeRetentionFilter{Status: DeliveryEffectStatusFailed, Cutoff: now, Limit: 1})
	require.NoError(t, err)
	require.Equal(t, int64(1), deletedEffects)
	deletedAttempts, err := store.DeleteDeliveryAttemptsBefore(ctx, RuntimeRetentionFilter{Status: DeliveryAttemptStatusFailed, Cutoff: now, Limit: 1})
	require.NoError(t, err)
	require.Equal(t, int64(1), deletedAttempts)

	requireServiceActorSingleInt(t, ctx, db, 0, `SELECT COUNT(*) FROM service_event_delivery_effects WHERE id = $1`, effect.ID)
	requireServiceActorSingleInt(t, ctx, db, 0, `SELECT COUNT(*) FROM service_event_delivery_attempts WHERE id = $1`, attempt.ID)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 7, 10, 10, 0, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_internal_handler", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-internal-handler")
	delivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	handlerRef, err := InternalHandlerConnectionRef("agentteam.approval")
	require.NoError(t, err)
	mustExec(t, db, `
UPDATE service_event_subscriptions
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4, updated_at = $5
WHERE id = $1
`, delivery.EventSubscriptionID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, now.Add(time.Second))
	mustExec(t, db, `
UPDATE service_event_deliveries
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4,
    event_type = 'agentteam.app.notification.action', updated_at = $5
WHERE id = $1
`, delivery.ID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, now.Add(time.Second))
	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "", nil, []string{AppRuntimeEventMessageMention, AppRuntimeEventNotificationAction})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
		if err != nil {
			return err
		}
		return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, []string{"app.notification_actions.read"}, now.Add(time.Second))
	}))
	targets, err := store.ListRuntimeInternalDispatchTargets(ctx, RuntimeInternalDispatchTargetList{Now: now.Add(2 * time.Second), Limit: 10})
	require.NoError(t, err)
	require.Equal(t, []RuntimeInternalDispatchTarget{{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		HandlerKey: "agentteam.approval", ServiceConnectionRef: handlerRef,
	}}, targets)

	external, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppNotificationAction,
		TransportKind:    RuntimeTransportKindCallback, LeaseOwner: "callback:test", Now: now.Add(2 * time.Second),
	})
	require.NoError(t, err)
	require.Empty(t, external, "external callback transport must not claim an internal route")

	envelopes, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppNotificationAction,
		DeliveryPolicy:   AppRuntimeDeliveryPolicyInternalHandler, ServiceConnectionRef: handlerRef,
		TransportKind: RuntimeTransportKindInternalHandler, LeaseOwner: "internal:test:agentteam.approval",
		LeaseDuration: time.Minute, Now: now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Len(t, envelopes, 1)
	envelope := envelopes[0]
	require.Empty(t, envelope.DeliveryToken)
	require.Equal(t, RuntimeTransportKindInternalHandler, envelope.Attempt.TransportKind)
	require.Empty(t, envelope.Attempt.ServiceRuntimeCredentialID)
	require.Empty(t, envelope.Attempt.ServiceRuntimeSessionID)
	require.Empty(t, envelope.Attempt.DeliveryTokenHash)
	require.Nil(t, envelope.Attempt.DeliveryTokenExpiresAt)

	completed, err := store.CompleteInternalHandlerDelivery(ctx, InternalHandlerDeliveryCompletion{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID, HandlerKey: "agentteam.approval",
		Effect:     runeffect.TaskDiscussionEffect("task_entry_internal_handler"),
		ResultJSON: []byte(`{"decision":"approved","sequence":9007199254740992}`), Now: now.Add(4 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusAcked, completed.Attempt.Status)
	require.Equal(t, string(runeffect.StatusSucceeded), completed.Delivery.Status)
	require.Equal(t, string(runeffect.EffectKindTaskDiscussion), completed.Effect.EffectKind)
	require.Equal(t, "task_entry_internal_handler", completed.Effect.EffectRef)
	replayed, err := store.CompleteInternalHandlerDelivery(ctx, InternalHandlerDeliveryCompletion{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID, HandlerKey: "agentteam.approval",
		Effect:     runeffect.TaskDiscussionEffect("task_entry_internal_handler"),
		ResultJSON: []byte(`{"decision":"approved","sequence":9007199254740993}`), Now: now.Add(5 * time.Second),
	})
	require.Error(t, err)
	require.Empty(t, replayed.Effect.ID)
	replayed, err = store.CompleteInternalHandlerDelivery(ctx, InternalHandlerDeliveryCompletion{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID, HandlerKey: "agentteam.approval",
		Effect:     runeffect.TaskDiscussionEffect("task_entry_internal_handler"),
		ResultJSON: []byte(`{"sequence":9007199254740992,"decision":"approved"}`), Now: now.Add(5 * time.Second),
	})
	require.NoError(t, err)
	require.Equal(t, completed.Effect.ID, replayed.Effect.ID)
	_, err = store.CompleteInternalHandlerDelivery(ctx, InternalHandlerDeliveryCompletion{
		WorkspaceID: fixture.WorkspaceID, AttemptID: envelope.Attempt.ID, HandlerKey: "agentteam.approval",
		Effect:     runeffect.TaskDiscussionEffect("another_decision"),
		ResultJSON: []byte(`{"decision":"approved"}`), Now: now.Add(6 * time.Second),
	})
	require.Error(t, err)
}

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

	ctx := context.Background()
	db := startServiceActorRollbackDB(t, ctx)
	now := time.Date(2026, 7, 10, 10, 30, 0, 0, time.UTC)
	fixture := seedServiceActorFixture(t, db, "runtime_internal_expiry", now)
	store := New(db)
	seed := seedRuntimeServiceActor(t, ctx, store, fixture, "runtime-internal-expiry")
	internalDelivery := seedRuntimeServiceEventDelivery(t, ctx, db, store, fixture, seed, now)
	baseSubscription, err := store.LoadEventSubscription(ctx, fixture.WorkspaceID, internalDelivery.EventSubscriptionID)
	require.NoError(t, err)
	var externalDelivery EventDelivery
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		subscription, _, err := txStore.UpsertEventSubscription(ctx, EventSubscriptionUpsert{
			ID: "svcesub_runtime_external_expiry", WorkspaceID: fixture.WorkspaceID,
			ServiceDefinitionID: seed.DefinitionID, ServiceInstallationID: seed.InstallationID,
			ServiceMemberID: seed.ServiceMemberID, ServiceConnectionRef: AppRuntimeConnectionRefExternal,
			ServiceTargetBindingID: baseSubscription.ServiceTargetBindingID, ServiceKind: KindApp,
			SubscriptionKind: SubscriptionKindAppMention,
			TargetKind:       baseSubscription.TargetKind, TargetRef: baseSubscription.TargetRef,
			DeliveryScopeID: baseSubscription.DeliveryScopeID, PermissionCheckScopeID: baseSubscription.PermissionCheckScopeID,
			OwnerMemberID: fixture.HumanMemberID, AuthorizerMemberID: fixture.HumanMemberID,
			CreatedByActorType: "human", CreatedByActorID: fixture.UserID, CreatedVia: "test",
			ExternalResourceType: "agentteam_message_mention", ExternalResourceKey: "msg_runtime_external_expiry",
			ExternalResourceRef: "msg_runtime_external_expiry", EventFilterKey: "external-expiry",
			EventFilters: []byte(`{"events":["agentteam.message.mention"]}`), DeliveryPolicy: AppRuntimeDeliveryPolicyExternal,
			ActiveSubscriptionKey: "svc-runtime-external-expiry", Now: now.Add(time.Second),
		})
		if err != nil {
			return err
		}
		externalDelivery, _, err = txStore.ReserveEventDelivery(ctx, EventDeliveryReserve{
			ID: "svced_runtime_external_expiry", WorkspaceID: fixture.WorkspaceID,
			EventSubscriptionID: subscription.ID, EventType: "agentteam.message.mention", EventAction: "mention",
			EventID: "msg_runtime_external_expiry", EventKey: "msg_runtime_external_expiry",
			EventPayload:   []byte(`{"message":{"id":"msg_runtime_external_expiry"}}`),
			IdempotencyKey: "app_mention:msg_runtime_external_expiry", Now: now.Add(time.Second),
		})
		return err
	}))
	handlerRef, err := InternalHandlerConnectionRef("agentteam.task")
	require.NoError(t, err)
	mustExec(t, db, `
UPDATE service_event_subscriptions
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4, updated_at = $5
WHERE id = $1
`, internalDelivery.EventSubscriptionID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, now.Add(2*time.Second))
	mustExec(t, db, `
UPDATE service_event_deliveries
SET subscription_kind = $2, delivery_policy = $3, service_connection_ref = $4,
    event_type = 'agentteam.app.notification.action', updated_at = $5
WHERE id = $1
`, internalDelivery.ID, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, handlerRef, now.Add(2*time.Second))
	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, "", nil, []string{AppRuntimeEventMessageMention, AppRuntimeEventNotificationAction})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
		if err != nil {
			return err
		}
		return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, []string{"app.mentions.read", "messages.write", "app.notification_actions.read"}, now.Add(2*time.Second))
	}))

	internal, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppNotificationAction,
		DeliveryPolicy:   AppRuntimeDeliveryPolicyInternalHandler, ServiceConnectionRef: handlerRef,
		TransportKind: RuntimeTransportKindInternalHandler, LeaseOwner: "internal:test:agentteam.task",
		LeaseDuration: time.Minute, Now: now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Len(t, internal, 1)

	external, err := store.ClaimDeliveryEnvelopes(ctx, DeliveryDispatchClaim{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppMention,
		TransportKind:    RuntimeTransportKindCallback, LeaseOwner: "callback:test",
		LeaseDuration: time.Minute, Now: now.Add(3 * time.Second),
	})
	require.NoError(t, err)
	require.Len(t, external, 1)
	require.Equal(t, externalDelivery.ID, external[0].Delivery.ID)

	changed, err := store.ExpireDeliveryAttemptLeases(ctx, DeliveryAttemptLeaseExpire{
		WorkspaceID: fixture.WorkspaceID, ServiceInstallationID: seed.InstallationID,
		SubscriptionKind: SubscriptionKindAppNotificationAction,
		DeliveryPolicy:   AppRuntimeDeliveryPolicyInternalHandler, ServiceConnectionRef: handlerRef,
		Now: now.Add(2 * time.Minute), Limit: 10,
	})
	require.NoError(t, err)
	require.Equal(t, int64(1), changed)
	internalAttempt, err := store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, internal[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusExpired, internalAttempt.Status)
	externalAttempt, err := store.LoadDeliveryAttempt(ctx, fixture.WorkspaceID, external[0].Attempt.ID)
	require.NoError(t, err)
	require.Equal(t, DeliveryAttemptStatusLeased, externalAttempt.Status)
}

type runtimeServiceActorSeed struct {
	DefinitionID    string
	InstallationID  string
	ServiceMemberID string
}

func seedRuntimePinnedAppManifest(t testing.TB, db coredb.QueryExecer, installationID string, callbackURL string, transports []string, events []string) {
	t.Helper()
	metadata, err := json.Marshal(map[string]any{
		"callbackURL": callbackURL,
		"manifest": map[string]any{
			"capabilities": map[string]any{
				"runtime": map[string]any{
					"events": events,
					"runtime": map[string]any{
						"transports": transports,
					},
				},
			},
		},
	})
	require.NoError(t, err)
	mustExec(t, db, `UPDATE service_installations SET metadata = $2 WHERE id = $1`, installationID, metadata)
}

func seedRuntimeAppAuthority(t testing.TB, ctx context.Context, db coredb.DB, store *Store, fixture serviceActorFixture, seed runtimeServiceActorSeed, callbackURL string, transports []string, events []string, scopes []string, now time.Time) {
	t.Helper()
	seedRuntimePinnedAppManifest(t, db, seed.InstallationID, callbackURL, transports, events)
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		installation, err := txStore.LoadInstallationForUpdate(ctx, seed.InstallationID)
		if err != nil {
			return err
		}
		return txStore.SyncGrantScopes(ctx, installation, fixture.HumanMemberID, scopes, now)
	}))
}

func seedRuntimeServiceActor(t testing.TB, ctx context.Context, store *Store, fixture serviceActorFixture, slug string) runtimeServiceActorSeed {
	t.Helper()
	definition, err := store.UpsertDefinition(ctx, DefinitionUpsert{
		ID:          "svc_runtime_" + slug,
		ServiceKind: "app",
		Slug:        slug,
		Name:        slug,
		Status:      StatusActive,
	})
	require.NoError(t, err)
	installation, err := store.UpsertInstallation(ctx, InstallationUpsert{
		ID:                  "svcinst_runtime_" + slug,
		WorkspaceID:         fixture.WorkspaceID,
		ServiceDefinitionID: definition.ID,
		ServiceKind:         definition.ServiceKind,
		Status:              StatusActive,
		CreatedByMemberID:   fixture.HumanMemberID,
	})
	require.NoError(t, err)
	member, err := store.UpsertMember(ctx, MemberUpsert{
		ID:                    "svcmem_runtime_" + slug,
		WorkspaceID:           fixture.WorkspaceID,
		ServiceDefinitionID:   definition.ID,
		ServiceInstallationID: installation.ID,
		MemberID:              fixture.ServiceMemberID,
		ServiceKind:           definition.ServiceKind,
		Handle:                slug,
		DisplayName:           slug,
		Status:                StatusActive,
	})
	require.NoError(t, err)
	return runtimeServiceActorSeed{DefinitionID: definition.ID, InstallationID: installation.ID, ServiceMemberID: member.ID}
}

func seedRuntimeCredential(t testing.TB, ctx context.Context, store *Store, fixture serviceActorFixture, seed runtimeServiceActorSeed, token string) RuntimeCredential {
	t.Helper()
	credential, err := store.CreateRuntimeCredential(ctx, RuntimeCredentialCreate{
		WorkspaceID:           fixture.WorkspaceID,
		ServiceDefinitionID:   seed.DefinitionID,
		ServiceInstallationID: seed.InstallationID,
		ServiceMemberID:       seed.ServiceMemberID,
		ServiceKind:           "app",
		CredentialKind:        RuntimeCredentialKindRunnerStreamToken,
		TransportKind:         RuntimeTransportKindRunnerStream,
		TokenHash:             secrets.HashToken(token),
		TokenHint:             token[:8],
		CreatedByMemberID:     fixture.HumanMemberID,
	})
	require.NoError(t, err)
	return credential
}

func seedRuntimeServiceEventDelivery(t testing.TB, ctx context.Context, db coredb.DB, store *Store, fixture serviceActorFixture, seed runtimeServiceActorSeed, now time.Time) EventDelivery {
	t.Helper()
	var delivery EventDelivery
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		txStore := store.InTx(tx)
		binding, err := txStore.UpsertTargetBinding(ctx, TargetBindingUpsert{
			ID:                       "svctb_runtime_" + stringsFromTime(now),
			WorkspaceID:              fixture.WorkspaceID,
			ServiceDefinitionID:      seed.DefinitionID,
			ServiceInstallationID:    seed.InstallationID,
			ServiceMemberID:          seed.ServiceMemberID,
			ServiceKind:              "app",
			TargetKind:               "channel",
			TargetRef:                fixture.ScopeID,
			DeliveryScopeID:          fixture.ScopeID,
			PermissionCheckScopeID:   fixture.ScopeID,
			AuthoritySubjectMemberID: fixture.HumanMemberID,
			CreatedByMemberID:        fixture.HumanMemberID,
			CreatedByActorType:       "human",
			CreatedByActorID:         fixture.UserID,
			CreatedVia:               "test",
			ActiveBindingKey:         "svc-runtime-binding-" + stringsFromTime(now),
			Now:                      now,
		})
		if err != nil {
			return err
		}
		_, _, err = txStore.UpsertEventSubscription(ctx, EventSubscriptionUpsert{
			ID:                     "svcesub_runtime_" + stringsFromTime(now),
			WorkspaceID:            fixture.WorkspaceID,
			ServiceDefinitionID:    seed.DefinitionID,
			ServiceInstallationID:  seed.InstallationID,
			ServiceMemberID:        seed.ServiceMemberID,
			ServiceConnectionRef:   AppRuntimeConnectionRefExternal,
			ServiceTargetBindingID: binding.ID,
			ServiceKind:            "app",
			SubscriptionKind:       "app_mention",
			TargetKind:             binding.TargetKind,
			TargetRef:              binding.TargetRef,
			DeliveryScopeID:        binding.DeliveryScopeID,
			PermissionCheckScopeID: binding.PermissionCheckScopeID,
			OwnerMemberID:          fixture.HumanMemberID,
			AuthorizerMemberID:     fixture.HumanMemberID,
			CreatedByActorType:     "human",
			CreatedByActorID:       fixture.UserID,
			CreatedVia:             "test",
			ExternalResourceType:   "agentteam_message_mention",
			ExternalResourceKey:    "msg_runtime_source:" + seed.InstallationID,
			ExternalResourceRef:    "msg_runtime_source",
			EventFilterKey:         "agentteam.message.mention",
			EventFilters:           []byte(`{"events":["agentteam.message.mention"]}`),
			DeliveryPolicy:         AppRuntimeDeliveryPolicyExternal,
			NotifyPrincipal:        false,
			ActiveSubscriptionKey:  "svc-runtime-subscription-" + stringsFromTime(now),
			Now:                    now,
		})
		if err != nil {
			return err
		}
		created, duplicate, err := txStore.ReserveEventDelivery(ctx, EventDeliveryReserve{
			ID:                  "svced_runtime_" + stringsFromTime(now),
			WorkspaceID:         fixture.WorkspaceID,
			EventSubscriptionID: "svcesub_runtime_" + stringsFromTime(now),
			EventType:           "agentteam.message.mention",
			EventAction:         "mention",
			EventID:             "msg_runtime_source",
			EventKey:            "msg_runtime_source:" + seed.InstallationID,
			EventPayload:        []byte(`{"message":{"id":"msg_runtime_source"}}`),
			ExternalResourceRef: "msg_runtime_source",
			IdempotencyKey:      "app_mention:msg_runtime_source:" + seed.InstallationID,
			Now:                 now,
		})
		if err != nil {
			return err
		}
		if duplicate {
			return conflict("expected new runtime delivery")
		}
		delivery = created
		return nil
	}))
	seedRuntimeAppAuthority(
		t,
		ctx,
		db,
		store,
		fixture,
		seed,
		"",
		[]string{RuntimeTransportKindRunnerStream, RuntimeTransportKindCallback},
		[]string{AppRuntimeEventMessageMention},
		[]string{"app.mentions.read", "messages.write"},
		now,
	)
	return delivery
}

func seedRuntimeEffectMessage(t testing.TB, db coredb.QueryExecer, fixture serviceActorFixture, seed runtimeServiceActorSeed, messageID string, now time.Time) {
	t.Helper()
	mustExec(t, db, `
INSERT INTO scope_messages (
  id, workspace_id, scope_id, seq, workspace_global_seq,
  author_member_id, author_type, message_type, content, metadata, created_at, updated_at
) VALUES ($1, $2, $3, 1, $4, $5, 'service', '', 'runtime effect', '{}', $6, $6)
`, messageID, fixture.WorkspaceID, fixture.ScopeID, now.UnixNano(), fixture.ServiceMemberID, now)
}

func stringsFromTime(value time.Time) string {
	return value.Format("20060102150405")
}