package app

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/app/capabilities"
	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/access"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/provider"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
)

var errProviderWatchFinalizationFailed = errors.New("provider watch finalization failed")

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

	var providerRequests atomic.Int64
	providerServer := httptest.NewServer(http.HandlerFunc(
		func(response http.ResponseWriter, request *http.Request) {
			providerRequests.Add(1)
			http.Error(response, "Gitea lookup must not run", http.StatusInternalServerError)
		},
	))
	t.Cleanup(providerServer.Close)

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{
		AppCredentialStorageKey: testAppCredentialStorageKey(),
		HTTPClient:              providerServer.Client(),
	})
	fixture := seedAppAdminFixture(
		t,
		pool,
		"provider_watch_gitea_resource_validation",
		time.Now().UTC(),
	)
	installation := requestAndApproveAppInstallationForTests(
		t,
		ctx,
		appStore,
		fixture.UserID,
		fixture.UserID,
		fixture.WorkspaceID,
		giteaAppSlug,
		map[string]any{
			giteaConfigInstanceURLKey:   providerServer.URL,
			giteaConfigAPITokenKey:      "gitea-api-token",
			giteaConfigWebhookSecretKey: "gitea-webhook-secret",
			giteaConfigDisplayNameKey:   "Gitea",
		},
	)
	upsertAppConnectionForTests(
		t,
		ctx,
		appStore,
		installation.ID,
		fixture.UserID,
		"acme",
		"acme",
		map[string]any{"webhook_secret": "gitea-webhook-secret"},
	)
	target, err := ResolveServiceEventTargetForScope(
		ctx,
		pool,
		fixture.WorkspaceID,
		fixture.MemberID,
		fixture.AllChannelID,
	)
	require.NoError(t, err)
	actor := ServiceEventActor{
		ActorType: "human",
		ActorID:   fixture.UserID,
		MemberID:  fixture.MemberID,
	}

	for _, test := range []struct {
		repository string
		want       error
	}{
		{repository: "acme/agentteam"},
		{repository: "other/agentteam", want: ErrPullRequestProviderAuthUnavailable},
	} {
		resource, err := NormalizePullRequestSubscriptionRef(
			"gitea",
			fmt.Sprintf("%s#42", test.repository),
		)
		require.NoError(t, err)

		_, err = appStore.CreateServiceEventSubscription(
			ctx,
			fixture.WorkspaceID,
			CreateServiceEventSubscriptionParams{
				Provider:         "gitea",
				InstallationID:   installation.ID,
				Actor:            actor,
				Target:           target,
				ResourceRef:      resource.ResourceRef,
				ResourceKey:      resource.ResourceKey,
				CreatedVia:       "app_command",
				ValidateResource: true,
			},
		)

		if test.want == nil {
			require.NoError(t, err)
		} else {
			require.ErrorIs(t, err, test.want)
		}
	}
	require.Zero(t, providerRequests.Load())
}

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

	providerServer := httptest.NewServer(http.HandlerFunc(
		func(response http.ResponseWriter, request *http.Request) {
			require.Contains(t, request.Header.Get("Authorization"), "Bearer ")
			switch {
			case request.Method == http.MethodPost &&
				request.URL.Path == "/app/installations/789/access_tokens":
				_, _ = response.Write([]byte(`{"token":"installation-token"}`))
			case request.Method == http.MethodGet &&
				request.URL.Path == "/repos/acme/agentteam/pulls/42":
				require.Equal(
					t,
					"Bearer installation-token",
					request.Header.Get("Authorization"),
				)
				_, _ = response.Write([]byte(`{"number":42}`))
			default:
				http.Error(response, "private provider detail", http.StatusNotFound)
			}
		},
	))
	t.Cleanup(providerServer.Close)

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	appStore := New(pool, Options{
		AppCredentialStorageKey: testAppCredentialStorageKey(),
		HTTPClient:              providerServer.Client(),
		GitHubAPIBaseURL:        providerServer.URL,
	})
	fixture := seedAppAdminFixture(
		t,
		pool,
		"provider_watch_github_resource_validation",
		time.Now().UTC(),
	)
	installation, err := appStore.installAppForUserForTest(
		ctx,
		fixture.UserID,
		fixture.WorkspaceID,
		InstallAppParams{AppSlug: "github"},
	)
	require.NoError(t, err)
	_, err = appStore.Provider().SaveGitHubAppConfigForWorkspaceAdmin(
		ctx,
		fixture.UserID,
		fixture.WorkspaceID,
		provider.UpsertGitHubAppConfigParams{
			AppID:         12345,
			AppSlug:       "agentteam-test",
			AppName:       "AgentTeam Test",
			HTMLURL:       "https://github.com/apps/agentteam-test",
			ClientID:      "Iv1.test",
			ClientSecret:  "client-secret",
			PrivateKeyPEM: testGitHubConnectionPrivateKey(t),
			WebhookSecret: "github-webhook-secret",
			OwnerLogin:    "acme",
			OwnerID:       67890,
		},
	)
	require.NoError(t, err)
	upsertAppConnectionForTests(
		t,
		ctx,
		appStore,
		installation.ID,
		fixture.UserID,
		"789",
		"acme",
		map[string]any{"webhook_secret": "github-webhook-secret"},
	)
	resource, err := NormalizePullRequestSubscriptionRef("github", "acme/agentteam#42")
	require.NoError(t, err)
	target, err := ResolveServiceEventTargetForScope(
		ctx,
		pool,
		fixture.WorkspaceID,
		fixture.MemberID,
		fixture.AllChannelID,
	)
	require.NoError(t, err)

	_, err = appStore.CreateServiceEventSubscription(
		ctx,
		fixture.WorkspaceID,
		CreateServiceEventSubscriptionParams{
			Provider:       "github",
			InstallationID: installation.ID,
			Actor: ServiceEventActor{
				ActorType: "human",
				ActorID:   fixture.UserID,
				MemberID:  fixture.MemberID,
			},
			Target:           target,
			ResourceRef:      resource.ResourceRef,
			ResourceKey:      resource.ResourceKey,
			CreatedVia:       "app_command",
			ValidateResource: true,
		},
	)

	require.NoError(t, err)
}

func TestCreateServiceEventSubscriptionCreatesBindingAndNoTargetParticipant(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, "provider_watch_create", now)

	reviewChannelID := "chn_provider_watch_create_review"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, reviewChannelID, "review", "", "private", fixture.MemberID, fixture.MemberID, now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	require.Equal(t, "channel", target.TargetKind)
	require.Equal(t, reviewChannelID, target.TargetRef)
	require.Equal(t, reviewChannelID, target.DeliveryScopeID)

	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/42",
		ResourceKey:  "github:acme/agentteam#42",
		EventFilters: map[string]any{"comments": true},
		CreatedVia:   "app_command",
	})
	require.NoError(t, err)
	require.False(t, result.AlreadyExists)
	binding, err := loadTargetAppBindingByID(ctx, pool, result.BindingID)
	require.NoError(t, err)
	require.NotEmpty(t, binding.ID)
	require.NotEmpty(t, result.Subscription.ID)
	require.Equal(t, TargetAppBindingStatusActive, binding.Status)
	require.Equal(t, ProviderWatchStatusActive, result.Subscription.Status)
	require.Equal(t, binding.ID, result.Subscription.AppBindingID)
	require.Equal(t, installation.MemberID, binding.AppMemberID)

	var serviceDefinitionID string
	var serviceInstallationID string
	var serviceTargetBindingID string
	var serviceConnectionRef string
	var subscriptionKind string
	var externalResourceKey string
	var activeSubscriptionKey string
	var serviceStatus string
	var ownerMemberID string
	var eventFilters []byte
	require.NoError(t, pool.QueryRow(ctx, `
SELECT service_definition_id, service_installation_id, service_target_binding_id,
       service_connection_ref, subscription_kind, external_resource_key,
       COALESCE(active_subscription_key, ''), status, owner_member_id, event_filters
FROM service_event_subscriptions
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, result.Subscription.ID).Scan(&serviceDefinitionID, &serviceInstallationID, &serviceTargetBindingID, &serviceConnectionRef, &subscriptionKind, &externalResourceKey, &activeSubscriptionKey, &serviceStatus, &ownerMemberID, &eventFilters))
	require.Equal(t, "app_github", serviceDefinitionID)
	require.Equal(t, installation.ID, serviceInstallationID)
	require.True(t, strings.HasPrefix(serviceTargetBindingID, "svctb_"))
	require.Equal(t, "789", serviceConnectionRef)
	require.Equal(t, "pull_request", subscriptionKind)
	require.Equal(t, "github:acme/agentteam#42", externalResourceKey)
	require.NotEmpty(t, activeSubscriptionKey)
	require.Equal(t, ProviderWatchStatusActive, serviceStatus)
	require.Equal(t, fixture.MemberID, ownerMemberID)
	require.JSONEq(t, `{"events":["comments"]}`, string(eventFilters))

	require.Equal(t, binding.ID, result.Subscription.AppBindingID)

	duplicate, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/42",
		ResourceKey:  "github:acme/agentteam#42",
		EventFilters: map[string]any{"comments": true},
		CreatedVia:   "app_command",
	})
	require.NoError(t, err)
	require.True(t, duplicate.AlreadyExists)
	require.Equal(t, binding.ID, duplicate.BindingID)
	require.Equal(t, result.Subscription.ID, duplicate.Subscription.ID)

	secondUser := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "provider_watch_create_second", now)
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, reviewChannelID, "review", "", "private", fixture.MemberID, secondUser.MemberID, now)
	secondTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, secondUser.MemberID, reviewChannelID)
	require.NoError(t, err)
	secondPrincipal, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   secondUser.UserID,
			MemberID:  secondUser.MemberID,
		},
		Target:       secondTarget,
		ResourceRef:  "https://github.com/acme/agentteam/pull/42",
		ResourceKey:  "github:acme/agentteam#42",
		EventFilters: map[string]any{"comments": true},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)
	require.False(t, secondPrincipal.AlreadyExists)
	require.Equal(t, binding.ID, secondPrincipal.BindingID)
	require.NotEqual(t, result.Subscription.ID, secondPrincipal.Subscription.ID)
	require.Equal(t, secondUser.MemberID, secondPrincipal.Subscription.CreatedByMemberID)

	var targetParticipantCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, reviewChannelID, installation.MemberID).Scan(&targetParticipantCount))
	require.Zero(t, targetParticipantCount)

	var bindingCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_target_bindings
WHERE workspace_id = $1 AND target_ref = $2 AND service_installation_id = $3 AND removed_at IS NULL
`, fixture.WorkspaceID, reviewChannelID, installation.ID).Scan(&bindingCount))
	require.Equal(t, 1, bindingCount)

	var watchCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE workspace_id = $1
  AND target_ref = $2
  AND external_resource_key = $3
  AND removed_at IS NULL
`, fixture.WorkspaceID, reviewChannelID, "github:acme/agentteam#42").Scan(&watchCount))
	require.Equal(t, 2, watchCount)
}

func TestProviderWatchUserVisibilityIsMemberScopedAndWorkspaceAdminWide(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, "provider_watch_visibility", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_visibility", now)
	secondUser := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "provider_watch_visibility_member", now)
	agentAuthorizer := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "provider_watch_visibility_authorizer", now)
	outsider := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "provider_watch_visibility_outsider", now)
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, reviewChannelID, "provider_watch_visibility_review", "", "private", fixture.MemberID, secondUser.MemberID, now)
	memberPrivateChannelID := "chn_provider_watch_visibility_member_private"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, memberPrivateChannelID, "provider-watch-member-private", "", "private", secondUser.MemberID, secondUser.MemberID, now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	adminResult, err := appStore.CreateServiceEventSubscriptionForUser(
		ctx,
		fixture.UserID,
		fixture.WorkspaceID,
		installation.ID,
		reviewChannelID,
		"https://github.com/acme/agentteam/pull/41",
		nil,
		nil,
	)
	require.NoError(t, err)

	memberResult, err := appStore.CreateServiceEventSubscriptionForUser(
		ctx,
		secondUser.UserID,
		fixture.WorkspaceID,
		installation.ID,
		memberPrivateChannelID,
		"https://github.com/acme/agentteam/pull/42",
		nil,
		nil,
	)
	require.NoError(t, err)
	require.Equal(t, secondUser.MemberID, memberResult.Subscription.CreatedByMemberID)

	agentID := "agt_provider_watch_visibility"
	agentMemberID := "mem_provider_watch_visibility_agent"
	agentPrivateChannelID := "chn_provider_watch_visibility_agent_private"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, agentPrivateChannelID, "provider-watch-agent-private", "", "private", fixture.MemberID, fixture.MemberID, now)
	seedAppPrivateAgentEntry(t, pool, fixture.WorkspaceID, secondUser.UserID, secondUser.MemberID, agentPrivateChannelID, "mac_provider_watch_visibility", agentID, agentMemberID, "visibility-agent", now)
	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:       fixture.WorkspaceID,
		AgentID:           agentID,
		AppInstallationID: installation.ID,
		// Deliberately authorize the agent through another member. Visibility
		// ownership must still follow the backing machine's owner, not the app
		// authorization actor.
		AuthorizedByMemberID: agentAuthorizer.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	agentTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, agentPrivateChannelID)
	require.NoError(t, err)
	// Deliberately pass a forged/stale authorizer projection. The create
	// transaction must overwrite it with the current authorization record.
	agentActor := ProviderWatchActor{ActorType: "agent", ActorID: agentID, MemberID: agentMemberID, AuthorizerUserID: outsider.MemberID}
	agentResult, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         agentTarget,
		ResourceRef:    "https://github.com/acme/agentteam/pull/43",
		ResourceKey:    "github:acme/agentteam#43",
		EventFilters:   map[string]any{},
		CreatedVia:     "mcp",
		Metadata: map[string]any{
			"caller_metadata": "preserved",
			serviceEventSubscriptionOwnerMemberIDSourceMetadataKey: "forged",
		},
	})
	require.NoError(t, err)
	require.Equal(t, secondUser.UserID, agentResult.Subscription.OwnerUserID)
	require.Equal(t, secondUser.MemberID, agentResult.Subscription.CreatedByMemberID)
	require.Equal(t, agentAuthorizer.MemberID, agentResult.Subscription.AuthorizerUserID)
	agentSubscriptionMetadata := decodeServiceEventSubscriptionJSONMap(agentResult.Subscription.Metadata)
	require.Equal(t, "preserved", agentSubscriptionMetadata["caller_metadata"])
	require.Equal(t, serviceEventSubscriptionOwnerMemberIDSourceAgentMachineV1, agentSubscriptionMetadata[serviceEventSubscriptionOwnerMemberIDSourceMetadataKey])

	// Workspace-wide management is human-only. An agent backed by an admin's
	// machine remains scoped to that admin's principal subscriptions and cannot
	// inherit the human admin's recovery authority over another member's row.
	adminOwnedAgentID := "agt_provider_watch_visibility_admin_owned"
	adminOwnedAgentMemberID := "mem_provider_watch_visibility_admin_owned"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, reviewChannelID, "mac_provider_watch_visibility_admin_owned", adminOwnedAgentID, adminOwnedAgentMemberID, "admin-owned-agent", now)
	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              adminOwnedAgentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	adminOwnedAgentActor := ProviderWatchActor{
		ActorType:        "agent",
		ActorID:          adminOwnedAgentID,
		MemberID:         adminOwnedAgentMemberID,
		AuthorizerUserID: fixture.MemberID,
	}
	adminOwnedAgentWatches, err := appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, adminOwnedAgentActor, installation.ID, "", "", "", ListServiceEventSubscriptionsOptions{})
	require.NoError(t, err)
	require.Len(t, adminOwnedAgentWatches, 1)
	require.Equal(t, adminResult.Subscription.ID, adminOwnedAgentWatches[0].ID)
	_, err = appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, adminOwnedAgentActor, "github", installation.ID, memberResult.Subscription.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, adminOwnedAgentActor, memberResult.Subscription.ID, "mcp")
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.RemoveServiceEventSubscription(ctx, fixture.WorkspaceID, adminOwnedAgentActor, memberResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, memberResult.Subscription.ID, ProviderWatchStatusActive, false, false)

	adminWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	adminWatchIDs := make(map[string]bool)
	for _, watch := range adminWatches {
		adminWatchIDs[watch.ID] = true
	}
	require.Len(t, adminWatches, 3)
	require.True(t, adminWatchIDs[adminResult.Subscription.ID])
	require.True(t, adminWatchIDs[memberResult.Subscription.ID])
	require.True(t, adminWatchIDs[agentResult.Subscription.ID])

	memberWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, secondUser.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	memberWatchIDs := make(map[string]bool)
	for _, watch := range memberWatches {
		memberWatchIDs[watch.ID] = true
	}
	require.Len(t, memberWatches, 2)
	require.True(t, memberWatchIDs[memberResult.Subscription.ID])
	require.True(t, memberWatchIDs[agentResult.Subscription.ID])
	require.False(t, memberWatchIDs[adminResult.Subscription.ID])

	authorizerWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	require.Empty(t, authorizerWatches, "authorizing another member's agent must not transfer subscription visibility ownership")

	agentWatches, err := appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, agentActor, installation.ID, "", "", "", ListServiceEventSubscriptionsOptions{})
	require.NoError(t, err)
	agentWatchIDs := make(map[string]bool)
	for _, watch := range agentWatches {
		agentWatchIDs[watch.ID] = true
	}
	require.Len(t, agentWatches, 2)
	require.True(t, agentWatchIDs[memberResult.Subscription.ID])
	require.True(t, agentWatchIDs[agentResult.Subscription.ID])
	require.False(t, agentWatchIDs[adminResult.Subscription.ID])

	// Delivery snapshots must preserve the same split as the resolved row:
	// owner is the backing machine owner, while authorizer remains the distinct
	// member who granted this agent access to the App.
	payload := providerWatchGitHubCommentPayload(43, "owner snapshot audit")
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-visibility-owner-snapshot")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))
	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 1, webhookResult.MatchedSubscriptionCount)
	require.Len(t, webhookResult.Deliveries, 1)
	var expectedOwnerMemberID, expectedAuthorizerMemberID string
	var deliveryOwnerMemberID, deliveryAuthorizerMemberID string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT subscription.owner_member_id, subscription.authorizer_member_id,
       delivery.owner_member_id_snapshot, delivery.authorizer_member_id_snapshot
FROM service_event_subscriptions subscription
JOIN service_event_deliveries delivery
  ON delivery.service_event_subscription_id = subscription.id
WHERE subscription.workspace_id = $1
  AND subscription.id = $2
ORDER BY delivery.created_at DESC, delivery.id DESC
LIMIT 1
`, fixture.WorkspaceID, agentResult.Subscription.ID).Scan(
		&expectedOwnerMemberID,
		&expectedAuthorizerMemberID,
		&deliveryOwnerMemberID,
		&deliveryAuthorizerMemberID,
	))
	require.Equal(t, secondUser.MemberID, expectedOwnerMemberID)
	require.Equal(t, agentAuthorizer.MemberID, expectedAuthorizerMemberID)
	require.Equal(t, expectedOwnerMemberID, deliveryOwnerMemberID)
	require.Equal(t, expectedAuthorizerMemberID, deliveryAuthorizerMemberID)

	// Simulate a legacy deleted-agent row whose owner snapshot was the app
	// authorizer and which predates the server-owned ownership marker. Neither
	// the authorizer nor the real machine owner may inherit access from this
	// ambiguous snapshot after the creator is gone.
	legacyAgentID := "agt_provider_watch_visibility_legacy_unmarked"
	legacyAgentMemberID := "mem_provider_watch_visibility_legacy_unmarked"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, secondUser.UserID, secondUser.MemberID, reviewChannelID, "mac_provider_watch_visibility_legacy_unmarked", legacyAgentID, legacyAgentMemberID, "legacy-unmarked-agent", now)
	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              legacyAgentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: agentAuthorizer.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	legacyTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, legacyAgentMemberID, reviewChannelID)
	require.NoError(t, err)
	legacyCreateParams := CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType:        "agent",
			ActorID:          legacyAgentID,
			MemberID:         legacyAgentMemberID,
			AuthorizerUserID: outsider.MemberID,
		},
		Target:       legacyTarget,
		ResourceRef:  "https://github.com/acme/agentteam/pull/44",
		ResourceKey:  "github:acme/agentteam#44",
		EventFilters: map[string]any{},
		CreatedVia:   "mcp",
	}
	legacyResult, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, legacyCreateParams)
	require.NoError(t, err)
	require.Equal(t, agentAuthorizer.MemberID, legacyResult.Subscription.AuthorizerUserID)
	// Reproduce the ambiguous pre-marker shape exactly: authorizer snapshotted
	// as owner and no trusted ownership metadata.
	mustExec(t, pool, `
UPDATE service_event_subscriptions
SET owner_member_id = $3,
    metadata = '{}'
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, legacyResult.Subscription.ID, agentAuthorizer.MemberID)
	legacyReplay, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, legacyCreateParams)
	require.NoError(t, err)
	require.True(t, legacyReplay.AlreadyExists)
	require.Equal(t, agentAuthorizer.MemberID, legacyReplay.Subscription.CreatedByMemberID)
	require.False(t, serviceEventSubscriptionHasAuthoritativeAgentMachineOwner(legacyReplay.Subscription), "replay must not retroactively sign an ambiguous legacy owner snapshot")
	mustExec(t, pool, `DELETE FROM agents WHERE workspace_id = $1 AND id = $2`, fixture.WorkspaceID, legacyAgentID)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, legacyResult.Subscription.ID, ProviderWatchStatusActive, false, false)

	// Runtime agent deletion is physical and intentionally does not cascade to
	// service-event subscriptions. The resolved owner snapshot must keep the
	// machine owner's subscription manageable without granting the authorizer or
	// an unrelated member access.
	mustExec(t, pool, `DELETE FROM agents WHERE workspace_id = $1 AND id = $2`, fixture.WorkspaceID, agentID)
	var orphanedSubscriptionCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE workspace_id = $1 AND id = $2 AND removed_at IS NULL
`, fixture.WorkspaceID, agentResult.Subscription.ID).Scan(&orphanedSubscriptionCount))
	require.Equal(t, 1, orphanedSubscriptionCount)

	orphanedOwnerWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, secondUser.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	orphanedOwnerWatchIDs := make(map[string]bool)
	for _, watch := range orphanedOwnerWatches {
		orphanedOwnerWatchIDs[watch.ID] = true
	}
	require.True(t, orphanedOwnerWatchIDs[agentResult.Subscription.ID])
	require.False(t, orphanedOwnerWatchIDs[legacyResult.Subscription.ID])
	authorizerWatches, err = appStore.ListServiceEventSubscriptionsForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	require.Empty(t, authorizerWatches)
	outsiderWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, outsider.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	require.Empty(t, outsiderWatches)

	_, err = appStore.GetServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.GetServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, legacyResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.GetServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, legacyResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.GetServiceEventSubscriptionForUser(ctx, outsider.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, adminResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, adminResult.Subscription.ID, ProviderWatchStatusActive, false, false)
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, legacyResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.RemoveServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, legacyResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, legacyResult.Subscription.ID, ProviderWatchStatusActive, false, false)
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, outsider.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusActive, false, false)

	// A member can get and run the full lifecycle for a subscription created by
	// an agent on their backing machine even when the target is agent-only and a
	// different member authorized that agent for the app.
	ownedAgentSubscription, err := appStore.GetServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.NoError(t, err)
	require.Equal(t, agentResult.Subscription.ID, ownedAgentSubscription.ID)
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusPaused, true, false)
	_, err = appStore.ResumeServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusPaused, true, false)
	_, err = appStore.ResumeServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusActive, false, false)
	_, err = appStore.RemoveServiceEventSubscriptionForUser(ctx, agentAuthorizer.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.RemoveServiceEventSubscriptionForUser(ctx, outsider.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusActive, false, false)
	_, err = appStore.RemoveServiceEventSubscriptionForUser(ctx, secondUser.UserID, fixture.WorkspaceID, agentResult.Subscription.ID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, agentResult.Subscription.ID, ProviderWatchStatusActive, false, true)

	// Workspace admins can manage any workspace subscription, including ones
	// created by another human or their agent.
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, fixture.UserID, fixture.WorkspaceID, memberResult.Subscription.ID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, memberResult.Subscription.ID, ProviderWatchStatusPaused, true, false)
}

func TestServiceEventAgentPrincipalFailsClosedWithoutActiveMachineOwner(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, "service_event_missing_agent_owner", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "service_event_missing_agent_owner", now)
	owner := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "service_event_missing_agent_owner_human", now)
	authorizer := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "service_event_missing_agent_owner_authorizer", now)
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, reviewChannelID, "service_event_missing_agent_owner_review", "", "private", fixture.MemberID, owner.MemberID, now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "790", "acme")

	agentID := "agt_service_event_missing_agent_owner"
	agentMemberID := "mem_service_event_missing_agent_owner_agent"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, owner.UserID, owner.MemberID, reviewChannelID, "mac_service_event_missing_agent_owner", agentID, agentMemberID, "missing-owner-agent", now)
	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: authorizer.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, reviewChannelID)
	require.NoError(t, err)
	agentActor := ProviderWatchActor{
		ActorType:        "agent",
		ActorID:          agentID,
		MemberID:         agentMemberID,
		AuthorizerUserID: fixture.MemberID, // stale/forged; live auth says authorizer
	}
	created, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/46",
		ResourceKey:    "github:acme/agentteam#46",
		EventFilters:   map[string]any{},
		CreatedVia:     "mcp",
	})
	require.NoError(t, err)
	require.Equal(t, owner.MemberID, created.Subscription.CreatedByMemberID)
	require.Equal(t, authorizer.MemberID, created.Subscription.AuthorizerUserID)

	humanTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	humanCreated, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType:        "human",
			ActorID:          fixture.UserID,
			MemberID:         fixture.MemberID,
			AuthorizerUserID: authorizer.MemberID, // forged; human must self-authorize
		},
		Target:       humanTarget,
		ResourceRef:  "https://github.com/acme/agentteam/pull/45",
		ResourceKey:  "github:acme/agentteam#45",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)
	require.Equal(t, fixture.MemberID, humanCreated.Subscription.CreatedByMemberID)
	require.Equal(t, fixture.MemberID, humanCreated.Subscription.AuthorizerUserID)

	// The machine still exists and the explicit app authorization remains
	// active, but its owning user no longer has an active human member. Agent
	// principal resolution must stop here instead of falling back to either the
	// authorization actor or the agent member.
	mustExec(t, pool, `UPDATE members SET status = 'inactive', updated_at = $2 WHERE id = $1`, owner.MemberID, now.Add(time.Minute))

	_, err = appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, agentActor, installation.ID, "", "", "", ListServiceEventSubscriptionsOptions{})
	require.ErrorContains(t, err, "service_event_principal_not_found")
	_, err = appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, agentActor, "github", installation.ID, created.Subscription.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "service_event_principal_not_found")
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, agentActor, created.Subscription.ID, "mcp")
	require.ErrorContains(t, err, "service_event_principal_not_found")
	_, err = appStore.RemoveServiceEventSubscription(ctx, fixture.WorkspaceID, agentActor, created.Subscription.ID)
	require.ErrorContains(t, err, "service_event_principal_not_found")
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/47",
		ResourceKey:    "github:acme/agentteam#47",
		EventFilters:   map[string]any{},
		CreatedVia:     "mcp",
	})
	require.ErrorContains(t, err, "service_event_principal_not_found")

	authorizerWatches, err := appStore.ListServiceEventSubscriptionsForUser(ctx, authorizer.UserID, fixture.WorkspaceID, installation.ID, "", "", "", 0)
	require.NoError(t, err)
	require.Empty(t, authorizerWatches)
	_, err = appStore.GetServiceEventSubscriptionForUser(ctx, authorizer.UserID, fixture.WorkspaceID, created.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, authorizer.UserID, fixture.WorkspaceID, created.Subscription.ID)
	require.ErrorContains(t, err, "subscription_not_found")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, created.Subscription.ID, ProviderWatchStatusActive, false, false)
}

func TestCreateServiceEventSubscriptionRequiresMatchingProviderConnection(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, "provider_watch_connection_access", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_connection_access", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchProviderConfig(t, ctx, appStore, fixture, "github")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	create := func(resource string) error {
		_, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
			Provider:       "github",
			InstallationID: installation.ID,
			Actor:          actor,
			Target:         target,
			ResourceRef:    resource,
			ResourceKey:    "github:acme/agentteam#42",
			EventFilters:   map[string]any{},
			CreatedVia:     "app_page",
		})
		return err
	}

	err = create("https://github.com/acme/agentteam/pull/42")
	require.ErrorContains(t, err, "workspace_not_configured")

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "other", map[string]any{"webhook_secret": "github-webhook-secret"})
	err = create("https://github.com/acme/agentteam/pull/42")
	require.ErrorContains(t, err, "resource_not_accessible")

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "790", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/42",
		ResourceKey:    "github:acme/agentteam#42",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	require.Equal(t, ProviderWatchStatusActive, result.Subscription.Status)
	require.Contains(t, string(result.Subscription.Metadata), "provider_connection_ref")
	require.Contains(t, string(result.Subscription.Metadata), "790")
}

func TestProviderWatchRateLimitsCreateAndWebhookDelivery(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, "provider_watch_rate_limit", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_rate_limit", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}

	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, actor, providerWatchRateLimitActionCreate, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionCreate).ActorLimit, now)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/123",
		ResourceKey:    "github:acme/agentteam#123",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.ErrorContains(t, err, "pull_request_subscription_rate_limited")

	mustExec(t, pool, `DELETE FROM service_action_rate_limit_events WHERE workspace_id = $1`, fixture.WorkspaceID)
	created, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/123",
		ResourceKey:    "github:acme/agentteam#123",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, ProviderWatchActor{ActorType: "app", ActorID: installation.ID}, providerWatchRateLimitActionWebhookDelivery, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionWebhookDelivery).InstallationLimit, now)
	payload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "acme/agentteam"},
	  "issue": {
	    "number": 123,
	    "html_url": "https://github.com/acme/agentteam/issues/123",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/123"}
	  },
	  "comment": {
	    "id": 1230,
	    "body": "rate limited comment",
	    "html_url": "https://github.com/acme/agentteam/issues/123#issuecomment-1230",
	    "user": {"login": "commenter"}
	  },
	  "sender": {"login": "commenter"}
	}`)
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-rate-limited")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	result, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.False(t, result.Duplicate)
	require.Zero(t, result.MatchedSubscriptionCount)
	require.Empty(t, result.Deliveries)

	var outcome string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT status
FROM service_webhook_receipts
WHERE service_installation_id = $1 AND delivery_key = 'delivery-provider-watch-rate-limited'
`, installation.ID).Scan(&outcome))
	require.Equal(t, "rate_limited", outcome)

	var receiptStatus, receiptErrorCode string
	var receiptMatched int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT status, error_code, matched_subscription_count
FROM service_webhook_receipts
WHERE service_installation_id = $1
  AND delivery_key = 'delivery-provider-watch-rate-limited'
`, installation.ID).Scan(&receiptStatus, &receiptErrorCode, &receiptMatched))
	require.Equal(t, "rate_limited", receiptStatus)
	require.Equal(t, "rate_limited", receiptErrorCode)
	require.Equal(t, 1, receiptMatched)

	requireServiceEventDeliveryFailure(t, pool, created.Subscription.ID, "provider_event_rate_limited")

	var serviceMessageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE author_member_id = $1
  AND content LIKE '%github:acme/agentteam#123%'
`, installation.MemberID).Scan(&serviceMessageCount))
	require.Zero(t, serviceMessageCount)
}

func TestProviderWatchRateLimitsListPauseResumeAndRemove(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, "provider_watch_rate_limit_lifecycle", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}

	activeWatch := createProviderWatchForRateLimitTest(t, ctx, appStore, fixture.WorkspaceID, installation.ID, actor, target, "124")
	pausedWatch := createProviderWatchForRateLimitTest(t, ctx, appStore, fixture.WorkspaceID, installation.ID, actor, target, "125")
	removeWatch := createProviderWatchForRateLimitTest(t, ctx, appStore, fixture.WorkspaceID, installation.ID, actor, target, "126")
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, actor, pausedWatch.ID, "setup pause")
	require.NoError(t, err)

	mustExec(t, pool, `DELETE FROM service_action_rate_limit_events WHERE workspace_id = $1`, fixture.WorkspaceID)
	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, actor, providerWatchRateLimitActionList, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionList).ActorLimit, now)
	_, err = appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, actor, installation.ID, target.TargetKind, target.TargetRef, "", ListServiceEventSubscriptionsOptions{})
	require.ErrorContains(t, err, "pull_request_subscription_rate_limited")

	mustExec(t, pool, `DELETE FROM service_action_rate_limit_events WHERE workspace_id = $1`, fixture.WorkspaceID)
	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, actor, providerWatchRateLimitActionPause, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionPause).ActorLimit, now)
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, actor, activeWatch.ID, "manual pause")
	require.ErrorContains(t, err, "pull_request_subscription_rate_limited")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, activeWatch.ID, ProviderWatchStatusActive, false, false)

	mustExec(t, pool, `DELETE FROM service_action_rate_limit_events WHERE workspace_id = $1`, fixture.WorkspaceID)
	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, actor, providerWatchRateLimitActionResume, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionResume).ActorLimit, now)
	_, err = appStore.ResumeServiceEventSubscription(ctx, fixture.WorkspaceID, actor, pausedWatch.ID)
	require.ErrorContains(t, err, "pull_request_subscription_rate_limited")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, pausedWatch.ID, ProviderWatchStatusPaused, true, false)

	mustExec(t, pool, `DELETE FROM service_action_rate_limit_events WHERE workspace_id = $1`, fixture.WorkspaceID)
	seedProviderWatchRateLimitEvents(t, pool, fixture.WorkspaceID, installation.ID, actor, providerWatchRateLimitActionRemove, providerWatchRateLimitConfigForAction(providerWatchRateLimitActionRemove).ActorLimit, now)
	_, err = appStore.RemoveServiceEventSubscription(ctx, fixture.WorkspaceID, actor, removeWatch.ID)
	require.ErrorContains(t, err, "pull_request_subscription_rate_limited")
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, removeWatch.ID, ProviderWatchStatusActive, false, false)
}

func TestProviderWatchUserCreateBootstrapsBindingWithoutTargetParticipant(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, "provider_watch_user_create", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_user_create", now)

	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	result, err := appStore.CreateServiceEventSubscriptionForUser(
		ctx,
		fixture.UserID,
		fixture.WorkspaceID,
		installation.ID,
		reviewChannelID,
		"acme/agentteam#7",
		nil,
		nil,
	)
	require.NoError(t, err)
	require.False(t, result.AlreadyExists)
	require.Equal(t, ProviderWatchStatusActive, result.Subscription.Status)
	require.Equal(t, "gitea:acme/agentteam#7", result.Subscription.ResourceKey)
	require.Equal(t, reviewChannelID, result.Subscription.TargetRef)
	require.Equal(t, reviewChannelID, result.Subscription.DeliveryScopeID)
	require.Equal(t, reviewChannelID, result.Subscription.PermissionCheckScopeID)
	require.Equal(t, providerWatchDeliveryPolicy, result.Subscription.DeliveryPolicy)
	require.True(t, result.Subscription.NotifyPrincipal)

	var bindingID, watchID string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT subscription.service_target_binding_id, subscription.id
FROM service_event_subscriptions subscription
WHERE subscription.workspace_id = $1
  AND subscription.target_ref = $2
  AND subscription.service_installation_id = $3
  AND subscription.external_resource_key = 'gitea:acme/agentteam#7'
  AND subscription.status = 'active'
  AND subscription.removed_at IS NULL
`, fixture.WorkspaceID, reviewChannelID, installation.ID).Scan(&bindingID, &watchID))
	require.NotEmpty(t, bindingID)
	require.NotEmpty(t, watchID)

	var targetParticipantCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, reviewChannelID, installation.MemberID).Scan(&targetParticipantCount))
	require.Zero(t, targetParticipantCount)

	require.True(t, strings.HasPrefix(watchID, "trg_"))
	_, err = appStore.PauseServiceEventSubscriptionForUser(ctx, fixture.UserID, fixture.WorkspaceID, watchID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, watchID, ProviderWatchStatusPaused, true, false)

	_, err = appStore.ResumeServiceEventSubscriptionForUser(ctx, fixture.UserID, fixture.WorkspaceID, watchID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, watchID, ProviderWatchStatusActive, false, false)

	_, err = appStore.RemoveServiceEventSubscriptionForUser(ctx, fixture.UserID, fixture.WorkspaceID, watchID)
	require.NoError(t, err)
	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, watchID, ProviderWatchStatusActive, false, true)
}

func TestProviderWatchWebhookDeliversAppServiceMessageAndDedupes(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, "provider_watch_webhook", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_webhook", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)

	_, err = appStore.Provider().SaveGitHubAppConfigForWorkspaceAdmin(ctx, fixture.UserID, fixture.WorkspaceID, provider.UpsertGitHubAppConfigParams{
		AppID:         12345,
		AppSlug:       "agentteam-test",
		AppName:       "AgentTeam Test",
		HTMLURL:       "https://github.com/apps/agentteam-test",
		ClientID:      "Iv1.test",
		ClientSecret:  "client-secret",
		PrivateKeyPEM: "-----BEGIN RSA PRIVATE KEY-----\nkey\n-----END RSA PRIVATE KEY-----",
		WebhookSecret: "github-webhook-secret",
		OwnerLogin:    "acme",
		OwnerID:       67890,
	})
	require.NoError(t, err)

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})

	watchAgentID := "agt_provider_watch_webhook_bot"
	watchAgentMemberID := "mem_provider_watch_webhook_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, reviewChannelID, "mac_provider_watch_webhook_bot", watchAgentID, watchAgentMemberID, "watch-agent", now)

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/42",
		ResourceKey:  "github:acme/agentteam#42",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)

	// PR1 runtime is resolved-only: the provider webhook verifies and routes
	// through the resolved service_connections row (the legacy app_connections
	// compat projection and its re-seed were removed). The connection is
	// deliberately retained here; deleting it would make the webhook
	// unverifiable and unmatched, which is the resolved-only contract.
	payload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "Acme/AgentTeam"},
	  "issue": {
	    "number": 42,
	    "html_url": "https://github.com/acme/agentteam/issues/42",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/42"}
	  },
	  "comment": {
	    "id": 1,
	    "body": "hello @watch-agent @all @online",
	    "html_url": "https://github.com/acme/agentteam/issues/42#issuecomment-1",
	    "created_at": "2026-04-19T09:08:07Z",
	    "user": {"login": "reviewer"}
	  },
	  "sender": {"login": "reviewer"}
	}`)
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-1")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.False(t, webhookResult.Duplicate)
	require.Equal(t, fixture.WorkspaceID, webhookResult.WorkspaceID)
	require.Equal(t, 1, webhookResult.MatchedSubscriptionCount)
	require.Len(t, webhookResult.Deliveries, 1)

	delivery := webhookResult.Deliveries[0]
	require.Equal(t, message.AuthorActorTypeService, delivery.Message.AuthorProfile.ActorType)
	require.Equal(t, installation.ID, delivery.AppMessage.AuthorProfile.PrincipalID)
	require.Equal(t, installation.MemberID, delivery.AppMessage.AuthorProfile.MemberID)
	require.Equal(t, "github", delivery.AppMessage.AuthorProfile.ServiceSlug)
	require.Contains(t, delivery.Message.Content, "github:acme/agentteam#42")
	require.Contains(t, delivery.Message.Content, "@\u200bwatch-agent")

	var authorType, authorMemberID, metadata string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT author_type, author_member_id, COALESCE(metadata, '')
FROM scope_messages
WHERE id = $1
`, delivery.MessageID).Scan(&authorType, &authorMemberID, &metadata))
	require.Equal(t, "service", authorType)
	require.Equal(t, installation.MemberID, authorMemberID)
	require.NotContains(t, metadata, "provider_watch")
	require.NotContains(t, metadata, "app_binding_id")

	requireScopeMessageAndSingleFanout(t, pool, delivery.MessageID, fixture.WorkspaceID, reviewChannelID)
	requireProviderWatchDeliverySource(t, pool, delivery, result.Subscription.ID)

	var expectedSubscriptionCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_subscriptions
WHERE id = $1
  AND service_target_binding_id = $2
`, result.Subscription.ID, result.Subscription.AppBindingID).Scan(&expectedSubscriptionCount))
	require.Equal(t, 1, expectedSubscriptionCount)

	var lastEventAt *time.Time
	var failureCode string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT last_success_at, COALESCE(last_failure_code, '')
FROM service_event_subscriptions
WHERE id = $1
`, result.Subscription.ID).Scan(&lastEventAt, &failureCode))
	require.NotNil(t, lastEventAt)
	require.Empty(t, failureCode)

	var (
		receiptID                string
		receiptDefinitionID      string
		receiptInstallationID    string
		receiptConnectionID      string
		receiptConnectionRef     string
		receiptEndpointID        string
		receiptServiceKind       string
		receiptEndpointKind      string
		receiptStatus            string
		receiptMatched           int
		receiptEventType         string
		receiptEventAction       string
		receiptEventID           string
		receiptEventKey          string
		receiptResourceType      string
		receiptResourceKey       string
		receiptResourceRef       string
		receiptResourceURL       string
		receiptProcessed         bool
		receiptNormalizedPayload string
	)
	require.NoError(t, pool.QueryRow(ctx, `
SELECT id, service_definition_id, service_installation_id,
       COALESCE(service_connection_id, ''), service_connection_ref,
       COALESCE(service_webhook_endpoint_id, ''), service_kind, endpoint_kind,
       status, matched_subscription_count, event_type, event_action, event_id,
       event_key, external_resource_type, external_resource_key,
       external_resource_ref, external_resource_url, processed_at IS NOT NULL,
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(normalized_event, '$.delivery_key')), '')
FROM service_webhook_receipts
WHERE service_installation_id = $1
  AND delivery_key = 'delivery-provider-watch-1'
`, installation.ID).Scan(
		&receiptID,
		&receiptDefinitionID,
		&receiptInstallationID,
		&receiptConnectionID,
		&receiptConnectionRef,
		&receiptEndpointID,
		&receiptServiceKind,
		&receiptEndpointKind,
		&receiptStatus,
		&receiptMatched,
		&receiptEventType,
		&receiptEventAction,
		&receiptEventID,
		&receiptEventKey,
		&receiptResourceType,
		&receiptResourceKey,
		&receiptResourceRef,
		&receiptResourceURL,
		&receiptProcessed,
		&receiptNormalizedPayload,
	))
	require.NotEmpty(t, receiptID)
	require.Equal(t, "app_github", receiptDefinitionID)
	require.Equal(t, installation.ID, receiptInstallationID)
	require.NotEmpty(t, receiptConnectionID)
	require.Equal(t, "789", receiptConnectionRef)
	require.NotEmpty(t, receiptEndpointID)
	require.Equal(t, "app", receiptServiceKind)
	require.Equal(t, "provider_webhook", receiptEndpointKind)
	require.Equal(t, "processed", receiptStatus)
	require.Equal(t, 1, receiptMatched)
	require.Equal(t, "issue_comment_created", receiptEventType)
	require.Equal(t, "created", receiptEventAction)
	require.Contains(t, receiptEventID, "delivery-provider-watch-1")
	require.Contains(t, receiptEventKey, receiptEventID)
	require.Equal(t, "pull_request", receiptResourceType)
	require.Equal(t, "github:acme/agentteam#42", receiptResourceKey)
	require.Equal(t, "github:acme/agentteam#42", receiptResourceRef)
	require.Equal(t, "https://github.com/acme/agentteam/issues/42#issuecomment-1", receiptResourceURL)
	require.True(t, receiptProcessed)
	require.Equal(t, "delivery-provider-watch-1", receiptNormalizedPayload)

	var (
		deliveryID             string
		deliveryReceiptID      string
		deliverySubscriptionID string
		deliveryStatus         string
		deliveryEffectKind     string
		deliveryEffectRef      string
		deliveryMessageID      string
		deliveryAttempts       int
		deliveryEventType      string
		deliveryEventAction    string
		deliveryEventID        string
		deliveryEventKey       string
		deliveryResourceKey    string
		deliveryTargetKind     string
		deliveryTargetRef      string
		deliveryNotify         bool
		deliveryPayload        string
	)
	require.NoError(t, pool.QueryRow(ctx, `
SELECT id, COALESCE(service_webhook_receipt_id, ''), service_event_subscription_id,
       status, effect_kind, effect_ref, COALESCE(message_id, ''), attempt_count,
       event_type, event_action, event_id, event_key, external_resource_key,
       target_kind, target_ref, notify_principal,
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(event_payload, '$.delivery_key')), '')
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
`, result.Subscription.ID).Scan(
		&deliveryID,
		&deliveryReceiptID,
		&deliverySubscriptionID,
		&deliveryStatus,
		&deliveryEffectKind,
		&deliveryEffectRef,
		&deliveryMessageID,
		&deliveryAttempts,
		&deliveryEventType,
		&deliveryEventAction,
		&deliveryEventID,
		&deliveryEventKey,
		&deliveryResourceKey,
		&deliveryTargetKind,
		&deliveryTargetRef,
		&deliveryNotify,
		&deliveryPayload,
	))
	require.NotEmpty(t, deliveryID)
	require.Equal(t, receiptID, deliveryReceiptID)
	require.Equal(t, result.Subscription.ID, deliverySubscriptionID)
	require.Equal(t, "succeeded", deliveryStatus)
	require.Equal(t, "message", deliveryEffectKind)
	require.Equal(t, delivery.MessageID, deliveryEffectRef)
	require.Equal(t, delivery.MessageID, deliveryMessageID)
	require.Equal(t, 1, deliveryAttempts)
	require.Equal(t, "issue_comment_created", deliveryEventType)
	require.Equal(t, "created", deliveryEventAction)
	require.Contains(t, deliveryEventID, "delivery-provider-watch-1")
	require.Contains(t, deliveryEventKey, deliveryEventID)
	require.Equal(t, "github:acme/agentteam#42", deliveryResourceKey)
	require.Equal(t, result.Subscription.TargetKind, deliveryTargetKind)
	require.Equal(t, result.Subscription.TargetRef, deliveryTargetRef)
	require.True(t, deliveryNotify)
	require.Equal(t, "delivery-provider-watch-1", deliveryPayload)

	duplicateResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.True(t, duplicateResult.Duplicate)
	require.Empty(t, duplicateResult.Deliveries)

	var serviceMessageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE author_member_id = $1 AND content LIKE '%github:acme/agentteam#42%'
`, installation.MemberID).Scan(&serviceMessageCount))
	require.Equal(t, 1, serviceMessageCount)

	var receiptCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_webhook_receipts
WHERE service_installation_id = $1
  AND delivery_key = 'delivery-provider-watch-1'
`, installation.ID).Scan(&receiptCount))
	require.Equal(t, 1, receiptCount)

	var deliveryCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
`, result.Subscription.ID).Scan(&deliveryCount))
	require.Equal(t, 1, deliveryCount)
}

func TestProviderWatchWebhookWritesParentMessageAndNotifiesPrincipalByPolicy(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, "provider_watch_app_policy_notify", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_app_policy_notify", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/43",
		ResourceKey:  "github:acme/agentteam#43",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)
	require.Equal(t, providerWatchDeliveryPolicy, result.Subscription.DeliveryPolicy)
	require.True(t, result.Subscription.NotifyPrincipal)

	payload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "Acme/AgentTeam"},
	  "issue": {
	    "number": 43,
	    "html_url": "https://github.com/acme/agentteam/issues/43",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/43"}
	  },
	  "comment": {
	    "id": 1,
	    "body": "notify principal",
	    "html_url": "https://github.com/acme/agentteam/issues/43#issuecomment-1",
	    "created_at": "2026-04-19T09:08:07Z",
	    "user": {"login": "reviewer"}
	  },
	  "sender": {"login": "reviewer"}
	}`)
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-notify-principal")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.Len(t, webhookResult.Deliveries, 1)
	require.Equal(t, "channel", webhookResult.Deliveries[0].Message.ChannelType)
	requireServiceEventDeliveryEffect(t, pool, result.Subscription.ID, "succeeded", "message", webhookResult.Deliveries[0].MessageID, webhookResult.Deliveries[0].MessageID)

	var parentMessageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE id = $1
  AND scope_id = $2
  AND author_member_id = $3
  AND content LIKE '%github:acme/agentteam#43%'
`, webhookResult.Deliveries[0].MessageID, reviewChannelID, installation.MemberID).Scan(&parentMessageCount))
	require.Equal(t, 1, parentMessageCount)

	var sourceCopyCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_message_sources
WHERE source_message_id = $1
`, webhookResult.Deliveries[0].MessageID).Scan(&sourceCopyCount))
	require.Zero(t, sourceCopyCount)

	requireScopeMessageAndSingleFanout(t, pool, webhookResult.Deliveries[0].MessageID, fixture.WorkspaceID, reviewChannelID)
}

func TestProviderWatchWebhookCanDisablePrincipalNotification(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, "provider_watch_app_policy_no_notify", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_app_policy_no_notify", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	notifyPrincipal := false
	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:          target,
		ResourceRef:     "https://github.com/acme/agentteam/pull/44",
		ResourceKey:     "github:acme/agentteam#44",
		EventFilters:    map[string]any{},
		NotifyPrincipal: &notifyPrincipal,
		CreatedVia:      "app_page",
	})
	require.NoError(t, err)
	require.False(t, result.Subscription.NotifyPrincipal)

	payload := providerWatchGitHubCommentPayload(44, "do not notify principal")
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-no-notify")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.Len(t, webhookResult.Deliveries, 1)
	require.Empty(t, webhookResult.Deliveries[0].FollowupDeliveries)
	requireServiceEventDeliveryPolicy(t, pool, result.Subscription.ID, false)

	requireScopeMessageAndSingleFanout(t, pool, webhookResult.Deliveries[0].MessageID, fixture.WorkspaceID, reviewChannelID)
}

func TestProviderWatchWebhookCoalescesMatchingSubscriptionsToOneMessage(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, "provider_watch_coalesce", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_coalesce", now)
	secondUser := seedWorkspaceMemberFixture(t, pool, fixture.WorkspaceID, "provider_watch_coalesce_second", now)
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, reviewChannelID, "provider_watch_coalesce", "", "private", fixture.MemberID, secondUser.MemberID, now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	notifyPrincipal := false
	firstTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	firstWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:          firstTarget,
		ResourceRef:     "https://github.com/acme/agentteam/pull/46",
		ResourceKey:     "github:acme/agentteam#46",
		EventFilters:    map[string]any{"comments": true},
		NotifyPrincipal: &notifyPrincipal,
		CreatedVia:      "app_page",
	})
	require.NoError(t, err)

	secondTarget, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, secondUser.MemberID, reviewChannelID)
	require.NoError(t, err)
	secondWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   secondUser.UserID,
			MemberID:  secondUser.MemberID,
		},
		Target:          secondTarget,
		ResourceRef:     "https://github.com/acme/agentteam/pull/46",
		ResourceKey:     "github:acme/agentteam#46",
		EventFilters:    map[string]any{"comments": true},
		NotifyPrincipal: &notifyPrincipal,
		CreatedVia:      "app_page",
	})
	require.NoError(t, err)
	require.NotEqual(t, firstWatch.Subscription.ID, secondWatch.Subscription.ID)
	require.Equal(t, firstWatch.BindingID, secondWatch.BindingID)

	payload := providerWatchGitHubCommentPayload(46, "coalesce matching subscriptions")
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-coalesce")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 2, webhookResult.MatchedSubscriptionCount)
	require.Len(t, webhookResult.Deliveries, 1)
	messageID := webhookResult.Deliveries[0].MessageID
	require.NotEmpty(t, messageID)

	requireServiceEventDeliveryEffect(t, pool, firstWatch.Subscription.ID, "succeeded", "message", messageID, messageID)
	requireServiceEventDeliveryEffect(t, pool, secondWatch.Subscription.ID, "succeeded", "message", messageID, messageID)

	var coalescedDeliveryRows int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_deliveries
WHERE service_event_subscription_id IN ($1, $2)
  AND status = 'succeeded'
  AND effect_kind = 'message'
  AND effect_ref = $3
  AND message_id = $3
`, firstWatch.Subscription.ID, secondWatch.Subscription.ID, messageID).Scan(&coalescedDeliveryRows))
	require.Equal(t, 2, coalescedDeliveryRows)

	var serviceMessageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE scope_id = $1
  AND author_member_id = $2
  AND content LIKE '%github:acme/agentteam#46%'
`, reviewChannelID, installation.MemberID).Scan(&serviceMessageCount))
	require.Equal(t, 1, serviceMessageCount)
	requireScopeMessageAndSingleFanout(t, pool, messageID, fixture.WorkspaceID, reviewChannelID)
}

func TestProviderWatchWebhookChannelDeliveryRollsBackWhenRunFinalizationFails(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, "provider_watch_channel_atomic", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_channel_atomic", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/45",
		ResourceKey:  "github:acme/agentteam#45",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)

	payload := providerWatchGitHubCommentPayload(45, "rollback channel delivery")
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-channel-atomic")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	failingStore := New(&providerWatchFailSuccessCompleteDB{DB: pool}, Options{AppCredentialStorageKey: testAppCredentialStorageKey()})
	webhookResult, err := failingStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.Empty(t, webhookResult.Deliveries)

	var messageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE scope_id = $1
  AND content LIKE '%rollback channel delivery%'
`, reviewChannelID).Scan(&messageCount))
	require.Zero(t, messageCount)

	var failedDeliveryCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM service_event_deliveries
WHERE event_id = $1
  AND status = 'failed'
`, "github:delivery-provider-watch-channel-atomic").Scan(&failedDeliveryCount))
	require.Equal(t, 1, failedDeliveryCount)
}

func TestProviderWatchWebhookSkipsPausedAndRemovedWatches(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, "provider_watch_webhook_inactive", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_webhook_inactive", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)

	paused, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/48",
		ResourceKey:    "github:acme/agentteam#48",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, actor, paused.Subscription.ID, "manual pause")
	require.NoError(t, err)

	removed, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/49",
		ResourceKey:    "github:acme/agentteam#49",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.RemoveServiceEventSubscription(ctx, fixture.WorkspaceID, actor, removed.Subscription.ID)
	require.NoError(t, err)

	for _, testCase := range []struct {
		name     string
		prNumber int
		delivery string
	}{
		{name: "paused", prNumber: 48, delivery: "delivery-provider-watch-paused-skip"},
		{name: "removed", prNumber: 49, delivery: "delivery-provider-watch-removed-skip"},
	} {
		t.Run(testCase.name, func(t *testing.T) {
			payload := providerWatchGitHubCommentPayload(testCase.prNumber, "inactive watch should not deliver")
			headers := http.Header{}
			headers.Set("X-GitHub-Event", "issue_comment")
			headers.Set("X-GitHub-Delivery", testCase.delivery)
			headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

			webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
			require.NoError(t, err)
			require.False(t, webhookResult.Duplicate)
			require.Zero(t, webhookResult.MatchedSubscriptionCount)
			require.Empty(t, webhookResult.Deliveries)
			if testCase.name == "paused" {
				requireServiceEventDeliveryCount(t, pool, paused.Subscription.ID, 0)
			} else {
				requireServiceEventDeliveryCount(t, pool, removed.Subscription.ID, 0)
			}

			var messageCount int
			require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE content LIKE $1
`, fmt.Sprintf("%%github:acme/agentteam#%d%%", testCase.prNumber)).Scan(&messageCount))
			require.Zero(t, messageCount)
		})
	}
}

func TestProviderWatchWebhookDeliversTaskDiscussionEntry(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, "provider_watch_task_delivery", now)
	taskID := "tsk_provider_watch_task_delivery"
	taskBoardID := "tbd_provider_watch_task_delivery"
	mustExec(t, pool, `
INSERT INTO task_boards (channel_id, workspace_id, next_task_number, created_at, updated_at, id)
VALUES ($1, $2, 2, $3, $3, $4)
`, fixture.AllChannelID, fixture.WorkspaceID, now, taskBoardID)
	mustExec(t, pool, `
INSERT INTO task_counters (board_id, next_task_number, updated_at)
VALUES ($1, 2, $2)
`, taskBoardID, now)
	mustExec(t, pool, `
INSERT INTO scopes (id, workspace_id, type, identity_key, status, created_by_member_id, created_at, updated_at)
VALUES ($1, $2, 'task', CONCAT('scope/task/', $1), 'active', $3, $4, $4)
`, taskID, fixture.WorkspaceID, fixture.MemberID, now)
	mustExec(t, pool, `
INSERT INTO tasks (
    id, task_scope_id, board_id, channel_id, task_number, title, status,
    created_by_member_id, created_by_ref, created_at, updated_at
) VALUES ($1, $1, $2, $3, 1, 'Review provider webhook', 'in_progress', $4, $4, $5, $5)
`, taskID, taskBoardID, fixture.AllChannelID, fixture.MemberID, now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)

	_, err = appStore.Provider().SaveGitHubAppConfigForWorkspaceAdmin(ctx, fixture.UserID, fixture.WorkspaceID, provider.UpsertGitHubAppConfigParams{
		AppID:         12345,
		AppSlug:       "agentteam-test",
		AppName:       "AgentTeam Test",
		HTMLURL:       "https://github.com/apps/agentteam-test",
		ClientID:      "Iv1.test",
		ClientSecret:  "client-secret",
		PrivateKeyPEM: "-----BEGIN RSA PRIVATE KEY-----\nkey\n-----END RSA PRIVATE KEY-----",
		WebhookSecret: "github-webhook-secret",
		OwnerLogin:    "acme",
		OwnerID:       67890,
	})
	require.NoError(t, err)

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, "task:"+taskID)
	require.NoError(t, err)
	require.Equal(t, "task", target.TargetKind)
	require.Equal(t, taskID, target.TargetRef)
	require.Equal(t, taskID, target.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, target.PermissionCheckScopeID)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/42",
		ResourceKey:  "github:acme/agentteam#42",
		EventFilters: map[string]any{"comments": true},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)
	binding, err := loadTargetAppBindingByID(ctx, pool, result.BindingID)
	require.NoError(t, err)
	require.Equal(t, taskID, binding.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, binding.PermissionCheckScopeID)
	require.Equal(t, fixture.MemberID, binding.AuthoritySubjectMemberID)
	require.JSONEq(t, fmt.Sprintf(`{
		"targetKind": "task",
		"targetRef": %q,
		"storageScopeId": %q,
		"storageScopeType": "task",
		"deliveryScopeId": %q,
		"deliveryScopeType": "task",
		"permissionCheckScopeId": %q,
		"authoritySubjectMemberId": %q,
		"permissionBasis": [{
			"kind": "task_board_channel",
			"scopeId": %q,
			"scopeType": "channel",
			"decisionSource": "channel_membership",
			"memberId": %q,
			"ownerMemberId": %q,
			"grantId": "",
			"reason": ""
		}],
		"authoritySubjects": [{
			"kind": "routing_principal",
			"principalType": "",
			"principalId": "",
			"memberId": %q,
			"installationId": ""
		}]
	}`, taskID, taskID, taskID, fixture.AllChannelID, fixture.MemberID, fixture.AllChannelID, fixture.MemberID, fixture.MemberID, fixture.MemberID), string(binding.DeliveryContract))
	require.Equal(t, taskID, result.Subscription.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, result.Subscription.PermissionCheckScopeID)
	require.Equal(t, fixture.MemberID, result.Subscription.AuthoritySubjectMemberID)
	require.JSONEq(t, string(binding.DeliveryContract), string(result.Subscription.DeliveryContract))

	payload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "Acme/AgentTeam"},
	  "issue": {
	    "number": 42,
	    "html_url": "https://github.com/acme/agentteam/issues/42",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/42"}
	  },
	  "comment": {
	    "id": 1,
	    "body": "task update @all",
	    "html_url": "https://github.com/acme/agentteam/issues/42#issuecomment-1",
	    "created_at": "2026-04-19T09:08:07Z",
	    "user": {"login": "reviewer"}
	  },
	  "sender": {"login": "reviewer"}
	}`)
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-task-1")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.False(t, webhookResult.Duplicate)
	require.Equal(t, 1, webhookResult.MatchedSubscriptionCount)
	require.Len(t, webhookResult.Deliveries, 1, "provider Task event must return its MessageDelivery through AppWebhookResult.Deliveries")

	var discussionCount int
	var discussionEntryID string
	var createdByRef string
	var body string
	var metadata string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*), MAX(sm.id), MAX(sm.author_member_id), MAX(sm.content), COALESCE(MAX(sm.metadata), '')
FROM tasks t
JOIN scope_messages sm
  ON sm.scope_id = t.task_scope_id
 AND sm.message_type = 'task_discussion'
WHERE t.id = $1
`, taskID).Scan(&discussionCount, &discussionEntryID, &createdByRef, &body, &metadata))
	require.Equal(t, 1, discussionCount)
	require.Equal(t, installation.MemberID, createdByRef)
	require.Contains(t, body, "github:acme/agentteam#42")
	require.Contains(t, body, "@\u200ball")
	require.Empty(t, metadata)

	var taskScopeID string
	var discussionSeq int64
	require.NoError(t, pool.QueryRow(ctx, `
SELECT t.task_scope_id, sm.seq
FROM tasks t
JOIN scope_messages sm
  ON sm.scope_id = t.task_scope_id
 AND sm.id = $2
 AND sm.message_type = 'task_discussion'
WHERE t.id = $1
`, taskID, discussionEntryID).Scan(&taskScopeID, &discussionSeq))

	var outboxCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_fanout_outbox
WHERE message_id = $1
  AND event_ref_type = 'message'
  AND fanout_type = 'scope_fanout'
`, discussionEntryID).Scan(&outboxCount))
	require.Equal(t, 1, outboxCount, "provider Task event must write the resolved scope_fanout outbox row")
	var outboxPayloadSet int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_fanout_outbox
WHERE message_id = $1
  AND payload IS NOT NULL
`, discussionEntryID).Scan(&outboxPayloadSet))
	require.Equal(t, 1, outboxPayloadSet, "provider Task event outbox row must carry the frozen scope fanout payload")

	var searchDocCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_message_search_docs
WHERE message_id = $1
  AND scope_id = $2
`, discussionEntryID, taskScopeID).Scan(&searchDocCount))
	require.Equal(t, 1, searchDocCount, "provider-watch silent task discussion remains a resolved searchable scope message")

	var syncDirtyCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_sync_dirty
WHERE workspace_id = $1
  AND scope_id = $2
  AND latest_message_seq >= $3
  AND reason = 'message'
`, fixture.WorkspaceID, taskScopeID, discussionSeq).Scan(&syncDirtyCount))
	require.Equal(t, 1, syncDirtyCount, "provider-watch silent task discussion still dirties message sync for read repair")

	var latestSeq, visibleMessageCount int64
	require.NoError(t, pool.QueryRow(ctx, `
SELECT latest_seq, visible_message_count
FROM scopes
WHERE id = $1
`, taskScopeID).Scan(&latestSeq, &visibleMessageCount))
	require.Equal(t, discussionSeq, latestSeq)
	require.Equal(t, int64(1), visibleMessageCount)

	var eventCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM tasks t
JOIN scope_activity_events ae
  ON ae.scope_id = t.task_scope_id
 AND ae.activity_ref_type = 'task_event'
WHERE t.id = $1
  AND ae.actor_member_id = $2
  AND ae.activity_type = 'task.discussion_added'
`, taskID, installation.MemberID).Scan(&eventCount))
	require.Equal(t, 1, eventCount)

	var taskDiscussionCount int
	require.NoError(t, pool.QueryRow(ctx, `SELECT discussion_count FROM tasks WHERE id = $1`, taskID).Scan(&taskDiscussionCount))
	require.Equal(t, 1, taskDiscussionCount)
	require.Equal(t, taskID, result.Subscription.TargetRef)

	requireServiceEventDeliveryEffect(t, pool, result.Subscription.ID, "succeeded", "message", discussionEntryID, discussionEntryID)

	// The prior Senior review is authoritative: the stored provider Task outbox
	// payload must decode to exactly the delivery returned through
	// AppWebhookResult.Deliveries, including notify principal, service author
	// presentation, and contribution source. Prove it field by field so replay
	// after a failed first fanout preserves the same projection.
	require.Len(t, webhookResult.Deliveries, 1)
	returnedDelivery := webhookResult.Deliveries[0]
	require.Equal(t, discussionEntryID, returnedDelivery.MessageID, "returned provider Task delivery must carry the committed discussion message id")
	var storedPayload []byte
	require.NoError(t, pool.QueryRow(ctx, `
SELECT payload
FROM scope_fanout_outbox
WHERE message_id = $1
  AND event_ref_type = 'message'
  AND event_ref_id = $1
  AND fanout_type = 'scope_fanout'
LIMIT 1
`, discussionEntryID).Scan(&storedPayload))
	decodedPayload, err := message.UnmarshalScopeFanoutPayload(storedPayload)
	require.NoError(t, err, "stored provider Task outbox payload must decode")
	require.Equal(t, returnedDelivery.MessageID, decodedPayload.Delivery.MessageID)
	require.Equal(t, returnedDelivery.NotifyPrincipalMemberID, decodedPayload.Delivery.NotifyPrincipalMemberID, "stored payload must freeze the returned notify principal")
	require.Equal(t, returnedDelivery.Message.NotifyPrincipalMemberID, decodedPayload.Delivery.Message.NotifyPrincipalMemberID)
	require.Equal(t, returnedDelivery.HumanRecipientsTargeted, decodedPayload.Delivery.HumanRecipientsTargeted)
	require.Equal(t, returnedDelivery.HumanRecipientMemberIDs, decodedPayload.Delivery.HumanRecipientMemberIDs)
	require.Equal(t, returnedDelivery.Message.Source, decodedPayload.Delivery.Message.Source, "stored payload must freeze the returned contribution source")
	require.Equal(t, returnedDelivery.Message.AuthorProfile, decodedPayload.Delivery.Message.AuthorProfile, "stored payload must freeze the returned service author profile")
	require.Equal(t, returnedDelivery.Message.AuthorProfile.ServiceSlug, "github")
	require.Equal(t, returnedDelivery.Message.AuthorProfile.ActorType, message.AuthorActorTypeService)
	require.Equal(t, returnedDelivery.Message.AuthorProfile.MemberID, installation.MemberID)
	require.Equal(t, returnedDelivery.AppMessage.Source, decodedPayload.AppMessage.Source)
	require.Equal(t, returnedDelivery.AppMessage.EffectiveAuthorProfile(), decodedPayload.AppMessage.EffectiveAuthorProfile(), "stored AppMessage author profile must match the returned projection")

	var timeline taskboard.GetTaskTimelineResponse
	var timelineErr error
	require.Eventually(t, func() bool {
		timeline, timelineErr = taskboard.New(pool).GetTaskTimeline(ctx, taskboard.GetTaskTimelineRequest{
			Context: taskboard.NewUserServiceContext(fixture.WorkspaceID, fixture.UserID, fixture.UserID),
			Board:   taskboard.BoardRef{ChannelID: fixture.AllChannelID},
			Task:    taskboard.TaskRef{TaskID: taskID},
		})
		return timelineErr == nil && len(taskboardTimelineDiscussionEntriesForTest(timeline)) == 1
	}, 3*time.Second, 50*time.Millisecond)
	require.NoError(t, timelineErr)
	discussions := taskboardTimelineDiscussionEntriesForTest(timeline)
	require.Len(t, discussions, 1)
	require.Equal(t, "app", discussions[0].CreatedBy.Type)
	require.Equal(t, installation.ID, discussions[0].CreatedBy.ID)
	require.Equal(t, "GitHub", discussions[0].CreatedBy.Name)
	require.Equal(t, message.AuthorActorTypeService, discussions[0].AuthorProfile.ActorType)
	require.Equal(t, installation.ID, discussions[0].AuthorProfile.PrincipalID)
	require.Equal(t, installation.MemberID, discussions[0].AuthorProfile.MemberID)
	require.Equal(t, "github", discussions[0].AuthorProfile.ServiceSlug)
}

func TestProviderWatchTaskTargetResolverAllowsPrivateAgentEntry(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, "provider_watch_task_private_agent", now)

	agentID := "agt_provider_watch_task_private"
	agentMemberID := "mem_provider_watch_task_private"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "private-task-agent", "private", false, now)

	taskBoardID := "tb_provider_watch_task_private"
	taskID := "tsk_provider_watch_task_private"
	mustExec(t, pool, `
INSERT INTO task_boards (id, channel_id, workspace_id, next_task_number, created_at, updated_at)
VALUES ($1, $2, $3, 2, $4, $4)
ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at)
`, taskBoardID, fixture.AllChannelID, fixture.WorkspaceID, now)
	mustExec(t, pool, `
INSERT INTO task_counters (board_id, next_task_number, updated_at)
VALUES ($1, 2, $2)
ON DUPLICATE KEY UPDATE next_task_number = VALUES(next_task_number), updated_at = VALUES(updated_at)
`, taskBoardID, now)
	mustExec(t, pool, `
INSERT INTO scopes (id, workspace_id, type, identity_key, status, created_by_member_id, created_at, updated_at)
VALUES ($1, $2, 'task', CONCAT('scope/task/', $1), 'active', $3, $4, $4)
ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at)
`, taskID, fixture.WorkspaceID, fixture.MemberID, now)
	mustExec(t, pool, `
INSERT INTO tasks (
    id,
    task_scope_id,
    board_id,
    channel_id,
    task_number,
    title,
    status,
    created_by_member_id,
    created_by_ref,
    current_agent_ref,
    created_at,
    updated_at
) VALUES ($1, $1, $2, $3, 1, 'Private agent provider watch task', 'in_progress', $4, $4, $5, $6, $6)
`, taskID, taskBoardID, fixture.AllChannelID, fixture.MemberID, agentMemberID, now)

	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	channelNumberTarget, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "task:#all:1")
	require.NoError(t, err)
	require.Equal(t, "task", channelNumberTarget.TargetKind)
	require.Equal(t, taskID, channelNumberTarget.TargetRef)
	require.Equal(t, taskID, channelNumberTarget.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, channelNumberTarget.PermissionCheckScopeID)

	taskIDTarget, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "task:"+taskID)
	require.NoError(t, err)
	require.Equal(t, "task", taskIDTarget.TargetKind)
	require.Equal(t, taskID, taskIDTarget.TargetRef)
	require.Equal(t, taskID, taskIDTarget.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, taskIDTarget.PermissionCheckScopeID)

	var participantCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, fixture.AllChannelID, agentMemberID).Scan(&participantCount))
	require.Equal(t, 1, participantCount)
}

func TestProviderWatchChannelTargetResolverAllowsPrivateAgentEntry(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, "provider_watch_channel_private_agent", now)

	agentID := "agt_provider_watch_channel_private"
	agentMemberID := "mem_provider_watch_channel_private"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "private-channel-agent", "private", false, now)

	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	resolved, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "#all")
	require.NoError(t, err)
	require.Equal(t, "channel", resolved.TargetKind)
	require.Equal(t, fixture.AllChannelID, resolved.TargetRef)
	require.Equal(t, fixture.AllChannelID, resolved.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, resolved.PermissionCheckScopeID)

	directResolved, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, fixture.AllChannelID)
	require.NoError(t, err)
	require.Equal(t, resolved, directResolved)

	var publicMemberships int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND visibility = 'public'
  AND removed_at IS NULL
`, fixture.AllChannelID, agentMemberID).Scan(&publicMemberships))
	require.Equal(t, 0, publicMemberships)
}

func TestProviderWatchThreadTargetResolverAllowsPrivateAgentEntry(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, "provider_watch_thread_private_agent", now)

	agentID := "agt_provider_watch_thread_private"
	agentMemberID := "mem_provider_watch_thread_private"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "private-thread-agent", "private", false, now)

	threadChannelID := "chn_provider_watch_thread_private_reply"
	threadShortID := "abc12345"
	seedProviderWatchReplyThread(t, pool, fixture, threadChannelID, threadShortID, now)

	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	resolved, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "#all:"+threadShortID)
	require.NoError(t, err)
	require.Equal(t, "thread", resolved.TargetKind)
	require.Equal(t, threadChannelID, resolved.TargetRef)
	require.Equal(t, threadChannelID, resolved.DeliveryScopeID)
	require.Equal(t, fixture.AllChannelID, resolved.PermissionCheckScopeID)

	directResolved, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, threadChannelID)
	require.NoError(t, err)
	require.Equal(t, resolved, directResolved)

	var participantCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, fixture.AllChannelID, agentMemberID).Scan(&participantCount))
	require.Equal(t, 1, participantCount)
}

func TestProviderWatchThreadTargetResolverAllowsPrivateAgentByParentRoster(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, "provider_watch_thread_private_hidden_root", now)

	agentID := "agt_provider_watch_thread_hidden"
	agentMemberID := "mem_provider_watch_thread_hidden"
	hiddenHumanMemberID := "mem_provider_watch_thread_hidden_human"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "private-thread-hidden", "private", false, now)
	mustExec(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, workspace_role, created_at, updated_at)
VALUES ($1, $2, 'human', 'Hidden Human', 'member', $3, $3)
`, hiddenHumanMemberID, fixture.WorkspaceID, now)

	threadChannelID := "chn_provider_watch_thread_hidden_reply"
	threadShortID := "abc12345"
	seedProviderWatchReplyThreadWithRootAuthor(t, pool, fixture, threadChannelID, threadShortID, hiddenHumanMemberID, now)

	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	resolved, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "#all:"+threadShortID)
	require.NoError(t, err)
	require.Equal(t, threadChannelID, resolved.TargetRef)
	require.Equal(t, access.ScopePermissionBasisOwnerPrivateMembership, resolved.ResolvedScopeTarget.PermissionBasis[0].Kind)
	directResolved, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, threadChannelID)
	require.NoError(t, err)
	require.Equal(t, resolved, directResolved)
}

func TestProviderWatchThreadTargetResolverRejectsPrivateAgentWithoutEntry(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, "provider_watch_thread_private_denied", now)

	agentID := "agt_provider_watch_thread_denied"
	agentMemberID := "mem_provider_watch_thread_denied"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "private-thread-denied", "private", false, now)
	mustExec(t, pool, `UPDATE scope_memberships SET removed_at = $3, updated_at = $3 WHERE scope_id = $1 AND member_id = $2`, fixture.AllChannelID, agentMemberID, now)

	threadShortID := "abc12345"
	seedProviderWatchReplyThread(t, pool, fixture, "chn_provider_watch_thread_denied_reply", threadShortID, now)

	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	_, err := appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "#all:"+threadShortID)
	require.ErrorIs(t, err, ErrForbidden)
	_, err = ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, "chn_provider_watch_thread_denied_reply")
	require.ErrorIs(t, err, ErrForbidden)
}

func TestProviderWatchWebhookRespectsEventFiltersAndIdempotency(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, "provider_watch_filters", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_filters", now)

	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)

	_, err = appStore.Provider().SaveGitHubAppConfigForWorkspaceAdmin(ctx, fixture.UserID, fixture.WorkspaceID, provider.UpsertGitHubAppConfigParams{
		AppID:         12345,
		AppSlug:       "agentteam-test",
		AppName:       "AgentTeam Test",
		HTMLURL:       "https://github.com/apps/agentteam-test",
		ClientID:      "Iv1.test",
		ClientSecret:  "client-secret",
		PrivateKeyPEM: "-----BEGIN RSA PRIVATE KEY-----\nkey\n-----END RSA PRIVATE KEY-----",
		WebhookSecret: "github-webhook-secret",
		OwnerLogin:    "acme",
		OwnerID:       67890,
	})
	require.NoError(t, err)

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{
		ActorType: "human",
		ActorID:   fixture.UserID,
		MemberID:  fixture.MemberID,
	}

	resolved, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/99",
		ResourceKey:    "github:acme/agentteam#99",
		EventFilters:   map[string]any{"events": []any{" reviews ", "COMMENT", "comments"}},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)
	duplicate, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/99",
		ResourceKey:    "github:acme/agentteam#99",
		EventFilters:   map[string]any{"events": "comments,reviews"},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)
	require.True(t, duplicate.AlreadyExists)
	require.Equal(t, resolved.Subscription.ID, duplicate.Subscription.ID)
	require.JSONEq(t, `{"events":["comments","reviews"]}`, string(resolved.Subscription.EventFilters))

	commentsWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/42",
		ResourceKey:    "github:acme/agentteam#42",
		EventFilters:   map[string]any{"events": []string{"comments"}},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)
	reviewsWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/42",
		ResourceKey:    "github:acme/agentteam#42",
		EventFilters:   map[string]any{"events": []string{"reviews"}},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)
	require.NotEqual(t, commentsWatch.Subscription.ID, reviewsWatch.Subscription.ID)

	commentPayload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "acme/agentteam"},
	  "issue": {
	    "number": 42,
	    "html_url": "https://github.com/acme/agentteam/issues/42",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/42"}
	  },
	  "comment": {
	    "id": 10,
	    "body": "comment event",
	    "html_url": "https://github.com/acme/agentteam/issues/42#issuecomment-10",
	    "user": {"login": "commenter"}
	  },
	  "sender": {"login": "commenter"}
	}`)
	commentHeaders := http.Header{}
	commentHeaders.Set("X-GitHub-Event", "issue_comment")
	commentHeaders.Set("X-GitHub-Delivery", "delivery-provider-watch-filter-comment")
	commentHeaders.Set("X-Hub-Signature-256", githubWebhookSignature(commentPayload, "github-webhook-secret"))

	commentResult, err := appStore.HandleAppWebhook(ctx, "github", commentHeaders, commentPayload)
	require.NoError(t, err)
	require.Equal(t, 1, commentResult.MatchedSubscriptionCount)
	require.Len(t, commentResult.Deliveries, 1)
	require.Contains(t, commentResult.Deliveries[0].Message.Content, "received a PR comment")
	require.Nil(t, commentResult.Deliveries[0].Message.Metadata)
	requireProviderWatchDeliverySource(t, pool, commentResult.Deliveries[0], commentsWatch.Subscription.ID)
	requireServiceEventDeliveryEffect(t, pool, commentsWatch.Subscription.ID, "succeeded", "message", commentResult.Deliveries[0].MessageID, commentResult.Deliveries[0].MessageID)

	reviewPayload := []byte(`{
	  "action": "submitted",
	  "installation": {"id": 789},
	  "repository": {"full_name": "acme/agentteam"},
	  "pull_request": {"number": 42, "html_url": "https://github.com/acme/agentteam/pull/42"},
	  "review": {
	    "id": 11,
	    "state": "approved",
	    "body": "review event",
	    "html_url": "https://github.com/acme/agentteam/pull/42#pullrequestreview-11",
	    "user": {"login": "reviewer"}
	  }
	}`)
	reviewHeaders := http.Header{}
	reviewHeaders.Set("X-GitHub-Event", "pull_request_review")
	reviewHeaders.Set("X-GitHub-Delivery", "delivery-provider-watch-filter-review")
	reviewHeaders.Set("X-Hub-Signature-256", githubWebhookSignature(reviewPayload, "github-webhook-secret"))

	reviewResult, err := appStore.HandleAppWebhook(ctx, "github", reviewHeaders, reviewPayload)
	require.NoError(t, err)
	require.Equal(t, 1, reviewResult.MatchedSubscriptionCount)
	require.Len(t, reviewResult.Deliveries, 1)
	require.Contains(t, reviewResult.Deliveries[0].Message.Content, "received a PR review")
	require.Nil(t, reviewResult.Deliveries[0].Message.Metadata)
	requireProviderWatchDeliverySource(t, pool, reviewResult.Deliveries[0], reviewsWatch.Subscription.ID)
	requireServiceEventDeliveryEffect(t, pool, reviewsWatch.Subscription.ID, "succeeded", "message", reviewResult.Deliveries[0].MessageID, reviewResult.Deliveries[0].MessageID)

	var serviceMessageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE author_member_id = $1
  AND content LIKE '%github:acme/agentteam#42%'
`, installation.MemberID).Scan(&serviceMessageCount))
	require.Equal(t, 2, serviceMessageCount)
}

func TestGiteaProviderWatchDeliversBaseBranchPushToMatchingPRWatches(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, "provider_watch_gitea_base_push", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_gitea_base_push", now)

	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}

	mainWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{"events": []string{"base_branch_push"}},
		Metadata:       map[string]any{"base_branch": "main"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/43",
		ResourceKey:    "gitea:acme/agentteam#43",
		EventFilters:   map[string]any{"events": []string{"base_branch_push"}},
		Metadata:       map[string]any{"base_branch": "release-1.0"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/44",
		ResourceKey:    "gitea:acme/agentteam#44",
		EventFilters:   map[string]any{},
		Metadata:       map[string]any{"base_branch": "main"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/46",
		ResourceKey:    "gitea:acme/agentteam#46",
		EventFilters:   map[string]any{"events": []string{"status"}},
		Metadata:       map[string]any{"base_branch": "main"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/47",
		ResourceKey:    "gitea:acme/agentteam#47",
		EventFilters:   map[string]any{"events": []string{"comments"}},
		Metadata:       map[string]any{"base_branch": "main"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	defaultBranchFallback, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/45",
		ResourceKey:    "gitea:acme/agentteam#45",
		EventFilters:   map[string]any{"events": []string{"base_branch_push"}},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	payload := []byte(`{
	  "ref": "refs/heads/main",
	  "before": "base-old",
	  "after": "base-new",
	  "compare_url": "https://gitea.example.com/acme/agentteam/compare/base-old...base-new",
	  "repository": {
	    "full_name": "acme/agentteam",
	    "name": "agentteam",
	    "default_branch": "main",
	    "owner": {"login": "acme"}
	  },
	  "sender": {"login": "maintainer"}
	}`)
	headers := http.Header{}
	headers.Set("X-Gitea-Event", "push")
	headers.Set("X-Gitea-Delivery", "delivery-gitea-base-branch-push")
	headers.Set("X-Gitea-Signature", giteaWebhookSignature(payload, "gitea-webhook-secret"))

	result, err := appStore.HandleAppWebhook(ctx, "gitea", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 2, result.MatchedSubscriptionCount)
	require.Len(t, result.Deliveries, 2)

	deliveredWatchIDs := map[string]bool{}
	for _, delivery := range result.Deliveries {
		require.Contains(t, delivery.Message.Content, "pushed to base branch main")
		require.Contains(t, delivery.Message.Content, "may now be out-of-date")
		require.Nil(t, delivery.Message.Metadata)
		deliveredWatchIDs[requireProviderWatchDeliverySource(t, pool, delivery, "")] = true
	}
	require.True(t, deliveredWatchIDs[mainWatch.Subscription.ID])
	require.True(t, deliveredWatchIDs[defaultBranchFallback.Subscription.ID])

	duplicate, err := appStore.HandleAppWebhook(ctx, "gitea", headers, payload)
	require.NoError(t, err)
	require.True(t, duplicate.Duplicate)
}

func TestGiteaProviderWatchDeliversCIStatusToMatchingHeadSHAWatch(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, "provider_watch_gitea_ci_status", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_gitea_ci_status", now)

	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}

	statusWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{},
		Metadata:       map[string]any{"head_sha": "head-sha"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/43",
		ResourceKey:    "gitea:acme/agentteam#43",
		EventFilters:   map[string]any{},
		Metadata:       map[string]any{"head_sha": "other-sha"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/44",
		ResourceKey:    "gitea:acme/agentteam#44",
		EventFilters:   map[string]any{"events": []string{"comments"}},
		Metadata:       map[string]any{"head_sha": "head-sha"},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	payload := []byte(`{
	  "sha": "head-sha",
	  "state": "failure",
	  "context": "ci/build",
	  "target_url": "https://ci.example.com/build/1",
	  "description": "build failed",
	  "updated_at": "2026-05-18T11:12:13Z",
	  "repository": {
	    "full_name": "acme/agentteam",
	    "name": "agentteam",
	    "owner": {"login": "acme"}
	  },
	  "sender": {"login": "ci-bot"}
	}`)
	headers := http.Header{}
	headers.Set("X-Gitea-Event", "status")
	headers.Set("X-Gitea-Delivery", "delivery-gitea-ci-status")
	headers.Set("X-Gitea-Signature", giteaWebhookSignature(payload, "gitea-webhook-secret"))

	result, err := appStore.HandleAppWebhook(ctx, "gitea", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 1, result.MatchedSubscriptionCount)
	require.Len(t, result.Deliveries, 1)
	require.Contains(t, result.Deliveries[0].Message.Content, "reported PR CI status (failure)")
	require.Contains(t, result.Deliveries[0].Message.Content, "gitea:acme/agentteam#42")
	require.Contains(t, result.Deliveries[0].Message.Content, "https://ci.example.com/build/1")
	require.Contains(t, result.Deliveries[0].Message.Content, "> build failed")
	require.Nil(t, result.Deliveries[0].Message.Metadata)
	requireProviderWatchDeliverySource(t, pool, result.Deliveries[0], statusWatch.Subscription.ID)

	successPayload := []byte(`{
	  "sha": "head-sha",
	  "state": "success",
	  "context": "ci/build",
	  "target_url": "https://ci.example.com/build/2",
	  "description": "build passed",
	  "updated_at": "2026-05-18T11:13:13Z",
	  "repository": {
	    "full_name": "acme/agentteam",
	    "name": "agentteam",
	    "owner": {"login": "acme"}
	  },
	  "sender": {"login": "ci-bot"}
	}`)
	successHeaders := http.Header{}
	successHeaders.Set("X-Gitea-Event", "status")
	successHeaders.Set("X-Gitea-Delivery", "delivery-gitea-ci-status-success")
	successHeaders.Set("X-Gitea-Signature", giteaWebhookSignature(successPayload, "gitea-webhook-secret"))

	successResult, err := appStore.HandleAppWebhook(ctx, "gitea", successHeaders, successPayload)
	require.NoError(t, err)
	require.Zero(t, successResult.MatchedSubscriptionCount)
	require.Empty(t, successResult.Deliveries)
}

func TestGiteaProviderWatchDeliversWorkflowRunStatusToPRWatch(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, "provider_watch_gitea_workflow_run", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_gitea_workflow_run", now)

	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}

	workflowWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{"events": []string{"status"}},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{"events": []string{"comments"}},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	payload := []byte(`{
	  "action": "completed",
	  "workflow": {"name": "CI"},
	  "workflow_run": {
	    "id": 1,
	    "name": "build",
	    "status": "completed",
	    "conclusion": "failure",
	    "html_url": "https://gitea.example.com/acme/agentteam/actions/runs/1",
	    "head_sha": "head-sha"
	  },
	  "repository": {
	    "full_name": "acme/agentteam",
	    "name": "agentteam",
	    "owner": {"login": "acme"}
	  },
	  "pull_request": {
	    "number": 42,
	    "html_url": "https://gitea.example.com/acme/agentteam/pulls/42"
	  },
	  "sender": {"login": "gitea-actions"}
	}`)
	headers := http.Header{}
	headers.Set("X-Gitea-Event", "workflow_run")
	headers.Set("X-Gitea-Delivery", "delivery-gitea-workflow-run")
	headers.Set("X-Gitea-Signature", giteaWebhookSignature(payload, "gitea-webhook-secret"))

	result, err := appStore.HandleAppWebhook(ctx, "gitea", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 1, result.MatchedSubscriptionCount)
	require.Len(t, result.Deliveries, 1)
	require.Contains(t, result.Deliveries[0].Message.Content, "reported PR workflow status (failure)")
	require.Contains(t, result.Deliveries[0].Message.Content, "gitea:acme/agentteam#42")
	require.Contains(t, result.Deliveries[0].Message.Content, "https://gitea.example.com/acme/agentteam/actions/runs/1")
	require.Nil(t, result.Deliveries[0].Message.Metadata)
	requireProviderWatchDeliverySource(t, pool, result.Deliveries[0], workflowWatch.Subscription.ID)
}

func TestGiteaProviderWatchAutoUnwatchesAfterClosedPRDelivery(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, "provider_watch_gitea_closed_auto", now)
	reviewChannelID := seedProviderWatchReviewChannel(t, pool, fixture, "provider_watch_gitea_closed_auto", now)

	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, reviewChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	watch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	payload := []byte(`{
	  "action": "closed",
	  "repository": {
	    "full_name": "acme/agentteam",
	    "name": "agentteam",
	    "owner": {"login": "acme"}
	  },
	  "pull_request": {
	    "number": 42,
	    "html_url": "https://gitea.example.com/acme/agentteam/pulls/42",
	    "merged": true,
	    "closed_at": "2026-05-19T01:02:03Z",
	    "merged_at": "2026-05-19T01:02:03Z",
	    "base": {"ref": "main", "sha": "base-sha"},
	    "head": {"ref": "feature", "sha": "head-sha"}
	  },
	  "sender": {"login": "maintainer"}
	}`)
	headers := http.Header{}
	headers.Set("X-Gitea-Event", "pull_request")
	headers.Set("X-Gitea-Delivery", "delivery-gitea-pr-closed-auto-unwatch")
	headers.Set("X-Gitea-Signature", giteaWebhookSignature(payload, "gitea-webhook-secret"))

	result, err := appStore.HandleAppWebhook(ctx, "gitea", headers, payload)
	require.NoError(t, err)
	require.Equal(t, 1, result.MatchedSubscriptionCount)
	require.Len(t, result.Deliveries, 1)
	require.Contains(t, result.Deliveries[0].Message.Content, "closed a PR (merged)")
	require.Contains(t, result.Deliveries[0].Message.Content, "gitea:acme/agentteam#42")
	require.Nil(t, result.Deliveries[0].Message.Metadata)
	requireProviderWatchDeliverySource(t, pool, result.Deliveries[0], watch.Subscription.ID)

	requireProviderWatchLifecycleState(t, pool, fixture.WorkspaceID, watch.Subscription.ID, ProviderWatchStatusCompleted, false, true)

	watches, err := appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, actor, installation.ID, "", "", "", ListServiceEventSubscriptionsOptions{})
	require.NoError(t, err)
	require.Empty(t, watches)

	recreated, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	require.False(t, recreated.AlreadyExists)
	require.NotEqual(t, watch.Subscription.ID, recreated.Subscription.ID)
	require.Equal(t, ProviderWatchStatusActive, recreated.Subscription.Status)
}

func TestPauseResumeAndRemoveProviderWatchUseSameRecord(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, "provider_watch_lifecycle", now)
	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://gitea.example.com/acme/agentteam/pulls/7",
		ResourceKey:  "gitea:acme/agentteam#7",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)

	loaded, err := appStore.GetProviderWatchForUser(ctx, fixture.UserID, fixture.WorkspaceID, result.Subscription.ID)
	require.NoError(t, err)
	require.Equal(t, result.Subscription.ID, loaded.ID)

	paused, err := appStore.PauseProviderWatchForUser(ctx, fixture.UserID, fixture.WorkspaceID, result.Subscription.ID)
	require.NoError(t, err)
	require.Equal(t, result.Subscription.ID, paused.Watch.ID)
	require.Equal(t, ProviderWatchStatusPaused, paused.Watch.Status)
	require.NotNil(t, paused.Watch.PausedAt)

	resumed, err := appStore.ResumeProviderWatchForUser(ctx, fixture.UserID, fixture.WorkspaceID, result.Subscription.ID)
	require.NoError(t, err)
	require.Equal(t, result.Subscription.ID, resumed.Watch.ID)
	require.Equal(t, ProviderWatchStatusActive, resumed.Watch.Status)
	require.Nil(t, resumed.Watch.PausedAt)

	removed, err := appStore.RemoveProviderWatchForUser(ctx, fixture.UserID, fixture.WorkspaceID, result.Subscription.ID)
	require.NoError(t, err)
	require.Equal(t, result.Subscription.ID, removed.Watch.ID)
	require.NotNil(t, removed.Watch.RemovedAt)

	watches, err := appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}, installation.ID, "", "", "", ListServiceEventSubscriptionsOptions{})
	require.NoError(t, err)
	require.Empty(t, watches)
}

func TestProviderWatchByIDRejectsNonProviderPullRequestSubscription(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, "provider_watch_non_pr_subscription", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor: ProviderWatchActor{
			ActorType: "human",
			ActorID:   fixture.UserID,
			MemberID:  fixture.MemberID,
		},
		Target:       target,
		ResourceRef:  "https://github.com/acme/agentteam/pull/801",
		ResourceKey:  "github:acme/agentteam#801",
		EventFilters: map[string]any{},
		CreatedVia:   "app_page",
	})
	require.NoError(t, err)

	nonPRID := "svcesub_provider_watch_non_pr"
	mustExec(t, pool, `
	INSERT INTO service_event_subscriptions (
	    id, workspace_id, service_definition_id, service_installation_id,
	    service_member_id, service_connection_id, service_connection_ref,
    service_target_binding_id, service_kind, subscription_kind, target_kind,
    target_ref, delivery_scope_id, permission_check_scope_id, owner_member_id,
    authorizer_member_id, created_by_actor_type, created_by_actor_id,
    created_via, external_resource_type, external_resource_key,
    external_resource_ref, external_resource_url, event_filter_key, event_filters,
    delivery_policy, notify_principal, status, active_subscription_key,
    last_failure_detail, metadata, created_at, updated_at
)
	SELECT
	    $2, workspace_id, service_definition_id, service_installation_id,
	    service_member_id, service_connection_id, service_connection_ref,
	    service_target_binding_id, service_kind, subscription_kind, target_kind,
    target_ref, delivery_scope_id, permission_check_scope_id, owner_member_id,
    authorizer_member_id, created_by_actor_type, created_by_actor_id,
    created_via, 'issue', 'github:acme/agentteam#801',
    external_resource_ref, external_resource_url, event_filter_key, event_filters,
    delivery_policy, notify_principal, status, 'non-pr-active-key',
    '', metadata, created_at, updated_at
FROM service_event_subscriptions
WHERE id = $1
`, result.Subscription.ID, nonPRID)

	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}, nonPRID, "manual")
	require.ErrorIs(t, err, storeerr.ErrNotFound)

	_, err = appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}, ProviderWatchProviderGitHub, installation.ID, nonPRID, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "watch_not_found")
}

func TestServiceEventSubscriptionConnectionBlockAndUnblock(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, "provider_watch_connection_block", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	activeWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/42",
		ResourceKey:    "github:acme/agentteam#42",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	pausedWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/43",
		ResourceKey:    "github:acme/agentteam#43",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, actor, pausedWatch.Subscription.ID, "manual pause")
	require.NoError(t, err)

	count, err := appStore.BlockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "missing-connection")
	require.NoError(t, err)
	require.Zero(t, count)

	count, err = appStore.BlockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "789")
	require.NoError(t, err)
	require.Equal(t, 2, count)
	requireProviderWatchStatus(t, ctx, pool, activeWatch.Subscription.ID, ProviderWatchStatusBlocked, ProviderWatchStatusActive)
	requireProviderWatchStatus(t, ctx, pool, pausedWatch.Subscription.ID, ProviderWatchStatusBlocked, ProviderWatchStatusPaused)
	blockedPayload := []byte(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "acme/agentteam"},
	  "issue": {
	    "number": 42,
	    "html_url": "https://github.com/acme/agentteam/issues/42",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/42"}
	  },
	  "comment": {
	    "id": 4242,
	    "body": "blocked delivery should not write",
	    "html_url": "https://github.com/acme/agentteam/issues/42#issuecomment-4242",
	    "user": {"login": "commenter"}
	  }
	}`)
	blockedHeaders := http.Header{}
	blockedHeaders.Set("X-GitHub-Event", "issue_comment")
	blockedHeaders.Set("X-GitHub-Delivery", "delivery-service-event-blocked")
	blockedHeaders.Set("X-Hub-Signature-256", githubWebhookSignature(blockedPayload, "github-webhook-secret"))
	blockedResult, err := appStore.HandleAppWebhook(ctx, "github", blockedHeaders, blockedPayload)
	require.NoError(t, err)
	require.Zero(t, blockedResult.MatchedSubscriptionCount)
	require.Empty(t, blockedResult.Deliveries)
	requireServiceEventDeliveryCount(t, pool, activeWatch.Subscription.ID, 0)

	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, actor, activeWatch.Subscription.ID, "manual pause")
	require.ErrorContains(t, err, "service_event_subscription_blocked")
	_, err = appStore.ResumeServiceEventSubscription(ctx, fixture.WorkspaceID, actor, pausedWatch.Subscription.ID)
	require.ErrorContains(t, err, "service_event_subscription_blocked")

	count, err = appStore.BlockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "789")
	require.NoError(t, err)
	require.Zero(t, count)

	mustExec(t, pool, `
UPDATE service_connections
SET status = 'disabled', disabled_at = $3, updated_at = $3
WHERE service_installation_id = $1 AND external_subject_ref = $2
`, installation.ID, "789", now.Add(time.Second))
	count, err = appStore.UnblockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "789")
	require.NoError(t, err)
	require.Zero(t, count)
	requireProviderWatchStatus(t, ctx, pool, activeWatch.Subscription.ID, ProviderWatchStatusBlocked, ProviderWatchStatusActive)
	requireProviderWatchStatus(t, ctx, pool, pausedWatch.Subscription.ID, ProviderWatchStatusBlocked, ProviderWatchStatusPaused)

	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})
	count, err = appStore.UnblockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "789")
	require.NoError(t, err)
	require.Equal(t, 2, count)
	requireProviderWatchStatus(t, ctx, pool, activeWatch.Subscription.ID, ProviderWatchStatusActive, "")
	requireProviderWatchStatus(t, ctx, pool, pausedWatch.Subscription.ID, ProviderWatchStatusPaused, "")

	count, err = appStore.UnblockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "github", "789")
	require.NoError(t, err)
	require.Zero(t, count)
}

func TestGiteaServiceEventSubscriptionConnectionBlockNormalizesConnectionRef(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, "gitea_provider_watch_connection_block", now)
	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme/agentteam", "Acme/AgentTeam")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/Acme/AgentTeam/pulls/42",
		ResourceKey:    "gitea:acme/agentteam#42",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	require.Equal(t, "acme/agentteam", result.Subscription.ProviderConnectionRef)

	count, err := appStore.BlockServiceEventSubscriptions(ctx, fixture.WorkspaceID, "gitea", "Acme/AgentTeam")
	require.NoError(t, err)
	require.Equal(t, 1, count)
	requireProviderWatchStatus(t, ctx, pool, result.Subscription.ID, ProviderWatchStatusBlocked, ProviderWatchStatusActive)

	currentConfig, err := appStore.Provider().GetGiteaAppConfigForUser(ctx, fixture.UserID, fixture.WorkspaceID)
	require.NoError(t, err)
	_, err = appStore.SaveGiteaAppConfigForWorkspaceAdmin(ctx, fixture.UserID, fixture.WorkspaceID, provider.UpsertGiteaAppConfigParams{
		InstanceURL:   currentConfig.InstanceURL,
		APIToken:      "",
		WebhookSecret: "",
		DisplayName:   currentConfig.DisplayName,
	})
	require.NoError(t, err)
	requireProviderWatchStatus(t, ctx, pool, result.Subscription.ID, ProviderWatchStatusActive, "")

	_, err = appStore.UninstallAppInstallationForUser(ctx, fixture.UserID, fixture.WorkspaceID, installation.ID)
	require.NoError(t, err)
	var subscriptionCount int
	require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM service_event_subscriptions WHERE id = $1`, result.Subscription.ID).Scan(&subscriptionCount))
	require.Zero(t, subscriptionCount)
	providerConfig, found, err := appStore.Provider().LoadGiteaAppConfigForWorkspace(ctx, fixture.WorkspaceID)
	require.NoError(t, err)
	require.False(t, found)
	require.Empty(t, providerConfig)
}

func TestAgentAppAuthorizationLifecycleScopesAndRevocation(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, "agent_app_authorization", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	machineID := "mac_agent_app_authorization"
	agentID := "agt_agent_app_authorization"
	agentMemberID := "mem_agent_app_authorization_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Auth Agent", now)

	defaultActive, err := appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Len(t, defaultActive, 1)
	require.Empty(t, defaultActive[0].ID)
	require.Equal(t, AgentAppAuthorizationStatusActive, defaultActive[0].Status)
	require.Equal(t, fixture.MemberID, defaultActive[0].AuthorizedByMemberID)
	require.ElementsMatch(t, []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite}, defaultActive[0].Scopes)

	defaultChecked, err := appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.NoError(t, err)
	require.Empty(t, defaultChecked.ID)
	require.Equal(t, fixture.MemberID, defaultChecked.AuthorizedByMemberID)

	authorization, err := appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{"provider_watch.read", "provider_watch.write", "provider_watch.read"},
		Metadata:             map[string]any{"source": "settings"},
	})
	require.NoError(t, err)
	require.Equal(t, AgentAppAuthorizationStatusActive, authorization.Status)
	require.Equal(t, agentMemberID, authorization.AgentMemberID)
	require.Equal(t, installation.ID, authorization.AppInstallationID)
	require.Equal(t, installation.MemberID, authorization.AppMemberID)
	require.Equal(t, "github", authorization.AppSlug)
	require.ElementsMatch(t, []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite}, authorization.Scopes)

	active, err := appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Len(t, active, 1)
	require.Equal(t, authorization.ID, active[0].ID)

	checked, err := appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{"provider_watch.read"})
	require.NoError(t, err)
	require.Equal(t, authorization.ID, checked.ID)

	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{"provider_watch.delete"})
	require.Error(t, err)

	disabled, err := appStore.DisableAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, "disabled by owner")
	require.NoError(t, err)
	require.Equal(t, authorization.ID, disabled.ID)
	require.Equal(t, AgentAppAuthorizationStatusDisabled, disabled.Status)
	require.NotNil(t, disabled.DisabledAt)
	require.Greater(t, disabled.PolicyVersion, authorization.PolicyVersion)

	active, err = appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Empty(t, active)
	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{"provider_watch.read"})
	require.Error(t, err)

	reauthorized, err := appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{"provider_watch.read"},
	})
	require.NoError(t, err)
	require.Equal(t, authorization.ID, reauthorized.ID)
	require.Equal(t, AgentAppAuthorizationStatusActive, reauthorized.Status)
	require.Greater(t, reauthorized.PolicyVersion, disabled.PolicyVersion)
	require.Equal(t, []string{ServiceEventSubscriptionScopeRead}, reauthorized.Scopes)
}

func TestAgentAppAuthorizationDefaultAllowRequiresEnabledConnection(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, "agent_app_default_connection", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)

	machineID := "mac_agent_app_default_connection"
	agentID := "agt_agent_app_default_connection"
	agentMemberID := "mem_agent_app_default_connection_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Default Connection Agent", now)

	active, err := appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Empty(t, active)
	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeRead})
	require.ErrorContains(t, err, "app_connection_not_configured")

	connection := upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})
	active, err = appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Len(t, active, 1)
	require.Empty(t, active[0].ID)

	mustExec(t, pool, `UPDATE service_connections SET status = 'disabled', disabled_at = $2, updated_at = $2 WHERE id = $1`, connection.ID, now)
	active, err = appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Empty(t, active)
	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeRead})
	require.ErrorContains(t, err, "app_connection_not_configured")
}

func TestAgentAppAuthorizationViewsDefaultAllowAndExplicitDisable(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, "agent_app_views_default", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})

	machineID := "mac_agent_app_views_default"
	agentID := "agt_agent_app_views_default"
	agentMemberID := "mem_agent_app_views_default_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "View Agent", now)

	installationView, found, err := appStore.loadAgentAuthorizationInstallationView(ctx, fixture.WorkspaceID, installation.ID)
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, installation.ID, installationView.InstallationID)

	views, err := appStore.listAgentAppAuthorizationViews(ctx, fixture.WorkspaceID, agentID, "", "")
	require.NoError(t, err)
	require.Len(t, views, 1)
	require.Empty(t, views[0].AuthorizationID)
	require.Equal(t, AgentAppAuthorizationStatusActive, views[0].AuthorizationStatus)
	require.Equal(t, int64(1), views[0].PolicyVersion)
	require.ElementsMatch(t, []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite}, views[0].Scopes)

	disabled, err := appStore.DisableAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, "disabled by owner")
	require.NoError(t, err)
	require.NotEmpty(t, disabled.ID)
	require.Equal(t, AgentAppAuthorizationStatusDisabled, disabled.Status)

	views, err = appStore.listAgentAppAuthorizationViews(ctx, fixture.WorkspaceID, agentID, "", "")
	require.NoError(t, err)
	require.Len(t, views, 1)
	require.Equal(t, disabled.ID, views[0].AuthorizationID)
	require.Equal(t, AgentAppAuthorizationStatusDisabled, views[0].AuthorizationStatus)
	require.Equal(t, "disabled by owner", views[0].DisabledReason)
}

func TestAgentAppAuthorizationViewsMatchPolicyForDisabledConnectionAndRemovedRows(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, "agent_app_views_policy", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	connection := upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, "789", "acme", map[string]any{"webhook_secret": "github-webhook-secret"})

	machineID := "mac_agent_app_views_policy"
	agentID := "agt_agent_app_views_policy"
	agentMemberID := "mem_agent_app_views_policy_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Policy View Agent", now)

	authorization, err := appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	require.Equal(t, AgentAppAuthorizationStatusActive, authorization.Status)

	views, err := appStore.listAgentAppAuthorizationViews(ctx, fixture.WorkspaceID, agentID, "", "")
	require.NoError(t, err)
	require.Len(t, views, 1)
	require.Equal(t, authorization.ID, views[0].AuthorizationID)
	require.Equal(t, AgentAppAuthorizationStatusActive, views[0].AuthorizationStatus)

	mustExec(t, pool, `UPDATE service_connections SET status = 'disabled', disabled_at = $2, updated_at = $2 WHERE id = $1`, connection.ID, now)
	views, err = appStore.listAgentAppAuthorizationViews(ctx, fixture.WorkspaceID, agentID, "", "")
	require.NoError(t, err)
	require.Len(t, views, 1)
	require.Equal(t, authorization.ID, views[0].AuthorizationID)
	require.Equal(t, AgentAppAuthorizationStatusDisabled, views[0].AuthorizationStatus)
	require.Equal(t, "app_connection_not_configured", views[0].DisabledReason)
	active, err := appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Empty(t, active)
	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeRead})
	require.ErrorContains(t, err, "app_connection_not_configured")

	mustExec(t, pool, `UPDATE service_connections SET status = 'active', disabled_at = NULL, updated_at = $2 WHERE id = $1`, connection.ID, now)
	removedAt := time.Now().UTC()
	mustExec(t, pool, `
UPDATE service_principal_authorizations
SET removed_at = $2, updated_at = $2
WHERE id = $1
`, authorization.ID, removedAt)
	views, err = appStore.listAgentAppAuthorizationViews(ctx, fixture.WorkspaceID, agentID, "", "")
	require.NoError(t, err)
	require.Len(t, views, 1)
	require.Equal(t, authorization.ID, views[0].AuthorizationID)
	require.Equal(t, AgentAppAuthorizationStatusDisabled, views[0].AuthorizationStatus)
	require.Equal(t, "app_authorization_removed", views[0].DisabledReason)
	active, err = appStore.ListActiveAgentAppAuthorizations(ctx, fixture.WorkspaceID, agentID)
	require.NoError(t, err)
	require.Empty(t, active)
	_, err = appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeRead})
	require.ErrorContains(t, err, "app_not_authorized_for_agent")
}

func TestCreateServiceEventSubscriptionAllowsDefaultAgentAuthorizationAndHonorsScopeOverrides(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, "provider_watch_agent_authz", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	machineID := "mac_provider_watch_agent_authz"
	agentID := "agt_provider_watch_agent_authz"
	agentMemberID := "mem_provider_watch_agent_authz_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Watch Agent", now)

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, fixture.AllChannelID)
	require.NoError(t, err)
	privateChannelID := "chn_provider_watch_agent_authz_private"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, privateChannelID, "authz-private", "", "private", fixture.MemberID, fixture.MemberID, now)
	authenticatedAgent := agent.AuthenticatedAgent{
		AgentID:     agentID,
		WorkspaceID: fixture.WorkspaceID,
		MemberID:    agentMemberID,
	}
	_, err = appStore.ResolveServiceEventSubscriptionTargetForAgentTarget(ctx, authenticatedAgent, "#authz-private")
	require.Error(t, err)

	defaultAuthorization, err := appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.NoError(t, err)
	require.Empty(t, defaultAuthorization.ID)
	require.Equal(t, fixture.MemberID, defaultAuthorization.AuthorizedByMemberID)
	agentActor := ProviderWatchActor{
		ActorType:        "agent",
		ActorID:          agentID,
		MemberID:         agentMemberID,
		AuthorizerUserID: defaultAuthorization.AuthorizedByMemberID,
	}

	defaultResult, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/43",
		ResourceKey:    "github:acme/agentteam#43",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "mcp",
	})
	require.NoError(t, err)
	require.Equal(t, ProviderWatchStatusActive, defaultResult.Subscription.Status)
	require.Equal(t, fixture.UserID, defaultResult.Subscription.OwnerUserID)
	require.Equal(t, fixture.MemberID, defaultResult.Subscription.AuthorizerUserID)
	var defaultOwnerMemberID string
	var defaultAuthorizerMemberID string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT owner_member_id, authorizer_member_id
FROM service_event_subscriptions
WHERE id = $1
`, defaultResult.Subscription.ID).Scan(&defaultOwnerMemberID, &defaultAuthorizerMemberID))
	require.Equal(t, fixture.MemberID, defaultOwnerMemberID)
	require.Equal(t, fixture.MemberID, defaultAuthorizerMemberID)

	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead},
	})
	require.NoError(t, err)
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/44",
		ResourceKey:    "github:acme/agentteam#44",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "mcp",
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "token_scope_denied")

	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/44",
		ResourceKey:    "github:acme/agentteam#44",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "mcp",
	})
	require.NoError(t, err)
	require.Equal(t, ProviderWatchStatusActive, result.Subscription.Status)

	listed, err := appStore.ListServiceEventSubscriptions(ctx, fixture.WorkspaceID, agentActor, installation.ID, target.TargetKind, target.TargetRef, "", ListServiceEventSubscriptionsOptions{})
	require.NoError(t, err)
	require.Len(t, listed, 2)

	_, err = appStore.DisableAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, "disabled by owner")
	require.NoError(t, err)
	_, err = appStore.PauseServiceEventSubscription(ctx, fixture.WorkspaceID, agentActor, result.Subscription.ID, "manual")
	require.Error(t, err)
	require.Contains(t, err.Error(), "app_not_authorized_for_agent")
	_, err = appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/45",
		ResourceKey:    "github:acme/agentteam#45",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "mcp",
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "app_not_authorized_for_agent")
}

func TestProviderWatchWebhookSkipsAgentWatchAfterAuthorizationDisabled(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, "provider_watch_agent_disabled_delivery", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	machineID := "mac_provider_watch_agent_disabled_delivery"
	agentID := "agt_provider_watch_agent_disabled_delivery"
	agentMemberID := "mem_provider_watch_agent_disabled_delivery_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Watch Disabled Agent", now)

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, agentMemberID, fixture.AllChannelID)
	require.NoError(t, err)
	defaultAuthorization, err := appStore.CheckAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.NoError(t, err)
	agentActor := ProviderWatchActor{
		ActorType:        "agent",
		ActorID:          agentID,
		MemberID:         agentMemberID,
		AuthorizerUserID: defaultAuthorization.AuthorizedByMemberID,
	}

	result, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installation.ID,
		Actor:          agentActor,
		Target:         target,
		ResourceRef:    "https://github.com/acme/agentteam/pull/47",
		ResourceKey:    "github:acme/agentteam#47",
		EventFilters:   map[string]any{},
		CreatedVia:     "mcp",
	})
	require.NoError(t, err)
	require.Equal(t, ProviderWatchStatusActive, result.Subscription.Status)

	_, err = appStore.DisableAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, "disabled by owner")
	require.NoError(t, err)

	payload := providerWatchGitHubCommentPayload(47, "agent auth disabled")
	headers := http.Header{}
	headers.Set("X-GitHub-Event", "issue_comment")
	headers.Set("X-GitHub-Delivery", "delivery-provider-watch-agent-disabled")
	headers.Set("X-Hub-Signature-256", githubWebhookSignature(payload, "github-webhook-secret"))

	webhookResult, err := appStore.HandleAppWebhook(ctx, "github", headers, payload)
	require.NoError(t, err)
	require.False(t, webhookResult.Duplicate)
	require.Zero(t, webhookResult.MatchedSubscriptionCount)
	require.Empty(t, webhookResult.Deliveries)

	var messageCount int
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_messages
WHERE content LIKE '%github:acme/agentteam#47%'
`).Scan(&messageCount))
	require.Zero(t, messageCount)

	var failureCode string
	require.NoError(t, pool.QueryRow(ctx, `
SELECT COALESCE(last_failure_code, '')
FROM service_event_subscriptions
WHERE id = $1
`, result.Subscription.ID).Scan(&failureCode))
	require.Equal(t, "app_not_authorized_for_agent", failureCode)
}

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

	ctx := context.Background()
	pool := startAppRollbackDB(t, ctx)
	signer, err := capabilities.NewSigner([]byte("agent-app-scoped-token-signing-key"))
	require.NoError(t, err)
	appStore := New(pool, Options{AppCredentialStorageKey: testAppCredentialStorageKey(), AppCapabilitySigner: signer})
	now := time.Now().UTC()
	fixture := seedAppAdminFixture(t, pool, "agent_app_token", now)
	installation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "github", "789", "acme")

	machineID := "mac_agent_app_token"
	agentID := "agt_agent_app_token"
	agentMemberID := "mem_agent_app_token_bot"
	seedAppAgentParticipant(t, pool, fixture.WorkspaceID, fixture.UserID, fixture.MemberID, fixture.AllChannelID, machineID, agentID, agentMemberID, "Token Agent", now)

	issued, err := appStore.IssueAgentAppScopedToken(ctx, fixture.WorkspaceID, agentID, installation.ID, "session-1", time.Hour)
	require.NoError(t, err)
	require.NotEmpty(t, issued.Token)
	require.Empty(t, issued.Authorization.ID)
	require.Equal(t, AgentAppAuthorizationStatusActive, issued.Authorization.Status)
	require.Equal(t, int64(1), issued.Authorization.PolicyVersion)
	require.Equal(t, fixture.MemberID, issued.Authorization.AuthorizedByMemberID)
	require.ElementsMatch(t, []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite}, issued.Authorization.Scopes)

	authenticated, err := appStore.AuthenticateAgentAppScopedToken(ctx, fixture.WorkspaceID, agentID, agentMemberID, "github", issued.Token, []string{ServiceEventSubscriptionScopeWrite})
	require.NoError(t, err)
	require.Empty(t, authenticated.ID)

	_, err = appStore.AuthenticateAgentAppScopedToken(ctx, fixture.WorkspaceID, "agt_other", agentMemberID, "github", issued.Token, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "token_policy_revoked")
	_, err = appStore.AuthenticateAgentAppScopedToken(ctx, fixture.WorkspaceID, agentID, agentMemberID, "gitea", issued.Token, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "token_policy_revoked")

	_, err = appStore.DisableAgentAppAuthorization(ctx, fixture.WorkspaceID, agentID, installation.ID, "disabled")
	require.NoError(t, err)
	_, err = appStore.AuthenticateAgentAppScopedToken(ctx, fixture.WorkspaceID, agentID, agentMemberID, "github", issued.Token, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "app_not_authorized_for_agent")

	_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
		WorkspaceID:          fixture.WorkspaceID,
		AgentID:              agentID,
		AppInstallationID:    installation.ID,
		AuthorizedByMemberID: fixture.MemberID,
		Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
	})
	require.NoError(t, err)
	_, err = appStore.AuthenticateAgentAppScopedToken(ctx, fixture.WorkspaceID, agentID, agentMemberID, "github", issued.Token, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "token_policy_revoked")
}

func TestGetServiceEventSubscriptionForAppInstallationRejectsCrossAppSubscriptionID(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, "provider_watch_cross_app_id", now)
	githubInstallation, err := appStore.installAppForUserForTest(ctx, fixture.UserID, fixture.WorkspaceID, InstallAppParams{AppSlug: "github"})
	require.NoError(t, err)
	giteaInstallation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, githubInstallation, "github", "789", "acme")
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, giteaInstallation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	humanActor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	giteaWatch, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: giteaInstallation.ID,
		Actor:          humanActor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/17",
		ResourceKey:    "gitea:acme/agentteam#17",
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)

	agentID := "agt_provider_watch_cross_app"
	agentMemberID := "mem_provider_watch_cross_app"
	seedProviderWatchAgentEntry(t, pool, fixture, agentID, agentMemberID, "cross-app-agent", "private", false, now)
	for _, installationID := range []string{githubInstallation.ID, giteaInstallation.ID} {
		_, err = appStore.UpsertAgentAppAuthorization(ctx, UpsertAgentAppAuthorizationParams{
			WorkspaceID:          fixture.WorkspaceID,
			AgentID:              agentID,
			AppInstallationID:    installationID,
			AuthorizedByMemberID: fixture.MemberID,
			Scopes:               []string{ServiceEventSubscriptionScopeRead, ServiceEventSubscriptionScopeWrite},
		})
		require.NoError(t, err)
	}
	agentActor := ProviderWatchActor{ActorType: "agent", ActorID: agentID, MemberID: agentMemberID, AuthorizerUserID: fixture.MemberID}

	_, err = appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, agentActor, ProviderWatchProviderGitHub, githubInstallation.ID, giteaWatch.Subscription.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "watch_not_found")

	mustExec(t, pool, `
INSERT INTO service_event_subscriptions (
    id, workspace_id, service_definition_id, service_installation_id,
    service_member_id, service_connection_id, service_connection_ref,
    service_target_binding_id, service_kind, subscription_kind, target_kind,
    target_ref, delivery_scope_id, permission_check_scope_id, owner_member_id,
    authorizer_member_id, created_by_actor_type, created_by_actor_id,
    created_via, external_resource_type, external_resource_key,
    external_resource_ref, external_resource_url, event_filter_key, event_filters,
    delivery_policy, notify_principal, status, active_subscription_key,
    last_failure_detail, metadata, created_at, updated_at
)
SELECT
    'svcesub_provider_watch_cross_app_other_source', workspace_id, service_definition_id,
    service_installation_id, service_member_id, service_connection_id,
    service_connection_ref, service_target_binding_id, service_kind, subscription_kind,
    target_kind, target_ref, delivery_scope_id, permission_check_scope_id,
    owner_member_id, authorizer_member_id, created_by_actor_type, created_by_actor_id,
    created_via, 'issue', 'gitea:acme/agentteam/issues/1', external_resource_ref,
    external_resource_url, event_filter_key, event_filters, delivery_policy,
    notify_principal, status, 'non-pr-active-key', '', metadata, $2, $2
FROM service_event_subscriptions
WHERE id = $1
`, giteaWatch.Subscription.ID, now.Add(time.Second))
	_, err = appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, agentActor, ProviderWatchProviderGitea, giteaInstallation.ID, "svcesub_provider_watch_cross_app_other_source", []string{ServiceEventSubscriptionScopeWrite})
	require.ErrorContains(t, err, "watch_not_found")

	found, err := appStore.GetServiceEventSubscriptionForAppInstallation(ctx, fixture.WorkspaceID, agentActor, ProviderWatchProviderGitea, giteaInstallation.ID, giteaWatch.Subscription.ID, []string{ServiceEventSubscriptionScopeWrite})
	require.NoError(t, err)
	require.Equal(t, giteaWatch.Subscription.ID, found.ID)
	require.Equal(t, giteaInstallation.ID, found.AppInstallationID)
}

func TestCreateServiceEventSubscriptionRestoresDisabledTargetAppBinding(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, "provider_watch_restore_binding", now)
	installation := installGiteaForProviderWatchTests(t, ctx, appStore, fixture)
	seedProviderWatchReadyConnection(t, ctx, appStore, fixture, installation, "gitea", "acme", "acme")

	target, err := ResolveProviderWatchTargetForScope(ctx, pool, fixture.WorkspaceID, fixture.MemberID, fixture.AllChannelID)
	require.NoError(t, err)
	actor := ProviderWatchActor{ActorType: "human", ActorID: fixture.UserID, MemberID: fixture.MemberID}
	first, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/8",
		ResourceKey:    "gitea:acme/agentteam#8",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)

	mustExec(t, pool, `
UPDATE service_target_bindings
SET status = 'disabled', disabled_reason = 'temporarily disabled', updated_at = $3
WHERE workspace_id = $1 AND id = $2
`, fixture.WorkspaceID, first.BindingID, now.Add(time.Second))

	restored, err := appStore.CreateServiceEventSubscription(ctx, fixture.WorkspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "gitea",
		InstallationID: installation.ID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    "https://gitea.example.com/acme/agentteam/pulls/8",
		ResourceKey:    "gitea:acme/agentteam#8",
		EventFilters:   map[string]any{"comments": true},
		CreatedVia:     "app_command",
	})
	require.NoError(t, err)
	require.True(t, restored.AlreadyExists)
	require.Equal(t, first.BindingID, restored.BindingID)
	require.Equal(t, first.Subscription.ID, restored.Subscription.ID)
	restoredBinding, err := loadTargetAppBindingByID(ctx, pool, restored.BindingID)
	require.NoError(t, err)
	require.Equal(t, TargetAppBindingStatusActive, restoredBinding.Status)
	require.Empty(t, restoredBinding.DisabledReason)
}

func requireProviderWatchStatus(t *testing.T, ctx context.Context, queryer coredb.Queryer, watchID string, expectedStatus string, expectedPreBlockStatus string) {
	t.Helper()

	var status, failureCode string
	var metadata []byte
	require.NoError(t, queryer.QueryRow(ctx, `
SELECT status, COALESCE(last_failure_code, ''), COALESCE(metadata, '{}')
FROM service_event_subscriptions
WHERE id = $1
`, watchID).Scan(&status, &failureCode, &metadata))
	require.Equal(t, expectedStatus, status)
	decoded := decodeServiceEventSubscriptionJSONMap(metadata)
	if expectedStatus == ProviderWatchStatusBlocked {
		require.Equal(t, serviceEventSubscriptionFailureCodeProviderConnectionDown, failureCode)
		require.Equal(t, expectedPreBlockStatus, serviceEventSubscriptionMetadataString(decoded, "pre_block_status"))
		return
	}
	require.Empty(t, failureCode)
	require.Empty(t, serviceEventSubscriptionMetadataString(decoded, "pre_block_status"))
}

func seedProviderWatchReviewChannel(t *testing.T, pool coredb.QueryExecer, fixture appAdminFixture, suffix string, now time.Time) string {
	t.Helper()

	channelID := "chn_" + suffix + "_review"
	seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, channelID, suffix+"_review", "", "private", fixture.MemberID, fixture.MemberID, now)
	return channelID
}

func seedProviderWatchReadyConnection(t *testing.T, ctx context.Context, appStore *Store, fixture appAdminFixture, installation AppInstallationView, providerName string, externalSubjectRef string, displayLabel string) {
	t.Helper()

	seedProviderWatchProviderConfig(t, ctx, appStore, fixture, providerName)
	upsertAppConnectionForTests(t, ctx, appStore, installation.ID, fixture.UserID, externalSubjectRef, displayLabel, map[string]any{"webhook_secret": providerName + "-webhook-secret"})
}

type providerWatchFailSuccessCompleteDB struct {
	coredb.DB
}

func (db *providerWatchFailSuccessCompleteDB) BeginTx(ctx context.Context, opts coredb.TxOptions) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, opts)
	if err != nil {
		return nil, err
	}
	return &providerWatchFailSuccessCompleteTx{Tx: tx}, nil
}

type providerWatchFailSuccessCompleteTx struct {
	coredb.Tx
}

func (tx *providerWatchFailSuccessCompleteTx) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	if strings.Contains(query, "UPDATE service_event_deliveries") && strings.Contains(query, "status = $2") && len(args) > 1 && args[1] == string(runeffect.StatusSucceeded) {
		return testExecResult{}, errProviderWatchFinalizationFailed
	}
	return tx.Tx.Exec(ctx, query, args...)
}

func seedProviderWatchRateLimitEvents(t *testing.T, pool coredb.QueryExecer, workspaceID string, installationID string, actor ProviderWatchActor, action string, count int, now time.Time) {
	t.Helper()

	for index := 0; index < count; index++ {
		mustExec(t, pool, `
INSERT INTO service_action_rate_limit_events (
    id,
    workspace_id,
    owner_kind,
    owner_ref,
    service_installation_id,
    actor_type,
    actor_id,
    action,
    target_kind,
    target_ref,
    source_type,
    source_subject_key,
    correlation_id,
    created_at
) VALUES ($1, $2, 'app', $3, $3, $4, $5, $6, 'channel', 'chn_rate_limit', 'github.pull_request', 'github:acme/agentteam#rate-limit', $7, $8)
`, fmt.Sprintf("atrl_seed_%s_%d", action, index), workspaceID, installationID, actor.ActorType, actor.ActorID, action, fmt.Sprintf("seed-%d", index), now.Add(-10*time.Second))
	}
}

func createProviderWatchForRateLimitTest(t *testing.T, ctx context.Context, appStore *Store, workspaceID string, installationID string, actor ProviderWatchActor, target ProviderWatchTarget, prNumber string) ServiceEventSubscriptionRecord {
	t.Helper()

	result, err := appStore.CreateServiceEventSubscription(ctx, workspaceID, CreateServiceEventSubscriptionParams{
		Provider:       "github",
		InstallationID: installationID,
		Actor:          actor,
		Target:         target,
		ResourceRef:    fmt.Sprintf("https://github.com/acme/agentteam/pull/%s", prNumber),
		ResourceKey:    fmt.Sprintf("github:acme/agentteam#%s", prNumber),
		EventFilters:   map[string]any{},
		CreatedVia:     "app_page",
	})
	require.NoError(t, err)
	return result.Subscription
}

func requireProviderWatchInList(t *testing.T, watches []ServiceEventSubscriptionRecord, watchID string) ServiceEventSubscriptionRecord {
	t.Helper()

	for _, watch := range watches {
		if strings.TrimSpace(watch.ID) == strings.TrimSpace(watchID) {
			return watch
		}
	}
	require.Failf(t, "provider watch not found", "watch %s not found in %+v", watchID, watches)
	return ServiceEventSubscriptionRecord{}
}

func providerWatchRecordIDs(watches []ServiceEventSubscriptionRecord) []string {
	ids := make([]string, 0, len(watches))
	for _, watch := range watches {
		ids = append(ids, watch.ID)
	}
	return ids
}

func requireProviderWatchNotInList(t *testing.T, watches []ServiceEventSubscriptionRecord, watchID string) {
	t.Helper()

	for _, watch := range watches {
		require.NotEqual(t, strings.TrimSpace(watchID), strings.TrimSpace(watch.ID))
	}
}

func requireProviderWatchLifecycleState(t *testing.T, pool coredb.QueryRower, workspaceID string, watchID string, expectedStatus string, expectedPausedAt bool, expectedRemovedAt bool) {
	t.Helper()

	var status string
	var hasPausedAt bool
	var hasRemovedAt bool
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT status, paused_at IS NOT NULL, removed_at IS NOT NULL
FROM service_event_subscriptions
WHERE workspace_id = $1 AND id = $2
`, workspaceID, watchID).Scan(&status, &hasPausedAt, &hasRemovedAt))
	require.Equal(t, expectedStatus, status)
	require.Equal(t, expectedPausedAt, hasPausedAt)
	require.Equal(t, expectedRemovedAt, hasRemovedAt)
}

func seedProviderWatchAgentEntry(t *testing.T, pool coredb.QueryExecer, fixture appAdminFixture, agentID string, agentMemberID string, agentName string, visibility string, addScopeMembershipProjection bool, now time.Time) {
	t.Helper()

	machineID := "mac_" + agentID
	mustExec(t, pool, `
INSERT IGNORE INTO machines (id, workspace_id, created_by_user_id, name, api_key, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $6)
`, machineID, fixture.WorkspaceID, fixture.UserID, agentName+"-machine", machineID+"-key", now)
	mustExec(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, created_at, updated_at)
VALUES ($1, $2, 'agent', $3, $4, $4)
`, agentMemberID, fixture.WorkspaceID, agentName, now)
	mustExec(t, pool, `
INSERT INTO agents (
  id, workspace_id, machine_id, member_id, name, runtime, model, description,
  auth_token_hash, auto_wake_enabled, status, activity, activity_detail,
  created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, 'codex', 'gpt-5-codex', '', NULL, TRUE, 'active', 'sleeping', '', $6, $6)
`, agentID, fixture.WorkspaceID, machineID, agentMemberID, agentName, now)
	if addScopeMembershipProjection {
		seedChannelParticipantProjectionForTests(t, pool, fixture.WorkspaceID, fixture.AllChannelID, "all", "", "public", fixture.MemberID, agentMemberID, now)
	}
	ownerMemberID := ""
	if visibility == "private" {
		ownerMemberID = fixture.MemberID
	}
	require.NoError(t, message.EnsureScopeMembership(context.Background(), pool, fixture.AllChannelID, agentMemberID, visibility, ownerMemberID, "", 0, 0, now))
}

func seedProviderWatchReplyThread(t *testing.T, pool coredb.QueryExecer, fixture appAdminFixture, threadChannelID string, shortID string, now time.Time) {
	t.Helper()
	seedProviderWatchReplyThreadWithRootAuthor(t, pool, fixture, threadChannelID, shortID, fixture.MemberID, now)
}

func seedProviderWatchReplyThreadWithRootAuthor(t *testing.T, pool coredb.QueryExecer, fixture appAdminFixture, threadChannelID string, shortID string, authorMemberID string, now time.Time) {
	t.Helper()

	parentMessageID := "msg_" + threadChannelID
	var nextSeq int64
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT COALESCE(MAX(seq), 0) + 1
FROM scope_messages
WHERE workspace_id = $1
  AND scope_id = $2
`, fixture.WorkspaceID, fixture.AllChannelID).Scan(&nextSeq))
	nextGlobalSeq := now.UnixNano()
	mustExec(t, pool, `
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, $4, $5, $6, 'human', '', 'provider watch reply root', '{}', $7, $7)
`, parentMessageID, fixture.WorkspaceID, fixture.AllChannelID, nextSeq, nextGlobalSeq, authorMemberID, now)
	mustExec(t, pool, `
INSERT IGNORE INTO scopes (
  id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id,
  identity_key, latest_seq, latest_global_seq, archived_until_seq, status, created_by_member_id, created_at, updated_at
) VALUES ($1, $2, 'thread', $3, $4, $5, $6, 0, 0, 0, 'active', $7, $8, $8)
`, threadChannelID, fixture.WorkspaceID, fixture.AllChannelID, parentMessageID, shortID, "thread:"+fixture.AllChannelID+":"+parentMessageID, fixture.MemberID, now)
	mustExec(t, pool, `
INSERT IGNORE INTO scope_channel_profiles (
  scope_id, workspace_id, slug, display_name, description, visibility,
  owner_member_id, is_all_channel, created_at, updated_at
) VALUES ($1, $2, $3, $3, '', 'private', NULL, FALSE, $4, $4)
`, threadChannelID, fixture.WorkspaceID, "thread-"+shortID, now)
}

func requireServiceEventDeliveryCount(t *testing.T, pool coredb.Queryer, subscriptionID string, want int) {
	t.Helper()
	var count int
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT COUNT(*)
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
`, subscriptionID).Scan(&count))
	require.Equal(t, want, count)
}

func requireServiceEventDeliveryPolicy(t *testing.T, pool coredb.Queryer, subscriptionID string, wantNotify bool) {
	t.Helper()
	var notify bool
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT notify_principal
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
ORDER BY created_at DESC, id DESC
LIMIT 1
`, subscriptionID).Scan(&notify))
	require.Equal(t, wantNotify, notify)
}

func requireServiceEventDeliveryEffect(t *testing.T, pool coredb.Queryer, subscriptionID string, wantStatus string, wantEffectKind string, wantEffectRef string, wantMessageID string) {
	t.Helper()
	var status, effectKind, effectRef, messageID string
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT status, effect_kind, effect_ref, COALESCE(message_id, '')
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
ORDER BY created_at DESC, id DESC
LIMIT 1
`, subscriptionID).Scan(&status, &effectKind, &effectRef, &messageID))
	require.Equal(t, wantStatus, status)
	require.Equal(t, wantEffectKind, effectKind)
	require.Equal(t, wantEffectRef, effectRef)
	require.Equal(t, wantMessageID, messageID)
}

func requireServiceEventDeliveryFailure(t *testing.T, pool coredb.Queryer, subscriptionID string, wantCode string) {
	t.Helper()
	var status, code string
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT status, COALESCE(error_code, '')
FROM service_event_deliveries
WHERE service_event_subscription_id = $1
ORDER BY created_at DESC, id DESC
LIMIT 1
`, subscriptionID).Scan(&status, &code))
	require.Equal(t, "failed", status)
	require.Equal(t, wantCode, code)
}

func requireProviderWatchDeliverySource(t *testing.T, pool coredb.QueryExecer, delivery message.MessageDelivery, expectedWatchID string) string {
	t.Helper()

	source := delivery.Message.Source.Normalize()
	require.Equal(t, message.ContributionSourceKindProviderEvent, source.Kind)
	require.Equal(t, message.ContributionSourceRefTypeServiceEventDelivery, source.RefType)
	require.NotEmpty(t, source.RunID)
	require.Equal(t, source.RunID, source.RefID)
	require.NotEmpty(t, source.EventID)
	require.NotEmpty(t, source.EventType)
	require.Empty(t, delivery.Message.Metadata)
	if delivery.AppMessage != nil {
		require.Equal(t, source, delivery.AppMessage.Source.Normalize())
		require.Empty(t, delivery.AppMessage.Metadata)
	}
	requireScopeFanoutPayloadSource(t, pool, delivery.MessageID, source)

	var subscriptionID string
	var deliveryEventID string
	var deliveryEventType string
	var deliveryEventAction string
	require.NoError(t, pool.QueryRow(context.Background(), `
SELECT service_event_subscription_id, event_id, event_type, event_action
FROM service_event_deliveries
WHERE id = $1
  AND message_id = $2
`, source.RunID, delivery.MessageID).Scan(&subscriptionID, &deliveryEventID, &deliveryEventType, &deliveryEventAction))
	if strings.TrimSpace(expectedWatchID) != "" {
		require.Equal(t, expectedWatchID, subscriptionID)
	}
	require.Equal(t, deliveryEventID, source.EventID)
	require.Contains(t, []string{deliveryEventType, deliveryEventAction}, source.EventType)

	return subscriptionID
}

func seedProviderWatchProviderConfig(t *testing.T, ctx context.Context, appStore *Store, fixture appAdminFixture, providerName string) {
	t.Helper()

	switch providerName {
	case "github":
		_, err := appStore.Provider().SaveGitHubAppConfigForWorkspaceAdmin(ctx, fixture.UserID, fixture.WorkspaceID, provider.UpsertGitHubAppConfigParams{
			AppID:         12345,
			AppSlug:       "agentteam-test",
			AppName:       "AgentTeam Test",
			HTMLURL:       "https://github.com/apps/agentteam-test",
			ClientID:      "Iv1.test",
			ClientSecret:  "client-secret",
			PrivateKeyPEM: "-----BEGIN RSA PRIVATE KEY-----\nkey\n-----END RSA PRIVATE KEY-----",
			WebhookSecret: "github-webhook-secret",
			OwnerLogin:    "acme",
			OwnerID:       67890,
		})
		require.NoError(t, err)
	case "gitea":
		view, err := appStore.Provider().GetGiteaAppConfigForUser(ctx, fixture.UserID, fixture.WorkspaceID)
		require.NoError(t, err)
		require.True(t, view.Registered, "Gitea provider config must be installed through the approval fixture")
	default:
		t.Fatalf("unsupported provider %q", providerName)
	}
}

func installGiteaForProviderWatchTests(t *testing.T, ctx context.Context, appStore *Store, fixture appAdminFixture) AppInstallationView {
	t.Helper()
	return requestAndApproveAppInstallationForTests(t, ctx, appStore, fixture.UserID, fixture.UserID, fixture.WorkspaceID, giteaAppSlug, map[string]any{
		giteaConfigInstanceURLKey:   "https://gitea.example.com",
		giteaConfigAPITokenKey:      "gitea-api-token",
		giteaConfigWebhookSecretKey: "gitea-webhook-secret",
		giteaConfigDisplayNameKey:   "Gitea",
	})
}

func providerWatchGitHubCommentPayload(prNumber int, body string) []byte {
	return []byte(fmt.Sprintf(`{
	  "action": "created",
	  "installation": {"id": 789},
	  "repository": {"full_name": "Acme/AgentTeam"},
	  "issue": {
	    "number": %d,
	    "html_url": "https://github.com/acme/agentteam/issues/%d",
	    "pull_request": {"html_url": "https://github.com/acme/agentteam/pull/%d"}
	  },
	  "comment": {
	    "id": 1,
	    "body": %q,
	    "html_url": "https://github.com/acme/agentteam/issues/%d#issuecomment-1",
	    "created_at": "2026-04-19T09:08:07Z",
	    "user": {"login": "reviewer"}
	  },
	  "sender": {"login": "reviewer"}
	}`, prNumber, prNumber, prNumber, body, prNumber))
}

func githubWebhookSignature(payload []byte, secret string) string {
	mac := hmac.New(sha256.New, []byte(secret))
	_, _ = mac.Write(payload)
	return "sha256=" + hex.EncodeToString(mac.Sum(nil))
}

func giteaWebhookSignature(payload []byte, secret string) string {
	mac := hmac.New(sha256.New, []byte(secret))
	_, _ = mac.Write(payload)
	return "sha256=" + hex.EncodeToString(mac.Sum(nil))
}