package kb

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/cgi"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/agsbridge/gittransport"
	"github.com/tidbcloud/AgentTeam/internal/agsbridge/wiki"
	"github.com/tidbcloud/AgentTeam/internal/gormtest"
	"github.com/tidbcloud/AgentTeam/internal/service/providertokens"
	"github.com/tidbcloud/AgentTeam/internal/service/resource/activity"
	"github.com/tidbcloud/AgentTeam/internal/service/resource/repos"

	gogit "github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/config"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/go-git/go-git/v5/plumbing/object"
	"github.com/go-git/go-git/v5/plumbing/transport"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/store/automation"
	"github.com/tidbcloud/AgentTeam/internal/store/kb"
	providertokensstore "github.com/tidbcloud/AgentTeam/internal/store/providertokens"
	activitystore "github.com/tidbcloud/AgentTeam/internal/store/resource/activity"
	"github.com/tidbcloud/AgentTeam/internal/store/resource/memory"
	reposstore "github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
)

func TestSourceMirrorSyncJobPayloadCarriesOnlySafeDiagnostics(t *testing.T) {
	encoded, err := encodeSourceMirrorSyncJobPayload(SourceMirrorSyncJobPayload{
		SourceID:    " src_1 ",
		Ref:         " refs/heads/main ",
		Status:      " queued ",
		Trigger:     " manual ",
		RequestedBy: " owner ",
		TraceID:     " trace_1 ",
	})
	require.NoError(t, err)
	require.JSONEq(t, `{"source_id":"src_1","ref":"refs/heads/main","status":"queued","trigger":"manual","requested_by":"owner","trace_id":"trace_1"}`, encoded)
	require.NotContains(t, strings.ToLower(encoded), "token")
	require.NotContains(t, strings.ToLower(encoded), "secret")
	require.NotContains(t, strings.ToLower(encoded), "credential")

	payload, err := parseSourceMirrorSyncJobPayload(encoded)
	require.NoError(t, err)
	require.Equal(t, "src_1", payload.SourceID)
	require.Equal(t, "refs/heads/main", payload.Ref)
	require.Equal(t, "queued", payload.Status)
	require.Equal(t, "manual", payload.Trigger)
	require.Equal(t, "owner", payload.RequestedBy)
	require.Equal(t, "trace_1", payload.TraceID)
}

func TestSourceMirrorSyncJobTypeConstant(t *testing.T) {
	require.Equal(t, "sync_source_mirror", KBJobTypeSyncSourceMirror)
}

type staticSourceMirrorResolver struct{ ips []net.IP }

func (r staticSourceMirrorResolver) LookupIP(context.Context, string) ([]net.IP, error) {
	return r.ips, nil
}

type fakeSourceMirrorSyncer struct {
	mu       sync.Mutex
	request  SourceMirrorSyncRequest
	progress []SourceMirrorGitProgress
	result   SourceMirrorSyncResult
	err      error
}

const fakeSourceMirrorCommit = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"

type blockingSourceMirrorSyncer struct {
	started  chan string
	releases map[string]chan struct{}
	failures map[string]error
}

func (s *blockingSourceMirrorSyncer) SyncSourceMirror(ctx context.Context, request SourceMirrorSyncRequest) (SourceMirrorSyncResult, error) {
	s.started <- request.Source.ID
	select {
	case <-ctx.Done():
		return SourceMirrorSyncResult{}, ctx.Err()
	case <-s.releases[request.Source.ID]:
	}
	if err := s.failures[request.Source.ID]; err != nil {
		return SourceMirrorSyncResult{}, err
	}
	commits := map[string]string{
		"src_1": "1111111111111111111111111111111111111111",
		"src_2": "2222222222222222222222222222222222222222",
		"src_3": "3333333333333333333333333333333333333333",
	}
	return SourceMirrorSyncResult{MirroredCommit: commits[request.Source.ID]}, nil
}

func (s *fakeSourceMirrorSyncer) SyncSourceMirror(_ context.Context, request SourceMirrorSyncRequest) (SourceMirrorSyncResult, error) {
	s.mu.Lock()
	s.request = request
	progressUpdates := append([]SourceMirrorGitProgress(nil), s.progress...)
	result := s.result
	err := s.err
	s.mu.Unlock()
	for _, progress := range progressUpdates {
		if request.OnGitProgress != nil {
			if err := request.OnGitProgress(progress); err != nil {
				return SourceMirrorSyncResult{}, err
			}
		}
	}
	if err != nil {
		return SourceMirrorSyncResult{}, err
	}
	if strings.TrimSpace(result.MirroredCommit) == "" {
		result.MirroredCommit = fakeSourceMirrorCommit
	}
	return result, nil
}

type fakeSourceMirrorCredentialResolver struct {
	request    SourceMirrorCredentialRequest
	credential SourceMirrorCredential
	err        error
}

func (r *fakeSourceMirrorCredentialResolver) ResolveSourceMirrorCredential(_ context.Context, request SourceMirrorCredentialRequest) (SourceMirrorCredential, error) {
	r.request = request
	if r.err != nil {
		return SourceMirrorCredential{}, r.err
	}
	return r.credential, nil
}

type fakePrincipalWorkspaceAccessVerifier struct {
	active bool
	err    error
}

func (v fakePrincipalWorkspaceAccessVerifier) HasWorkspaceAccess(context.Context, string, string) (bool, error) {
	if v.err != nil {
		return false, v.err
	}
	return v.active, nil
}

func newSourceMirrorServiceForTest(t *testing.T, syncer KBSourceMirrorRepositorySyncer, wikiStore wiki.AGSRepositoryStore) *KBService {
	t.Helper()
	db, err := gormtest.OpenConcurrentSQLiteDB(t)
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(
		&kb.KnowledgeBase{},
		&kb.KnowledgeBaseGroup{},
		&kb.KnowledgeBaseTag{},
		&kb.KnowledgeBaseTagAssignment{},
		&kb.KnowledgeBasePin{},
		&kb.KnowledgeBaseGrant{},
		&kb.KnowledgeBaseSourceRepositoryAccessOverride{},
		&kb.Source{},
		&kb.SourceMirror{},
		&reposstore.ResourceRepository{},
		&reposstore.ResourceRepositoryLifecycleEvent{},
		&reposstore.ResourceRepositoryAccessGrant{},
		&providertokensstore.PersonalAccessToken{},

		&kb.AnalysisRun{},
		&kb.SourceRevision{},
		&kb.Job{},
		&kb.AuditEvent{},
		&activitystore.ResourceActivityEvent{},
		&activitystore.ResourceActivityLatestCommit{},
		&activitystore.ResourceActivityProjectorCheckpoint{},
		&memory.AgentAssetAccessChangeOutbox{},
	))
	fixed := time.Date(2026, 6, 12, 1, 2, 3, 0, time.UTC)
	return NewKBServiceWithOptions(kb.New(db), func() time.Time { return fixed }, KBServiceOptions{
		SourceMirrorSyncer:       syncer,
		WikiStore:                wikiStore,
		RepositoryStore:          reposstore.New(db),
		PersonalAccessTokenStore: providertokensstore.New(db),
		ActivityStore:            activitystore.New(db),
		AssetAccessOutbox:        memory.New(db),
		WorkspaceGovernanceAuthorizer: fakeWorkspaceGovernanceAuthorizer{authorize: func(_ context.Context, _, principalID string) (bool, error) {
			return principalID == "owner", nil
		}},
		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
	})
}

func seedSourceMirror(t *testing.T, svc *KBService, enabled bool) kb.SourceMirror {
	t.Helper()
	now := time.Date(2026, 6, 12, 1, 2, 3, 0, time.UTC)
	base := kb.KnowledgeBase{ID: "kb_1", WorkspaceID: "wsp_1", Slug: "runbooks", Name: "Runbooks", RetrievalPolicy: "{}", SourceCollectionState: "{}", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now}
	_, err := svc.store.CreateKnowledgeBase(context.Background(), base)
	require.NoError(t, err)
	_, err = svc.store.CreateSource(context.Background(), kb.Source{ID: "src_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceType: "codebase", Status: KBSourceStatusReady, Name: "AgentTeam", DefaultRef: "main", SyncStatus: KBSourceStatusReady, IndexStatus: KBSourceStatusReady, Config: "{}", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	_, err = svc.repositoryStore.Upsert(context.Background(), reposstore.ResourceRepository{ID: "rr_mirror", WorkspaceID: "wsp_1", ResourceType: reposstore.ResourceTypeKBSource, ResourceID: "src_1", RepoRole: reposstore.RepoRoleMirror, OwnerKind: "workspace", RepoOwner: "wsp_1", RepoName: "agentteam", RepoFullName: "wsp_1/agentteam", AGSRepoID: "repo_mirror", DefaultRef: "main", Visibility: "private", UserVisible: true, CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	_, err = svc.repositoryStore.Upsert(context.Background(), reposstore.ResourceRepository{ID: "rr_insights", WorkspaceID: "wsp_1", ResourceType: reposstore.ResourceTypeKBSource, ResourceID: "src_1", RepoRole: reposstore.RepoRoleInsights, OwnerKind: "workspace", RepoOwner: "wsp_1", RepoName: "agentteam.kb", RepoFullName: "wsp_1/agentteam.kb", AGSRepoID: "repo_insights", DefaultRef: "main", Visibility: "private", UserVisible: false, CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	mirror, err := svc.store.CreateSourceMirror(context.Background(), kb.SourceMirror{SourceID: "src_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: "https://93.184.216.34/org/repo.git", DefaultRef: "main", IntervalSeconds: 60, Enabled: enabled, SyncStatus: kb.SourceMirrorSyncStatusIdle, CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	return mirror
}

func TestValidateSourceMirrorRemoteURLRejectsUnsafeAndSecretBearingURLs(t *testing.T) {
	ctx := context.Background()
	_, err := ValidateSourceMirrorRemoteURL(ctx, "file:///tmp/repo.git", SourceMirrorRemoteValidationOptions{})
	require.Error(t, err)
	_, err = ValidateSourceMirrorRemoteURL(ctx, "https://token@example.com/org/repo.git", SourceMirrorRemoteValidationOptions{})
	require.Error(t, err)
	_, err = ValidateSourceMirrorRemoteURL(ctx, "https://127.0.0.1/org/repo.git", SourceMirrorRemoteValidationOptions{})
	require.Error(t, err)
	_, err = ValidateSourceMirrorRemoteURL(ctx, "https://git.example.invalid/org/repo.git", SourceMirrorRemoteValidationOptions{Resolver: staticSourceMirrorResolver{ips: []net.IP{net.ParseIP("10.0.0.5")}}})
	require.Error(t, err)

	sanitized, err := ValidateSourceMirrorRemoteURL(ctx, "https://git.example.invalid/org/repo.git", SourceMirrorRemoteValidationOptions{Resolver: staticSourceMirrorResolver{ips: []net.IP{net.ParseIP("93.184.216.34")}}})
	require.NoError(t, err)
	require.Equal(t, "https://git.example.invalid/org/repo.git", sanitized)
}

func TestValidateGitHubSourceMirrorRemoteRequiresPublicHTTPSRepositoryURL(t *testing.T) {
	ctx := context.Background()
	resolver := staticSourceMirrorResolver{ips: []net.IP{net.ParseIP("140.82.112.4")}}

	for _, remote := range []string{
		"https://github.com/pingcap/agentteam.git",
		"https://github.com/pingcap/agentteam",
		"https://github.com/org-name/.github",
	} {
		sanitized, err := ValidateSourceMirrorRemoteForUpstream(ctx, kb.SourceMirrorUpstreamTypeGitHub, remote, SourceMirrorRemoteValidationOptions{Resolver: resolver})
		require.NoError(t, err)
		require.Equal(t, remote, sanitized)
	}

	for _, remote := range []string{
		"git@github.com:pingcap/agentteam.git",
		"ssh://github.com/pingcap/agentteam.git",
		"http://github.com/pingcap/agentteam.git",
		"https://github.enterprise.local/pingcap/agentteam.git",
		"https://github.com/org_name/agentteam.git",
		"https://github.com/-org/agentteam.git",
		"https://github.com/pingcap/agentteam/tree/main",
		"https://github.com/pingcap",
	} {
		_, err := ValidateSourceMirrorRemoteForUpstream(ctx, kb.SourceMirrorUpstreamTypeGitHub, remote, SourceMirrorRemoteValidationOptions{Resolver: resolver})
		require.Error(t, err, remote)
	}
}

func TestValidateSelfHostedGitMirrorRemoteAcceptsValidURLs(t *testing.T) {
	ctx := context.Background()
	resolver := staticSourceMirrorResolver{ips: []net.IP{net.ParseIP("93.184.216.34")}}

	for _, upstreamType := range []string{
		kb.SourceMirrorUpstreamTypeGitLab,
		kb.SourceMirrorUpstreamTypeGitea,
	} {
		for _, remote := range []string{
			"https://gitlab.example.com/org/repo.git",
			"https://gitlab.example.com/group/subgroup/repo.git",
			"git@gitlab.example.com:org/repo.git",
		} {
			_, err := ValidateSourceMirrorRemoteForUpstream(ctx, upstreamType, remote, SourceMirrorRemoteValidationOptions{Resolver: resolver})
			require.NoError(t, err, fmt.Sprintf("%s %s", upstreamType, remote))
		}

		for _, remote := range []string{
			"http://gitlab.example.com/org/repo.git",
			"https://gitlab.example.com/repo.git",
			"https://gitlab.example.com/org",
			"git@gitlab.example.com:repo.git",
		} {
			_, err := ValidateSourceMirrorRemoteForUpstream(ctx, upstreamType, remote, SourceMirrorRemoteValidationOptions{Resolver: resolver})
			require.Error(t, err, fmt.Sprintf("%s %s", upstreamType, remote))
		}
	}
}

func TestKBServiceRunsSourceMirrorJobThroughSyncerAndReconciles(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	wikiStore := &fakeAGSRepositoryStore{}
	svc := newSourceMirrorServiceForTest(t, syncer, wikiStore)
	var logs bytes.Buffer
	svc.logger = slog.New(slog.NewJSONHandler(&logs, nil))
	seedSourceMirror(t, svc, true)
	_, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued, Progress: `{"source_id":"src_1","ref":"main"}`, CreatedAt: svc.now(), UpdatedAt: svc.now()})
	require.NoError(t, err)

	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "test-worker"})

	require.NoError(t, err)
	require.Equal(t, SourceMirrorRunResult{Claimed: 1, Succeeded: 1}, result)
	require.Equal(t, "https://93.184.216.34/org/repo.git", syncer.request.RemoteURL)
	require.Equal(t, "main", syncer.request.DefaultRef)
	require.Empty(t, syncer.request.CredentialRef)
	require.Equal(t, "wsp_1/agentteam", syncer.request.Target.FullName)
	require.True(t, wikiStore.reconciled)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, kb.SourceMirrorSyncStatusSuccess, mirror.SyncStatus)
	require.NotNil(t, mirror.LastSyncAt)
	require.Equal(t, svc.now(), *mirror.LastSyncAt)
	require.NotNil(t, mirror.LastSuccessAt)
	require.Equal(t, svc.now(), *mirror.LastSuccessAt)
	require.Equal(t, fakeSourceMirrorCommit, mirror.LastMirroredCommit)
	revision, err := svc.store.GetCurrentSourceRevision(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, fakeSourceMirrorCommit, revision.SourceCommitSHA)
	require.True(t, revision.IsCurrent)
	require.Empty(t, revision.UADashboardCommitSHA)
	status, err := svc.GetUADashboardStatus(ctx, "wsp_1", "owner", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, "empty", status.State)
	require.Equal(t, fakeSourceMirrorCommit, status.SourceCommitSHA)
	require.True(t, status.CanImport)
	require.NotNil(t, mirror.NextSyncAt)
	job, err := svc.store.GetJob(ctx, "wsp_1", "job_1")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusSucceeded, job.Status)
	require.NotContains(t, job.Progress, "credential")
	activityPage, err := svc.ListKnowledgeBaseActivity(ctx, "wsp_1", "owner", activity.ResourceActivityListOptions{EventType: "kb_source_mirror_sync_succeeded", Limit: 10})
	require.NoError(t, err)
	require.Len(t, activityPage.Items, 1)
	require.Equal(t, activity.ActivityActorTypeUser, activityPage.Items[0].Actor.Type)
	require.Equal(t, "owner", activityPage.Items[0].Actor.ID)
	require.Equal(t, "AgentTeam", activityPage.Items[0].Details["sourceName"])
	require.Equal(t, "source_mirror_sync", activityPage.Items[0].Target.Kind)
	require.Equal(t, "job_1", activityPage.Items[0].Target.ID)
	require.Contains(t, logs.String(), `"event":"succeeded"`)
	require.Contains(t, logs.String(), `"mirrored_commit":"`+fakeSourceMirrorCommit+`"`)
	require.NotContains(t, logs.String(), `"mirroredCommit"`)
	events, err := svc.assetAccessOutbox.LeaseDueAgentAssetAccessChanges(ctx, svc.now(), "daemon-1", svc.now().Add(time.Minute), 10, nil)
	require.NoError(t, err)
	require.Empty(t, events)
}

func TestCompleteSourceMirrorSyncJobRejectsStaleRevisionCompletion(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	now := svc.now()
	_, _, err := svc.store.AdvanceSourceRevision(ctx, kb.AdvanceSourceRevisionParams{Revision: kb.SourceRevision{
		ID: "rev_new", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		SourceCommitSHA: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", FreshnessStatus: "current", CreatedAt: now, UpdatedAt: now,
	}})
	require.NoError(t, err)
	job, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_stale", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning, Progress: `{"source_id":"src_1","ref":"main"}`, LeaseOwner: "worker", AttemptCount: 1, CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)

	err = svc.completeSourceMirrorSyncJob(ctx, job, "worker", sourceMirrorSyncPayload{SourceID: "src_1", SourceRevisionFenceSet: true}, source, mirror, SourceMirrorSyncResult{MirroredCommit: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"})
	require.ErrorContains(t, err, "superseded")
	current, loadErr := svc.store.GetCurrentSourceRevision(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, loadErr)
	require.Equal(t, "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", current.SourceCommitSHA)
	staleJob, loadErr := svc.store.GetJob(ctx, "wsp_1", "job_stale")
	require.NoError(t, loadErr)
	require.Equal(t, KBJobStatusRunning, staleJob.Status, "stale completion transaction must roll back the job success")
}

func TestCompleteSourceMirrorSyncJobRejectsNonImmutableCommit(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	now := svc.now()
	job, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_invalid", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning, Progress: `{"source_id":"src_1","ref":"main"}`, LeaseOwner: "worker", AttemptCount: 1, CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)

	err = svc.completeSourceMirrorSyncJob(ctx, job, "worker", sourceMirrorSyncPayload{SourceID: "src_1"}, source, mirror, SourceMirrorSyncResult{MirroredCommit: "main"})

	require.ErrorContains(t, err, "immutable commit")
	_, loadErr := svc.store.GetCurrentSourceRevision(ctx, "wsp_1", "kb_1", "src_1")
	require.Error(t, loadErr)
	storedJob, loadErr := svc.store.GetJob(ctx, "wsp_1", "job_invalid")
	require.NoError(t, loadErr)
	require.Equal(t, KBJobStatusRunning, storedJob.Status)
}

func TestMirrorWorkerCannotRollbackNewerConfiguration(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	stale := seedSourceMirror(t, svc, true)
	now := svc.now()
	job, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_config", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning, Progress: `{"source_id":"src_1","ref":"main"}`, LeaseOwner: "worker", AttemptCount: 1, CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)

	current := stale
	current.IntervalSeconds = 120
	current.UpdatedAt = now.Add(time.Minute)
	_, err = svc.store.UpdateSourceMirror(ctx, current)
	require.NoError(t, err)

	err = svc.markSourceMirrorRunning(ctx, &job, "worker", source, stale, "main", stale.RemoteURL)
	require.NoError(t, err)
	afterRunning, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, 120, afterRunning.IntervalSeconds)

	err = svc.completeSourceMirrorSyncJob(ctx, job, "worker", sourceMirrorSyncPayload{SourceID: "src_1", SourceRevisionFenceSet: true}, source, stale, SourceMirrorSyncResult{MirroredCommit: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"})
	require.NoError(t, err)
	afterCompletion, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, 120, afterCompletion.IntervalSeconds)
}

func TestMirrorWorkerManualRefDoesNotChangeConfiguredDefaultRef(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	mirror := seedSourceMirror(t, svc, true)
	now := svc.now()
	job, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_manual_ref", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning, Progress: `{"source_id":"src_1","ref":"release"}`, LeaseOwner: "worker", AttemptCount: 1, CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)

	require.NoError(t, svc.markSourceMirrorRunning(ctx, &job, "worker", source, mirror, "release", mirror.RemoteURL))
	afterRunning, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, "main", afterRunning.DefaultRef)

	require.NoError(t, svc.completeSourceMirrorSyncJob(ctx, job, "worker", sourceMirrorSyncPayload{SourceID: "src_1", Ref: "release", SourceRevisionFenceSet: true}, source, mirror, SourceMirrorSyncResult{MirroredCommit: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}))
	afterCompletion, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, "main", afterCompletion.DefaultRef)
}

func TestSourceMirrorRemoteChangePublishesFromJobRevisionFence(t *testing.T) {
	ctx := context.Background()
	newCommit := "cccccccccccccccccccccccccccccccccccccccc"
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: newCommit}}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	now := svc.now()
	oldCommit := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	_, _, err := svc.store.AdvanceSourceRevision(ctx, kb.AdvanceSourceRevisionParams{Revision: kb.SourceRevision{
		ID: "rev_old", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		SourceCommitSHA: oldCommit, FreshnessStatus: "current", CreatedAt: now, UpdatedAt: now,
	}})
	require.NoError(t, err)

	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/new-repo.git",
		DefaultRef:   "release",
	})
	require.NoError(t, err)
	queued, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "release", "owner", KBSourceMirrorStageQueued, true)
	require.NoError(t, err)
	payload, err := parseSourceMirrorSyncJobPayload(queued.Progress)
	require.NoError(t, err)
	require.True(t, payload.SourceRevisionFenceSet)
	require.Equal(t, oldCommit, payload.ExpectedCurrentSourceCommitSHA)

	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "worker"})
	require.NoError(t, err)
	require.Equal(t, SourceMirrorRunResult{Claimed: 1, Succeeded: 1}, result)
	current, err := svc.store.GetCurrentSourceRevision(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, newCommit, current.SourceCommitSHA)
}

func TestSourceMirrorConfigChangeCancelsAnOlderQueuedGeneration(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	queued, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, true)
	require.NoError(t, err)

	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/new-repo.git",
		DefaultRef:   "release",
	})
	require.NoError(t, err)
	stored, err := svc.store.GetJob(ctx, "wsp_1", queued.ID)
	require.NoError(t, err)
	require.Equal(t, KBJobStatusCancelled, stored.Status)

	newer, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "release", "owner", KBSourceMirrorStageQueued, true)
	require.NoError(t, err)
	require.NotEqual(t, queued.ID, newer.ID)
}

func TestUpdateSourceMirrorRejectsStaleExpectedDefaultRef(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	expected := "release"

	_, err := svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		DefaultRef:         "next",
		DefaultRefSet:      true,
		ExpectedDefaultRef: &expected,
	})
	require.ErrorIs(t, err, ErrConflict)
	require.ErrorContains(t, err, "default branch changed")

	stored, loadErr := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, loadErr)
	require.Equal(t, "main", stored.DefaultRef)
}

func TestSourceMirrorGenerationComparisonRejectsStaleConfig(t *testing.T) {
	old := kb.SourceMirror{SourceID: "src_1", Enabled: true, UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: "https://93.184.216.34/org/old.git", DefaultRef: "main", CredentialRef: "secret://old"}
	require.True(t, sameSourceMirrorSyncGeneration(old, old))

	for name, mutate := range map[string]func(*kb.SourceMirror){
		"disabled":   func(m *kb.SourceMirror) { m.Enabled = false },
		"upstream":   func(m *kb.SourceMirror) { m.UpstreamType = kb.SourceMirrorUpstreamTypeGitHub },
		"remote":     func(m *kb.SourceMirror) { m.RemoteURL = "https://93.184.216.34/org/new.git" },
		"ref":        func(m *kb.SourceMirror) { m.DefaultRef = "release" },
		"credential": func(m *kb.SourceMirror) { m.CredentialRef = "secret://new" },
	} {
		t.Run(name, func(t *testing.T) {
			current := old
			mutate(&current)
			require.False(t, sameSourceMirrorSyncGeneration(old, current))
		})
	}
}

func TestStaleMirrorMutationCannotRollbackConfigOrMismatchedActiveJob(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	old, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	base, err := svc.store.GetKnowledgeBase(ctx, "wsp_1", "kb_1")
	require.NoError(t, err)

	newRemote := "https://93.184.216.34/org/new-repo.git"
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: newRemote, DefaultRef: "release",
	})
	require.NoError(t, err)
	queued, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "release", "owner", KBSourceMirrorStageQueued, true)
	require.NoError(t, err)

	staleUpdate := old
	staleUpdate.IntervalSeconds = 120
	staleUpdate.UpdatedAt = svc.now().Add(time.Minute)
	_, err = svc.saveSourceMirrorUpdate(ctx, base, source, old, staleUpdate, nil, nil, "owner", staleUpdate.UpdatedAt, false)
	require.ErrorIs(t, err, ErrConflict)
	current, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, newRemote, current.RemoteURL)
	require.Equal(t, "release", current.DefaultRef)
	active, err := svc.store.GetJob(ctx, "wsp_1", queued.ID)
	require.NoError(t, err)
	require.Equal(t, KBJobStatusQueued, active.Status)
	payload, err := parseSourceMirrorSyncJobPayload(active.Progress)
	require.NoError(t, err)
	require.Equal(t, "release", payload.Ref)
}

func TestStaleMirrorMutationCannotOverwriteWinningAccessToken(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	require.NoError(t, svc.store.DB().AutoMigrate(&automation.AutomationSecret{}))
	key := []byte("0123456789abcdef0123456789abcdef")
	svc.automationSecretStore = automation.New(svc.store.DB())
	svc.automationCredentialsKey = key
	seedSourceMirror(t, svc, true)

	initial, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	base, err := svc.store.GetKnowledgeBase(ctx, "wsp_1", "kb_1")
	require.NoError(t, err)

	winner, err := svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		RemoteURL: "https://93.184.216.34/org/winner.git", AccessToken: "winner-token",
	})
	require.NoError(t, err)

	loser := initial
	loser.RemoteURL = "https://93.184.216.34/org/loser.git"
	loser.CredentialRef = "automation-secret://" + kbMirrorCredentialSecretName
	loser.UpdatedAt = svc.now().Add(time.Minute)
	pending, err := svc.prepareMirrorAccessToken("wsp_1", "wsp_1", "agentteam", "loser-token", "owner", loser.UpdatedAt)
	require.NoError(t, err)
	_, err = svc.saveSourceMirrorUpdate(ctx, base, source, initial, loser, nil, &pending, "owner", loser.UpdatedAt, true)
	require.ErrorIs(t, err, ErrConflict)

	current, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, winner.RemoteURL, current.RemoteURL)
	require.Equal(t, winner.CredentialRef, current.CredentialRef)
	secret, err := svc.automationSecretStore.GetAutomationSecret(ctx, "wsp_1", "wsp_1", "agentteam", kbMirrorCredentialSecretName)
	require.NoError(t, err)
	codec, err := secrets.NewAESGCMCodec(key)
	require.NoError(t, err)
	plaintext, err := codec.Open(secret.ValueCiphertext)
	require.NoError(t, err)
	require.Equal(t, "winner-token", plaintext)
}

func TestMirrorAccessTokenRollsBackWhenMirrorUpdateFails(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	require.NoError(t, svc.store.DB().AutoMigrate(&automation.AutomationSecret{}))
	key := []byte("0123456789abcdef0123456789abcdef")
	svc.automationSecretStore = automation.New(svc.store.DB())
	svc.automationCredentialsKey = key
	initial := seedSourceMirror(t, svc, true)
	pendingInitial, err := svc.prepareMirrorAccessToken("wsp_1", "wsp_1", "agentteam", "initial-token", "owner", svc.now())
	require.NoError(t, err)
	_, err = persistMirrorAccessToken(ctx, svc.automationSecretStore, pendingInitial)
	require.NoError(t, err)

	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	base, err := svc.store.GetKnowledgeBase(ctx, "wsp_1", "kb_1")
	require.NoError(t, err)
	invalid := initial
	invalid.UpstreamType = "invalid"
	invalid.CredentialRef = "automation-secret://" + kbMirrorCredentialSecretName
	invalid.UpdatedAt = svc.now().Add(time.Minute)
	pendingReplacement, err := svc.prepareMirrorAccessToken("wsp_1", "wsp_1", "agentteam", "replacement-token", "owner", invalid.UpdatedAt)
	require.NoError(t, err)

	_, err = svc.saveSourceMirrorUpdate(ctx, base, source, initial, invalid, nil, &pendingReplacement, "owner", invalid.UpdatedAt, false)
	require.Error(t, err)
	secret, err := svc.automationSecretStore.GetAutomationSecret(ctx, "wsp_1", "wsp_1", "agentteam", kbMirrorCredentialSecretName)
	require.NoError(t, err)
	codec, err := secrets.NewAESGCMCodec(key)
	require.NoError(t, err)
	plaintext, err := codec.Open(secret.ValueCiphertext)
	require.NoError(t, err)
	require.Equal(t, "initial-token", plaintext)
}

func TestRunPendingSourceMirrorJobPersistsSafeFailureDiagnosticsAndReturnsError(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{
		progress: []SourceMirrorGitProgress{
			{Operation: "clone", Attempt: 1, MaxAttempts: 3},
			{Operation: "clone", Attempt: 2, MaxAttempts: 3, Retrying: true, Failure: errors.New("connection reset by peer")},
		},
		err: fmt.Errorf("clone https://user:fixture-secret@example.com/org/repo.git?access_token=fixture-secret: authentication failed"),
	}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	var logs bytes.Buffer
	svc.logger = slog.New(slog.NewJSONHandler(&logs, nil))
	seedSourceMirror(t, svc, true)
	queued, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "test-worker"})

	require.Error(t, err)
	require.NotContains(t, err.Error(), "fixture-secret")
	require.Equal(t, SourceMirrorRunResult{Claimed: 1, Failed: 1}, result)
	job, err := svc.store.GetJob(ctx, "wsp_1", queued.ID)
	require.NoError(t, err)
	require.Equal(t, KBJobStatusFailed, job.Status)
	require.NotContains(t, job.Error, "fixture-secret")
	require.NotContains(t, job.Progress, "fixture-secret")
	diagnostics := SourceMirrorJobDiagnosticsForJob(job)
	require.Equal(t, "main", diagnostics.Ref)
	require.Equal(t, KBSourceMirrorTriggerManual, diagnostics.Trigger)
	require.Equal(t, "owner", diagnostics.RequestedBy)
	require.NotEmpty(t, diagnostics.TraceID)
	require.Equal(t, KBSourceMirrorStageSyncingAGS, diagnostics.FailedStage)
	require.Equal(t, sourceMirrorErrorUpstreamAuth, diagnostics.ErrorCode)
	require.NotNil(t, diagnostics.Retryable)
	require.False(t, *diagnostics.Retryable)
	require.Equal(t, "clone", diagnostics.GitOperation)
	require.Equal(t, 2, diagnostics.GitAttempt)
	require.Equal(t, 3, diagnostics.GitMaxAttempts)

	activityPage, err := svc.ListKnowledgeBaseActivity(ctx, "wsp_1", "owner", activity.ResourceActivityListOptions{EventType: "kb_source_mirror_sync_failed", Limit: 10})
	require.NoError(t, err)
	require.Len(t, activityPage.Items, 1)
	require.Equal(t, "source_mirror_sync", activityPage.Items[0].Target.Kind)
	require.Equal(t, queued.ID, activityPage.Items[0].Target.ID)
	require.NotEmpty(t, activityPage.Items[0].Details["error"])
	require.NotContains(t, fmt.Sprint(activityPage.Items[0].Details["error"]), "fixture-secret")
	require.NotContains(t, logs.String(), "fixture-secret")
	for _, event := range []string{"enqueued", "started", "stage_changed", "retrying", "failed"} {
		require.Contains(t, logs.String(), `"event":"`+event+`"`)
	}
	require.Contains(t, logs.String(), `"error_class":"upstream_auth"`)
}

func TestFailSourceMirrorSyncJobReturnsSafePersistenceFailure(t *testing.T) {
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	svc.store = kb.New(nil)
	now := svc.now()
	err := svc.failSourceMirrorSyncJob(context.Background(), kb.Job{
		ID:              "job_1",
		WorkspaceID:     "wsp_1",
		KnowledgeBaseID: "kb_1",
		SourceID:        "src_1",
		Status:          KBJobStatusRunning,
		Progress:        `{"kind":"source_mirror_sync","source_id":"src_1","stage":"syncing_ags_repo"}`,
		LeaseOwner:      "worker_1",
		AttemptCount:    1,
		CreatedAt:       now,
		UpdatedAt:       now,
	}, sourceMirrorSyncPayload{SourceID: "src_1"}, errors.New("https://user:fixture-secret@example.com/repo.git failed"))

	require.Error(t, err)
	require.ErrorContains(t, err, "persist source mirror failure state")
	require.ErrorContains(t, err, "kb store is not configured")
	require.NotContains(t, err.Error(), "fixture-secret")
}

func TestClassifySourceMirrorError(t *testing.T) {
	tests := []struct {
		name      string
		stage     string
		err       error
		code      string
		retryable bool
	}{
		{name: "auth", err: transport.ErrAuthorizationFailed, code: sourceMirrorErrorUpstreamAuth},
		{name: "missing repository", err: transport.ErrRepositoryNotFound, code: sourceMirrorErrorUpstreamMissing},
		{name: "missing ref", err: errors.New("remote ref does not exist"), code: sourceMirrorErrorRefMissing},
		{name: "dns", err: &net.DNSError{Err: "no such host", Name: "git.example.invalid"}, code: sourceMirrorErrorDNS, retryable: true},
		{name: "tls", err: errors.New("tls handshake timeout"), code: sourceMirrorErrorTLS, retryable: true},
		{name: "proxy", err: errors.New("proxyconnect tcp: proxy error"), code: sourceMirrorErrorProxy, retryable: true},
		{name: "timeout", err: context.DeadlineExceeded, code: sourceMirrorErrorCloneTimeout, retryable: true},
		{name: "reconcile", stage: KBSourceMirrorStageReconciling, err: errors.New("write failed"), code: sourceMirrorErrorReconcile, retryable: true},
		{name: "reconcile timeout", stage: KBSourceMirrorStageReconciling, err: context.DeadlineExceeded, code: sourceMirrorErrorReconcile, retryable: true},
		{name: "AGS push", err: fmt.Errorf("push source mirror to AGS failed: %w", context.DeadlineExceeded), code: sourceMirrorErrorAGSPush, retryable: true},
		{name: "network", err: errors.New("connection reset by peer"), code: sourceMirrorErrorNetwork, retryable: true},
		{name: "internal", err: errors.New("unexpected state"), code: sourceMirrorErrorInternal},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			diagnostic := classifySourceMirrorError(test.stage, test.err)
			require.Equal(t, test.code, diagnostic.Code)
			require.Equal(t, test.retryable, diagnostic.Retryable)
		})
	}
}

func TestKBServiceMirrorSuccessRefreshesGrantedAgents(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	now := svc.now()
	for _, grant := range []kb.KnowledgeBaseGrant{
		{ID: "kbg_agent_reader", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SubjectType: "agent", SubjectID: "agt_reader", Role: "view", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now},
		{ID: "kbg_agent_other", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SubjectType: "agent", SubjectID: "agt_other", Role: "view", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now},
		{ID: "kbg_user_viewer", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SubjectType: "user", SubjectID: "viewer", Role: "view", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now},
	} {
		_, err := svc.store.UpsertKnowledgeBaseGrant(ctx, grant)
		require.NoError(t, err)
	}
	_, err := svc.store.CreateJob(ctx, kb.Job{ID: "job_refresh", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued, Progress: `{"source_id":"src_1","ref":"main"}`, CreatedAt: now, UpdatedAt: now})
	require.NoError(t, err)

	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "test-worker"})

	require.NoError(t, err)
	require.Equal(t, SourceMirrorRunResult{Claimed: 1, Succeeded: 1}, result)
	events, err := svc.assetAccessOutbox.LeaseDueAgentAssetAccessChanges(ctx, now, "daemon-1", now.Add(time.Minute), 10, nil)
	require.NoError(t, err)
	require.Len(t, events, 2)
	require.ElementsMatch(t, []string{"agt_reader", "agt_other"}, []string{events[0].AgentID, events[1].AgentID})
	for _, event := range events {
		require.Equal(t, "wsp_1", event.WorkspaceID)
		require.Equal(t, memory.AssetTypeKB, event.AssetType)
		require.Equal(t, now, event.OccurredAt)
	}
}

func TestKBServiceScheduledSourceMirrorActivityUsesSystemActor(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	due := svc.now().Add(-time.Minute)
	mirror.NextSyncAt = &due
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	result, err := svc.EnqueueDueSourceMirrorSyncs(ctx, SourceMirrorScheduleOptions{Limit: 1})

	require.NoError(t, err)
	require.Equal(t, SourceMirrorScheduleResult{Scanned: 1, Enqueued: 1}, result)
	_, err = svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "test-worker"})
	require.NoError(t, err)
	activityPage, err := svc.ListKnowledgeBaseActivity(ctx, "wsp_1", "owner", activity.ResourceActivityListOptions{EventType: "kb_source_mirror_sync_succeeded", Limit: 10})
	require.NoError(t, err)
	require.Len(t, activityPage.Items, 1)
	require.Equal(t, activity.ActivityActorTypeSystem, activityPage.Items[0].Actor.Type)
	require.Empty(t, activityPage.Items[0].Actor.ID)
}

func TestEnqueueDueSourceMirrorSyncsReturnsPerSourceFailures(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	due := svc.now().Add(-time.Minute)
	mirror.NextSyncAt = &due
	mirror.RemoteURL = "https://127.0.0.1/org/repo.git"
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	result, err := svc.EnqueueDueSourceMirrorSyncs(ctx, SourceMirrorScheduleOptions{Limit: 1})

	require.Error(t, err)
	require.Equal(t, SourceMirrorScheduleResult{Scanned: 1, Failed: 1}, result)
}

func TestAGSHTTPSourceMirrorSyncerResolvesCredentialRefOnlyAtRuntime(t *testing.T) {
	ctx := context.Background()
	resolver := &fakeSourceMirrorCredentialResolver{credential: SourceMirrorCredential{Username: "x-access-token", Password: "fixture-nonsecret-token"}}
	syncer := AGSHTTPSourceMirrorSyncer{
		BaseURL:            "://invalid-target",
		CredentialResolver: resolver,
		WorkDir:            t.TempDir(),
		MaxAttempts:        1,
		RetryDelay:         time.Millisecond,
	}

	_, err := syncer.SyncSourceMirror(ctx, SourceMirrorSyncRequest{
		WorkspaceID:          "wsp_1",
		RemoteURL:            "https://93.184.216.34/org/repo.git",
		Mirror:               kb.SourceMirror{UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit},
		DefaultRef:           "main",
		CredentialRef:        "automation-secret://GITHUB_TOKEN",
		CredentialRepository: repos.RepositoryIdentity{FullName: "wsp_1/agentteam"},
		Target:               repos.RepositoryIdentity{FullName: "wsp_1/agentteam"},
	})

	require.Error(t, err)
	require.Equal(t, SourceMirrorCredentialRequest{
		WorkspaceID:     "wsp_1",
		CredentialRef:   "automation-secret://GITHUB_TOKEN",
		RemoteURL:       "https://93.184.216.34/org/repo.git",
		UpstreamType:    kb.SourceMirrorUpstreamTypeGenericGit,
		RepositoryOwner: "wsp_1",
		RepositoryName:  "agentteam",
	}, resolver.request)
	auth, err := sourceMirrorGitAuth(SourceMirrorCredential{Password: "fixture-nonsecret-token"})
	require.NoError(t, err)
	require.NotNil(t, auth)
}

func TestAGSHTTPSourceMirrorSyncerAllowsPrivateHostWhenRequested(t *testing.T) {
	ctx := context.Background()
	syncer := AGSHTTPSourceMirrorSyncer{
		BaseURL:     "://invalid-target",
		WorkDir:     t.TempDir(),
		MaxAttempts: 1,
		RetryDelay:  time.Millisecond,
	}

	_, err := syncer.SyncSourceMirror(ctx, SourceMirrorSyncRequest{
		WorkspaceID:       "wsp_1",
		RemoteURL:         "https://10.0.0.5/org/repo.git",
		Mirror:            kb.SourceMirror{UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit},
		DefaultRef:        "main",
		AllowPrivateHosts: true,
		Target:            repos.RepositoryIdentity{Owner: "wsp_1", Name: "agentteam", FullName: "wsp_1/agentteam"},
	})

	require.Error(t, err)
	require.ErrorContains(t, err, "AGS base URL")
}

func TestAGSHTTPSourceMirrorSyncerTargetRejectsNonHTTPExternalProtocols(t *testing.T) {
	for _, baseURL := range []string{
		"file:///tmp/ags",
		"ssh://ags.example.test",
		"git://ags.example.test",
	} {
		_, err := (AGSHTTPSourceMirrorSyncer{BaseURL: baseURL}).targetGitURL(repos.RepositoryIdentity{FullName: "wsp_1/agentteam"})
		require.EqualError(t, err, "AGS base URL must be an absolute http(s) URL", baseURL)
	}
}

func TestAGSHTTPSourceMirrorSyncerTargetRequiresRegisteredEmbeddedCapability(t *testing.T) {
	_, err := (AGSHTTPSourceMirrorSyncer{BaseURL: "meshxags://unregistered.embedded.invalid"}).targetGitURL(repos.RepositoryIdentity{FullName: "wsp_1/agentteam"})
	require.EqualError(t, err, "AGS base URL must be an absolute http(s) URL")
}

type fakePublicRepoCacheResolver struct {
	result PublicRepoCacheResolution
	found  bool
	remote string
}

func (r *fakePublicRepoCacheResolver) ResolveReady(_ context.Context, remoteURL string) (PublicRepoCacheResolution, bool, error) {
	r.remote = remoteURL
	return r.result, r.found, nil
}

func TestAGSHTTPSourceMirrorSyncerUsesReadyPublicRepoCache(t *testing.T) {
	root := t.TempDir()
	var logs bytes.Buffer
	cachePath := filepath.Join(root, "cache.git")
	writeSourceMirrorFixtureRepository(t, cachePath)
	resolver := &fakePublicRepoCacheResolver{result: PublicRepoCacheResolution{ID: "prc_cache", Path: cachePath}, found: true}
	syncer := AGSHTTPSourceMirrorSyncer{
		BaseURL:         "https://ags.example.test",
		WorkDir:         filepath.Join(root, "work"),
		MaxAttempts:     1,
		PublicRepoCache: resolver,
		Logger:          slog.New(slog.NewJSONHandler(&logs, nil)),
	}

	_, err := syncer.SyncSourceMirror(context.Background(), SourceMirrorSyncRequest{
		WorkspaceID:     "wsp_1",
		KnowledgeBaseID: "kb_1",
		RemoteURL:       "https://93.184.216.34/org/repo.git",
		Mirror:          kb.SourceMirror{UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit},
		Source:          kb.Source{ID: "src_1"},
		Target:          repos.RepositoryIdentity{FullName: "wsp_1/agentteam"},
	})

	require.ErrorContains(t, err, "Authorization is required")
	require.Equal(t, "https://93.184.216.34/org/repo.git", resolver.remote)
	require.Contains(t, logs.String(), `"msg":"public repository cache hit"`)
	require.Contains(t, logs.String(), `"cache_id":"prc_cache"`)
	require.Contains(t, logs.String(), `"clone_source":"cache"`)
	require.Contains(t, logs.String(), `"workspace_id":"wsp_1"`)
	require.Contains(t, logs.String(), `"knowledge_base_id":"kb_1"`)
	require.Contains(t, logs.String(), `"source_id":"src_1"`)
	require.NotContains(t, logs.String(), "93.184.216.34")
	require.NotContains(t, logs.String(), cachePath)
}

func TestAGSHTTPSourceMirrorSyncerDoesNotUseCacheForCredentialedRemote(t *testing.T) {
	resolver := &fakePublicRepoCacheResolver{result: PublicRepoCacheResolution{ID: "prc_private", Path: t.TempDir()}, found: true}
	syncer := AGSHTTPSourceMirrorSyncer{
		BaseURL:            "://invalid-target",
		WorkDir:            t.TempDir(),
		MaxAttempts:        1,
		CredentialResolver: &fakeSourceMirrorCredentialResolver{credential: SourceMirrorCredential{Password: "fixture-token"}},
		PublicRepoCache:    resolver,
	}

	_, err := syncer.SyncSourceMirror(context.Background(), SourceMirrorSyncRequest{
		RemoteURL:            "https://93.184.216.34/org/repo.git",
		Mirror:               kb.SourceMirror{UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit},
		CredentialRef:        "automation-secret://TOKEN",
		CredentialRepository: repos.RepositoryIdentity{FullName: "wsp_1/agentteam"},
		Target:               repos.RepositoryIdentity{FullName: "wsp_1/agentteam"},
	})

	require.ErrorContains(t, err, "AGS base URL")
	require.Empty(t, resolver.remote)
}

func TestAGSHTTPSourceMirrorPreflightReturnsFriendlyAuthenticationError(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		http.Error(w, "authentication required", http.StatusUnauthorized)
	}))
	t.Cleanup(server.Close)

	err := (AGSHTTPSourceMirrorSyncer{}).PreflightSourceMirror(context.Background(), SourceMirrorPreflightRequest{
		RemoteURL:         server.URL + "/pingcap/agentteam.git",
		UpstreamType:      kb.SourceMirrorUpstreamTypeGenericGit,
		AllowPrivateHosts: true,
	})

	require.ErrorIs(t, err, ErrInvalidArgument)
	require.Equal(t, "source_mirror_authentication_failed", Code(err))
	require.Equal(t, "Could not access the repository. Check the URL and provide an access token with read permission.", err.Error())
}

func TestAGSHTTPSourceMirrorPreflightReturnsFriendlyRepositoryNotFoundError(t *testing.T) {
	server := httptest.NewServer(http.NotFoundHandler())
	t.Cleanup(server.Close)

	err := (AGSHTTPSourceMirrorSyncer{}).PreflightSourceMirror(context.Background(), SourceMirrorPreflightRequest{
		RemoteURL:         server.URL + "/pingcap/missing.git",
		UpstreamType:      kb.SourceMirrorUpstreamTypeGenericGit,
		AllowPrivateHosts: true,
	})

	require.ErrorIs(t, err, ErrInvalidArgument)
	require.Equal(t, "source_mirror_repository_not_found", Code(err))
	require.Equal(t, "Repository was not found. Check the repository URL and access token.", err.Error())
}

func TestAGSHTTPSourceMirrorPreflightReturnsFriendlyUnreachableError(t *testing.T) {
	err := sourceMirrorPreflightError(assertionError("dial tcp 127.0.0.1:1: connect: connection refused"))

	require.ErrorIs(t, err, ErrInvalidArgument)
	require.Equal(t, "source_mirror_unreachable", Code(err))
	require.Equal(t, "Could not reach the repository. Check the URL, network connection, and proxy settings.", err.Error())
}

func TestAGSHTTPSourceMirrorPreflightReturnsFriendlyFallbackError(t *testing.T) {
	err := sourceMirrorPreflightError(assertionError("unexpected upstream response"))

	require.ErrorIs(t, err, ErrInvalidArgument)
	require.Equal(t, "source_mirror_preflight_failed", Code(err))
	require.Equal(t, "Could not verify the repository. Check the repository URL and access token, then try again.", err.Error())
}

func TestGoGitSourceMirrorPushUsesConfiguredAGSAuthHeader(t *testing.T) {
	ctx := context.Background()
	requests := make(chan http.Header, 4)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requests <- r.Header.Clone()
		http.Error(w, `{"message":"Bad credentials"}`, http.StatusUnauthorized)
	}))
	t.Cleanup(server.Close)

	root := t.TempDir()
	sourceBare := filepath.Join(root, "source.git")
	writeSourceMirrorFixtureRepository(t, sourceBare)
	repo, err := cloneGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, sourceBare, filepath.Join(root, "mirror.git"), nil, transport.ProxyOptions{})
	require.NoError(t, err)

	err = pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, repo, filepath.Join(root, "mirror.git"), server.URL+"/wsp_1/agentteam.git", gitAuthorizationAuth("X-AgentTeam-JWT", "Bearer internal.jwt."), agsGitMirrorTargetControl{})

	require.Error(t, err)
	var captured http.Header
	select {
	case captured = <-requests:
	default:
		require.Fail(t, "expected go-git push to reach test server")
	}
	require.Equal(t, "Bearer internal.jwt.", captured.Get("X-AgentTeam-JWT"))
	require.Empty(t, captured.Get("Authorization"))
}

func TestAutomationSecretSourceMirrorCredentialResolverDecryptsScopedSecret(t *testing.T) {
	ctx := context.Background()
	db, err := openServiceTestSQLiteDB(t)
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(&automation.AutomationSecret{}))
	store := automation.New(db)
	key := []byte("0123456789abcdef0123456789abcdef")
	codec, err := secrets.NewAESGCMCodec(key)
	require.NoError(t, err)
	ciphertext, err := codec.Seal("ghp_fixture_token")
	require.NoError(t, err)
	_, err = store.UpsertAutomationSecret(ctx, automation.AutomationSecret{
		WorkspaceID:     "wsp_1",
		RepositoryOwner: "wsp_1",
		RepositoryName:  "agentteam",
		Name:            "GITHUB_TOKEN",
		ValueCiphertext: ciphertext,
		CreatedBy:       "owner",
		UpdatedBy:       "owner",
		CreatedAt:       time.Now().UTC(),
		UpdatedAt:       time.Now().UTC(),
	})
	require.NoError(t, err)
	resolver := AutomationSecretSourceMirrorCredentialResolver{Store: store, CredentialsKey: key}

	credential, err := resolver.ResolveSourceMirrorCredential(ctx, SourceMirrorCredentialRequest{
		WorkspaceID:     "wsp_1",
		CredentialRef:   "automation-secret://GITHUB_TOKEN",
		RepositoryOwner: "wsp_1",
		RepositoryName:  "agentteam",
	})

	require.NoError(t, err)
	require.Equal(t, SourceMirrorCredential{Username: "x-access-token", Password: "ghp_fixture_token"}, credential)
}

func TestSourceMirrorGitAuthUsesGitHubCompatibleBasicAuth(t *testing.T) {
	auth, err := sourceMirrorGitAuth(SourceMirrorCredential{Password: "ghp_fixture_token"})
	require.NoError(t, err)

	request, err := http.NewRequest(http.MethodGet, "https://github.com/pingcap/agentteam.git/info/refs", nil)
	require.NoError(t, err)
	httpAuth, ok := auth.(interface{ SetAuth(*http.Request) })
	require.True(t, ok)
	httpAuth.SetAuth(request)
	username, password, ok := request.BasicAuth()
	require.True(t, ok)
	require.Equal(t, "x-access-token", username)
	require.Equal(t, "ghp_fixture_token", password)
	require.NotContains(t, auth.String(), "ghp_fixture_token")
}

func TestAutomationSecretSourceMirrorCredentialResolverRejectsCrossRepositoryRef(t *testing.T) {
	_, err := parseSourceMirrorAutomationSecretRef("automation-secret://other/agentteam/GITHUB_TOKEN", "wsp_1", "agentteam")
	require.Error(t, err)
	require.Contains(t, err.Error(), "target repository scope")

	_, err = parseSourceMirrorAutomationSecretRef("secret://kb/source-token", "wsp_1", "agentteam")
	require.Error(t, err)
	require.Contains(t, err.Error(), "automation secret")
}

func TestGoGitSourceMirrorPushMirrorsBranchesTagsAndPrunes(t *testing.T) {
	ctx := context.Background()
	root := t.TempDir()
	sourceWorktree := filepath.Join(root, "source-worktree")
	sourceRepo, err := gogit.PlainInit(sourceWorktree, false)
	require.NoError(t, err)
	mainCommit := commitSourceMirrorFixture(t, sourceRepo, sourceWorktree, "main", "main content")
	worktree, err := sourceRepo.Worktree()
	require.NoError(t, err)
	require.NoError(t, worktree.Checkout(&gogit.CheckoutOptions{Branch: plumbing.NewBranchReferenceName("feature"), Create: true}))
	featureCommit := commitSourceMirrorFixture(t, sourceRepo, sourceWorktree, "feature", "feature content")
	tagRef := plumbing.NewHashReference(plumbing.NewTagReferenceName("v1"), mainCommit)
	require.NoError(t, sourceRepo.Storer.SetReference(tagRef))
	annotatedTag, err := sourceRepo.CreateTag("v2", mainCommit, &gogit.CreateTagOptions{
		Tagger:  &object.Signature{Name: "MeshX Test", Email: "meshx@example.local", When: time.Now().UTC()},
		Message: "annotated tag",
	})
	require.NoError(t, err)
	require.NoError(t, sourceRepo.Storer.SetReference(plumbing.NewHashReference(plumbing.ReferenceName("refs/changes/1"), featureCommit)))

	sourceBare := filepath.Join(root, "source.git")
	_, err = gogit.PlainInit(sourceBare, true)
	require.NoError(t, err)
	_, err = sourceRepo.CreateRemote(&config.RemoteConfig{Name: "origin", URLs: []string{sourceBare}})
	require.NoError(t, err)
	require.NoError(t, sourceRepo.Push(&gogit.PushOptions{RemoteName: "origin", RefSpecs: []config.RefSpec{
		"+refs/*:refs/*",
	}}))

	targetBare := filepath.Join(root, "target.git")
	targetRepo, err := gogit.PlainInit(targetBare, true)
	require.NoError(t, err)
	require.NoError(t, targetRepo.Storer.SetReference(plumbing.NewHashReference(plumbing.NewBranchReferenceName("stale"), mainCommit)))
	targetCustomRef := plumbing.ReferenceName("refs/changes/local")
	require.NoError(t, targetRepo.Storer.SetReference(plumbing.NewHashReference(targetCustomRef, mainCommit)))
	installRejectCustomRefsHook(t, targetBare)

	repo, err := cloneGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, sourceBare, filepath.Join(root, "mirror.git"), nil, transport.ProxyOptions{})
	require.NoError(t, err)
	require.Equal(t, featureCommit.String(), sourceMirrorMirroredCommit(repo, "feature"))
	require.Equal(t, mainCommit.String(), sourceMirrorMirroredCommit(repo, "v2"))
	require.NoError(t, pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, repo, filepath.Join(root, "mirror.git"), targetBare, nil, agsGitMirrorTargetControl{}))
	require.NoError(t, pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, repo, filepath.Join(root, "mirror.git"), targetBare, nil, agsGitMirrorTargetControl{}))

	targetRepo, err = gogit.PlainOpen(targetBare)
	require.NoError(t, err)
	mainRef, err := targetRepo.Reference(plumbing.NewBranchReferenceName("master"), true)
	require.NoError(t, err)
	require.Equal(t, mainCommit, mainRef.Hash())
	featureRef, err := targetRepo.Reference(plumbing.NewBranchReferenceName("feature"), true)
	require.NoError(t, err)
	require.Equal(t, featureCommit, featureRef.Hash())
	tag, err := targetRepo.Reference(plumbing.NewTagReferenceName("v1"), true)
	require.NoError(t, err)
	require.Equal(t, mainCommit, tag.Hash())
	annotated, err := targetRepo.Reference(plumbing.NewTagReferenceName("v2"), true)
	require.NoError(t, err)
	require.Equal(t, annotatedTag.Hash(), annotated.Hash())
	_, err = targetRepo.Reference(plumbing.ReferenceName("refs/changes/1"), true)
	require.Error(t, err, "source custom refs must not be pushed into the AGS mirror target")
	preservedCustomRef, err := targetRepo.Reference(targetCustomRef, true)
	require.NoError(t, err, "target custom refs are outside KB mirror ownership and must not be pruned")
	require.Equal(t, mainCommit, preservedCustomRef.Hash())
	_, err = targetRepo.Reference(plumbing.NewBranchReferenceName("stale"), true)
	require.Error(t, err)
}

func TestSourceMirrorPushFallsBackToNativeGitForBareTreePath(t *testing.T) {
	for _, test := range []struct {
		name          string
		treeEntryName string
		directError   string
	}{
		{name: "single backslash", treeEntryName: `\`, directError: `invalid path: "\\"`},
		{name: "control character", treeEntryName: "report\t.md", directError: "contains control character"},
	} {
		t.Run(test.name, func(t *testing.T) {
			testSourceMirrorPushFallsBackToNativeGitForBareTreePath(t, test.treeEntryName, test.directError)
		})
	}
}

func testSourceMirrorPushFallsBackToNativeGitForBareTreePath(t *testing.T, treeEntryName, directError string) {
	ctx := t.Context()
	root := t.TempDir()
	sourceBare := filepath.Join(root, "source.git")
	sourceRepo, err := gogit.PlainInit(sourceBare, true)
	require.NoError(t, err)

	blob := &plumbing.MemoryObject{}
	blob.SetType(plumbing.BlobObject)
	_, err = blob.Write([]byte("content"))
	require.NoError(t, err)
	blobHash, err := sourceRepo.Storer.SetEncodedObject(blob)
	require.NoError(t, err)
	tree := &plumbing.MemoryObject{}
	require.NoError(t, (&object.Tree{Entries: []object.TreeEntry{{Name: treeEntryName, Mode: 0o100644, Hash: blobHash}}}).Encode(tree))
	treeHash, err := sourceRepo.Storer.SetEncodedObject(tree)
	require.NoError(t, err)
	commit := &plumbing.MemoryObject{}
	require.NoError(t, (&object.Commit{
		Author:    object.Signature{Name: "MeshX Test", Email: "meshx@example.local", When: time.Now().UTC()},
		Committer: object.Signature{Name: "MeshX Test", Email: "meshx@example.local", When: time.Now().UTC()},
		Message:   "special path",
		TreeHash:  treeHash,
	}).Encode(commit))
	commitHash, err := sourceRepo.Storer.SetEncodedObject(commit)
	require.NoError(t, err)
	mainRef := plumbing.NewBranchReferenceName("main")
	require.NoError(t, sourceRepo.Storer.SetReference(plumbing.NewHashReference(mainRef, commitHash)))
	require.NoError(t, sourceRepo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, mainRef)))

	mirrorPath := filepath.Join(root, "mirror.git")
	repo, err := cloneGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1}, sourceBare, mirrorPath, nil, transport.ProxyOptions{})
	require.NoError(t, err, "go-git mirror clone stores the tree without materialising it")
	fileTargetBare := filepath.Join(root, "file-target.git")
	_, err = gogit.PlainInit(fileTargetBare, true)
	require.NoError(t, err)
	require.NoError(t, pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1}, repo, mirrorPath, fileTargetBare, nil, agsGitMirrorTargetControl{}))
	fileTargetRepo, err := gogit.PlainOpen(fileTargetBare)
	require.NoError(t, err)
	fileTargetRef, err := fileTargetRepo.Reference(plumbing.NewBranchReferenceName("main"), true)
	require.NoError(t, err)
	require.Equal(t, commitHash, fileTargetRef.Hash())

	targetRoot := filepath.Join(root, "target")
	targetBare := filepath.Join(targetRoot, "owner", "target.git")
	require.NoError(t, os.MkdirAll(filepath.Dir(targetBare), 0o755))
	_, err = gogit.PlainInit(targetBare, true)
	require.NoError(t, err)
	requireCommandOutput(t, "git", "--git-dir", targetBare, "config", "http.receivepack", "true")
	backendPath := strings.TrimSpace(string(requireCommandOutput(t, "git", "--exec-path")))
	const authHeader = "Bearer internal.jwt."
	targetHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("X-AgentTeam-JWT") != authHeader {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		(&cgi.Handler{
			Path: filepath.Join(backendPath, "git-http-backend"),
			Dir:  root,
			Env: []string{
				"GIT_PROJECT_ROOT=" + targetRoot,
				"GIT_HTTP_EXPORT_ALL=1",
				"PATH_INFO=" + r.URL.Path,
			},
			InheritEnv: []string{"PATH", "HOME", "USER", "LOGNAME", "TMPDIR", "LANG", "LC_ALL"},
		}).ServeHTTP(w, r)
	})
	targetBaseURL, unregister, err := gittransport.Register(targetHandler)
	require.NoError(t, err)
	t.Cleanup(unregister)
	targetURL := targetBaseURL + "/owner/target.git"
	targetAuth := gitAuthorizationAuth("X-AgentTeam-JWT", authHeader)
	directRemote := gogit.NewRemote(repo.Storer, &config.RemoteConfig{Name: "target", URLs: []string{targetURL}})
	directErr := directRemote.PushContext(ctx, &gogit.PushOptions{RemoteName: "target", Auth: targetAuth, RefSpecs: sourceMirrorPushRefSpecs()})
	require.ErrorContains(t, directErr, directError)

	target := repos.RepositoryIdentity{FullName: "owner/target"}
	pusher := sourceMirrorNativePusherFunc(func(ctx context.Context, request NativeSourceMirrorPushRequest) error {
		require.Equal(t, mirrorPath, request.SourceGitDir)
		require.Equal(t, target, request.Target)
		require.Equal(t, []string{
			"+refs/heads/*:refs/heads/*",
			"+refs/tags/*:refs/tags/*",
		}, request.RefSpecs)
		args := []string{"--git-dir", request.SourceGitDir, "push", "--no-verify", "--", targetBare}
		args = append(args, request.RefSpecs...)
		output, err := exec.CommandContext(ctx, "git", args...).CombinedOutput()
		if err != nil {
			return fmt.Errorf("push test mirror: %w: %s", err, output)
		}
		return nil
	})
	require.NoError(t, pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1}, repo, mirrorPath, targetURL, targetAuth, agsGitMirrorTargetControl{
		Repository:   target,
		NativePusher: pusher,
	}))
	targetRepo, err := gogit.PlainOpen(targetBare)
	require.NoError(t, err)
	targetRef, err := targetRepo.Reference(plumbing.NewBranchReferenceName("main"), true)
	require.NoError(t, err)
	require.Equal(t, commitHash, targetRef.Hash())
}

type sourceMirrorNativePusherFunc func(context.Context, NativeSourceMirrorPushRequest) error

func (f sourceMirrorNativePusherFunc) PushSourceMirror(ctx context.Context, request NativeSourceMirrorPushRequest) error {
	return f(ctx, request)
}

func requireCommandOutput(t *testing.T, name string, args ...string) []byte {
	t.Helper()
	output, err := exec.Command(name, args...).Output()
	require.NoError(t, err)
	return output
}

func TestSourceMirrorResolvedDefaultRef(t *testing.T) {
	repo, err := gogit.PlainInit(t.TempDir(), true)
	require.NoError(t, err)
	mainHash := plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
	trunkHash := plumbing.NewHash("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
	require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(plumbing.NewBranchReferenceName("main"), mainHash)))
	require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(plumbing.NewBranchReferenceName("trunk"), trunkHash)))
	require.NoError(t, repo.Storer.SetReference(plumbing.NewSymbolicReference(plumbing.HEAD, plumbing.NewBranchReferenceName("trunk"))))

	resolved, auto, err := sourceMirrorResolvedDefaultRef(repo, "", nil)
	require.NoError(t, err)
	require.Equal(t, "trunk", resolved)
	require.True(t, auto)

	explicit := true
	resolved, auto, err = sourceMirrorResolvedDefaultRef(repo, "refs/heads/main", &explicit)
	require.NoError(t, err)
	require.Equal(t, "main", resolved)
	require.False(t, auto)

	_, _, err = sourceMirrorResolvedDefaultRef(repo, "release", &explicit)
	require.ErrorContains(t, err, "does not exist")
	require.Empty(t, sourceMirrorMirroredCommit(repo, "release"), "an invalid explicit branch must not fall back to HEAD")

	autoConfigured := false
	resolved, auto, err = sourceMirrorResolvedDefaultRef(repo, "stale-main", &autoConfigured)
	require.NoError(t, err)
	require.Equal(t, "trunk", resolved)
	require.True(t, auto)

	resolved, auto, err = sourceMirrorResolvedDefaultRef(repo, "stale-main", nil)
	require.NoError(t, err, "legacy rows created before explicit intent was persisted must repair stale defaults")
	require.Equal(t, "trunk", resolved)
	require.True(t, auto)
}

func TestCompleteSourceMirrorSyncPersistsRepairedLegacyDefaultRef(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{
		MirroredCommit:     "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		ResolvedDefaultRef: "trunk",
		DefaultRefAuto:     true,
	}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	job, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_legacy_default", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: svc.now(), UpdatedAt: svc.now(),
	})
	require.NoError(t, err)
	_, err = svc.store.ClaimQueuedJobsFair(ctx, KBJobTypeSyncSourceMirror, "worker", time.Minute, 1, svc.now())
	require.NoError(t, err)
	job, err = svc.store.GetJob(ctx, "wsp_1", job.ID)
	require.NoError(t, err)
	require.NoError(t, svc.runSourceMirrorSyncJob(ctx, job))

	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, "trunk", source.DefaultRef)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, "trunk", mirror.DefaultRef)
	require.NotNil(t, mirror.DefaultRefExplicit)
	require.False(t, *mirror.DefaultRefExplicit)
	repositoryRow, err := svc.repositoryStore.Get(ctx, "wsp_1", reposstore.ResourceTypeKBSource, "src_1", reposstore.RepoRoleMirror)
	require.NoError(t, err)
	require.Equal(t, "trunk", repositoryRow.DefaultRef)
}

func TestPrepareSourceMirrorMutationRecordsDefaultRefIntent(t *testing.T) {
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	source := kb.Source{ID: "src_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", DefaultRef: "legacy-main"}

	automatic, err := svc.prepareSourceMirrorMutation(context.Background(), source, KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: "https://93.184.216.34/org/repo.git", DefaultRefSet: true,
	}, "owner", svc.now())
	require.NoError(t, err)
	require.Empty(t, automatic.DefaultRef)
	require.NotNil(t, automatic.DefaultRefExplicit)
	require.False(t, *automatic.DefaultRefExplicit)

	explicit, err := svc.prepareSourceMirrorMutation(context.Background(), source, KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: "https://93.184.216.34/org/repo.git", DefaultRef: "release", DefaultRefSet: true,
	}, "owner", svc.now())
	require.NoError(t, err)
	require.Equal(t, "release", explicit.DefaultRef)
	require.NotNil(t, explicit.DefaultRefExplicit)
	require.True(t, *explicit.DefaultRefExplicit)
}

func TestSourceMirrorResolvedDefaultRefUsesOnlyBranchForDanglingHead(t *testing.T) {
	repo, err := gogit.PlainInit(t.TempDir(), true)
	require.NoError(t, err)
	require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(
		plumbing.NewBranchReferenceName("legacy"),
		plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
	)))

	resolved, _, err := sourceMirrorResolvedDefaultRef(repo, "", nil)
	require.NoError(t, err)
	require.Equal(t, "legacy", resolved)

	require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(
		plumbing.NewBranchReferenceName("other"),
		plumbing.NewHash("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
	)))
	_, _, err = sourceMirrorResolvedDefaultRef(repo, "", nil)
	require.ErrorContains(t, err, "HEAD does not identify")
}

func TestSourceMirrorResolvedDefaultRefPDRemote(t *testing.T) {
	if os.Getenv("AGENTTEAM_TEST_GITHUB_PD_REMOTE") != "1" {
		t.Skip("set AGENTTEAM_TEST_GITHUB_PD_REMOTE=1 to verify the production resolver against github.com/tikv/pd")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	repo, err := gogit.PlainCloneContext(ctx, filepath.Join(t.TempDir(), "pd"), false, &gogit.CloneOptions{
		URL: "https://github.com/tikv/pd.git", Depth: 1, SingleBranch: true,
	})
	require.NoError(t, err)

	resolved, auto, err := sourceMirrorResolvedDefaultRef(repo, "main", nil)
	require.NoError(t, err, "legacy stale main must repair from the PD remote HEAD")
	require.Equal(t, "master", resolved)
	require.True(t, auto)

	automatic := false
	resolved, auto, err = sourceMirrorResolvedDefaultRef(repo, "main", &automatic)
	require.NoError(t, err, "persisted automatic main must repair from the PD remote HEAD")
	require.Equal(t, "master", resolved)
	require.True(t, auto)

	explicit := true
	_, _, err = sourceMirrorResolvedDefaultRef(repo, "main", &explicit)
	require.ErrorContains(t, err, "does not exist", "an explicit missing branch must stay strict")
}

func TestGoGitSourceMirrorSkipsLegacyDefaultBranchDuringPrune(t *testing.T) {
	ctx := context.Background()
	root := t.TempDir()
	sourceWorktree := filepath.Join(root, "source-worktree")
	sourceRepo, err := gogit.PlainInitWithOptions(sourceWorktree, &gogit.PlainInitOptions{InitOptions: gogit.InitOptions{DefaultBranch: plumbing.NewBranchReferenceName("master")}})
	require.NoError(t, err)
	masterCommit := commitSourceMirrorFixture(t, sourceRepo, sourceWorktree, "master", "master content")

	sourceBare := filepath.Join(root, "source.git")
	_, err = gogit.PlainInitWithOptions(sourceBare, &gogit.PlainInitOptions{Bare: true, InitOptions: gogit.InitOptions{DefaultBranch: plumbing.NewBranchReferenceName("master")}})
	require.NoError(t, err)
	_, err = sourceRepo.CreateRemote(&config.RemoteConfig{Name: "origin", URLs: []string{sourceBare}})
	require.NoError(t, err)
	require.NoError(t, sourceRepo.Push(&gogit.PushOptions{RemoteName: "origin", RefSpecs: []config.RefSpec{"+refs/*:refs/*"}}))

	targetBare := filepath.Join(root, "target.git")
	targetRepo, err := gogit.PlainInitWithOptions(targetBare, &gogit.PlainInitOptions{Bare: true, InitOptions: gogit.InitOptions{DefaultBranch: plumbing.NewBranchReferenceName("main")}})
	require.NoError(t, err)
	require.NoError(t, targetRepo.Storer.SetReference(plumbing.NewHashReference(plumbing.NewBranchReferenceName("main"), masterCommit)))
	staleRef := plumbing.NewBranchReferenceName("stale")
	require.NoError(t, targetRepo.Storer.SetReference(plumbing.NewHashReference(staleRef, masterCommit)))

	control := agsGitMirrorTargetControl{Repository: repos.RepositoryIdentity{FullName: "wsp_1/jio", DefaultBranch: "main"}}
	repo, err := cloneGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, sourceBare, filepath.Join(root, "mirror.git"), nil, transport.ProxyOptions{})
	require.NoError(t, err)

	require.NoError(t, pushGoGitMirrorWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 1, Delay: time.Millisecond}, repo, filepath.Join(root, "mirror.git"), targetBare, nil, control))

	targetRepo, err = gogit.PlainOpen(targetBare)
	require.NoError(t, err)
	masterRef, err := targetRepo.Reference(plumbing.NewBranchReferenceName("master"), true)
	require.NoError(t, err)
	require.Equal(t, masterCommit, masterRef.Hash())
	_, err = targetRepo.Reference(staleRef, true)
	require.Error(t, err)
	_, err = targetRepo.Reference(plumbing.NewBranchReferenceName("main"), true)
	require.NoError(t, err)
}

func TestGoGitSourceMirrorRetryClassification(t *testing.T) {
	require.True(t, isRetryableGitMirrorError(assertionError("unexpected EOF")))
	require.True(t, isRetryableGitMirrorError(assertionError("The requested URL returned error: 503")))
	require.False(t, isRetryableGitMirrorError(assertionError("unexpected requesting http://127.0.0.1:50321/repo.git status code: 500")))
	require.False(t, isRetryableGitMirrorError(assertionError("authentication required")))
	require.False(t, isRetryableGitMirrorError(assertionError("repository not found")))
}

func TestGoGitBareTreePathFallbackClassification(t *testing.T) {
	require.True(t, isGoGitBareTreePathError(errors.New(`invalid path: "\\"`)))
	require.True(t, isGoGitBareTreePathError(errors.New("invalid path \"report\\t.md\": contains control character")))
	require.False(t, isGoGitBareTreePathError(errors.New(`invalid path "../report.md": cannot use ".."`)))
	require.False(t, isGoGitBareTreePathError(errors.New("authentication required")))
	require.False(t, isGoGitBareTreePathError(nil))
}

func TestSourceMirrorCloneProxyOptions(t *testing.T) {
	require.Equal(t, transport.ProxyOptions{}, sourceMirrorCloneProxyOptions(""))
	require.Equal(
		t,
		transport.ProxyOptions{URL: "http://127.0.0.1:8888"},
		sourceMirrorCloneProxyOptions(" http://127.0.0.1:8888 "),
	)
}

func TestGoGitSourceMirrorCleansPartialCloneBeforeRetry(t *testing.T) {
	ctx := context.Background()
	root := t.TempDir()
	repoDir := filepath.Join(root, "repo.git")
	attempts := 0
	err := runGoGitMirrorOperationWithRetry(ctx, gitMirrorRetryOptions{MaxAttempts: 2, Delay: time.Millisecond}, "clone source mirror remote", true, repoDir, func() error {
		attempts++
		if attempts == 1 {
			require.NoError(t, os.MkdirAll(repoDir, 0o700))
			return assertionError("unexpected EOF")
		}
		_, statErr := os.Stat(repoDir)
		require.True(t, os.IsNotExist(statErr), "partial clone directory must be removed before retry")
		return nil
	})

	require.NoError(t, err)
	require.Equal(t, 2, attempts)
}

func commitSourceMirrorFixture(t *testing.T, repo *gogit.Repository, worktreeDir string, message string, content string) plumbing.Hash {
	t.Helper()
	require.NoError(t, os.WriteFile(filepath.Join(worktreeDir, "file.txt"), []byte(content), 0o644))
	worktree, err := repo.Worktree()
	require.NoError(t, err)
	_, err = worktree.Add("file.txt")
	require.NoError(t, err)
	hash, err := worktree.Commit(message, &gogit.CommitOptions{Author: &object.Signature{Name: "MeshX Test", Email: "meshx@example.local", When: time.Now().UTC()}})
	require.NoError(t, err)
	return hash
}

func installRejectCustomRefsHook(t *testing.T, barePath string) {
	t.Helper()
	hooksDir := filepath.Join(barePath, "hooks")
	require.NoError(t, os.MkdirAll(hooksDir, 0o755))
	hook := `#!/bin/sh
while read -r oldrev newrev refname; do
    case "$refname" in
        refs/heads/*|refs/tags/*)
            continue
            ;;
    esac
    echo "custom ref update rejected: $refname" >&2
    exit 1
done
exit 0
`
	require.NoError(t, os.WriteFile(filepath.Join(hooksDir, "pre-receive"), []byte(hook), 0o755))
}

func writeSourceMirrorFixtureRepository(t *testing.T, barePath string) plumbing.Hash {
	t.Helper()
	root := t.TempDir()
	worktreePath := filepath.Join(root, "worktree")
	repo, err := gogit.PlainInit(worktreePath, false)
	require.NoError(t, err)
	commit := commitSourceMirrorFixture(t, repo, worktreePath, "initial", "content")
	_, err = gogit.PlainInit(barePath, true)
	require.NoError(t, err)
	_, err = repo.CreateRemote(&config.RemoteConfig{Name: "origin", URLs: []string{barePath}})
	require.NoError(t, err)
	require.NoError(t, repo.Push(&gogit.PushOptions{RemoteName: "origin", RefSpecs: []config.RefSpec{"+refs/*:refs/*"}}))
	return commit
}

type assertionError string

func (e assertionError) Error() string { return string(e) }

func TestKBServiceSourceMirrorLeaseDurationExceedsSyncTimeout(t *testing.T) {
	require.Equal(t, 45*time.Minute, SourceMirrorLeaseDuration(30*time.Minute))
	require.Greater(t, SourceMirrorLeaseDuration(time.Minute), time.Minute)
}

func TestKBServiceDisableSourceMirrorCancelsQueuedJobs(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	_, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	jobs, err := svc.store.ListSourceJobs(ctx, "wsp_1", "kb_1", "src_1", KBJobTypeSyncSourceMirror, 1, 10)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	require.Equal(t, KBJobStatusQueued, jobs[0].Status)

	disabled := false
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/repo.git",
		DefaultRef:   "main",
		Enabled:      &disabled,
	})
	require.NoError(t, err)

	jobs, err = svc.store.ListSourceJobs(ctx, "wsp_1", "kb_1", "src_1", KBJobTypeSyncSourceMirror, 1, 10)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	require.Equal(t, KBJobStatusCancelled, jobs[0].Status, "queued job must be cancelled, not failed")

	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.False(t, mirror.Enabled)
	require.Equal(t, kb.SourceMirrorSyncStatusDisabled, mirror.SyncStatus)
	require.Empty(t, mirror.LastError, "disabled mirror must not carry a failure error")

	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.NotEqual(t, KBSourceStatusFailed, source.Status, "source must not be marked failed when mirror is disabled")
}

func TestKBServiceDisableSourceMirrorSkipsInvalidCurrentPersonalToken(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	mirror := seedSourceMirror(t, svc, true)
	mirror.CredentialRef = providertokens.CredentialRef("pat_missing")
	_, err := svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	disabled := false
	updated, err := svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		Enabled: &disabled,
	})

	require.NoError(t, err)
	require.False(t, updated.Enabled)
	require.Equal(t, providertokens.CredentialRef("pat_missing"), updated.CredentialRef)
	require.Equal(t, kb.SourceMirrorSyncStatusDisabled, updated.SyncStatus)
}

func TestKBServiceUpdateSourceMirrorRequiresPersonalTokenOwnerOnReplacement(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	now := svc.now()
	_, err := svc.personalAccessTokenStore.Create(ctx, providertokensstore.PersonalAccessToken{
		ID:                "pat_other",
		WorkspaceID:       "wsp_1",
		OwnerPrincipalID:  "other-user",
		DisplayName:       "Other user's token",
		ProviderType:      providertokens.ProviderGenericGit,
		ProviderBaseURL:   "https://93.184.216.34",
		Username:          "x-access-token",
		AllowedUsagesJSON: `["kb_mirror"]`,
		ValueCiphertext:   "ciphertext",
		Status:            providertokens.StatusActive,
		CreatedBy:         "other-user",
		UpdatedBy:         "other-user",
		CreatedAt:         now,
		UpdatedAt:         now,
	})
	require.NoError(t, err)

	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		CredentialRef: providertokens.CredentialRef("pat_other"),
	})

	require.ErrorContains(t, err, "must be owned by the current user")
	mirror, loadErr := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, loadErr)
	require.Empty(t, mirror.CredentialRef)
}

func TestKBServicePersonalTokenOwnerCheckUsesWorkspaceAccessVerifier(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	err := svc.validatePersonalAccessTokenOwnerCanSyncSource(ctx, "wsp_1", "kb_1", "owner")
	require.ErrorContains(t, err, "workspace access cannot be verified")

	svc.principalWorkspaceAccessVerifier = fakePrincipalWorkspaceAccessVerifier{active: false}

	err = svc.validatePersonalAccessTokenOwnerCanSyncSource(ctx, "wsp_1", "kb_1", "owner")

	require.ErrorContains(t, err, "no longer has workspace access")

	svc.principalWorkspaceAccessVerifier = fakePrincipalWorkspaceAccessVerifier{active: true}
	require.NoError(t, svc.validatePersonalAccessTokenOwnerCanSyncSource(ctx, "wsp_1", "kb_1", "owner"))
}

func TestKBServiceDeleteSourceCancelsActiveJobs(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	_, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	jobs, err := svc.store.ListSourceJobs(ctx, "wsp_1", "kb_1", "src_1", KBJobTypeSyncSourceMirror, 1, 10)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	require.Equal(t, KBJobStatusQueued, jobs[0].Status)

	err = svc.DeleteSource(ctx, "wsp_1", "owner", "kb_1", "src_1")
	require.NoError(t, err)

	job, err := svc.store.GetJob(ctx, "wsp_1", jobs[0].ID)
	require.NoError(t, err)
	require.Equal(t, KBJobStatusCancelled, job.Status, "orphaned job must be cancelled, not failed")
}

func TestKBServiceRunSourceMirrorJobCancelsWhenMirrorDisabled(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: svc.now(), UpdatedAt: svc.now(),
	})
	require.NoError(t, err)

	disabled := false
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/repo.git",
		DefaultRef:   "main",
		Enabled:      &disabled,
	})
	require.NoError(t, err)

	// The job was cancelled by the disable call. ClaimNextQueuedJob should
	// not pick it up because its status is "cancelled".
	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 1, LeaseOwner: "test-worker"})
	require.NoError(t, err)
	require.Equal(t, SourceMirrorRunResult{Claimed: 0}, result)

	job, err := svc.store.GetJob(ctx, "wsp_1", "job_1")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusCancelled, job.Status)

	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, kb.SourceMirrorSyncStatusDisabled, mirror.SyncStatus)
	require.Empty(t, mirror.LastError)
}

func TestKBServiceRunSourceMirrorJobCancelsWhenSourceDeleted(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Create a queued job, then simulate a race: manually re-queue it after
	// the source is deleted so the worker picks up an orphan job.
	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_orphan", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: svc.now().Add(time.Second), UpdatedAt: svc.now().Add(time.Second),
	})
	require.NoError(t, err)

	err = svc.DeleteSource(ctx, "wsp_1", "owner", "kb_1", "src_1")
	require.NoError(t, err)

	// The cancel call in DeleteSource should have cancelled the job.
	job, err := svc.store.GetJob(ctx, "wsp_1", "job_orphan")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusCancelled, job.Status, "orphan job must be cancelled on delete")
}

func TestSourceImportProgressDoesNotRegressForSubsequentSync(t *testing.T) {
	now := time.Date(2026, 6, 12, 1, 2, 3, 0, time.UTC)
	successAt := now.Add(-time.Hour)

	source := kb.Source{
		ID: "src_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1",
		Status: KBSourceStatusReady, SyncStatus: KBSourceStatusReady, IndexStatus: KBSourceStatusReady,
		Mirror: &kb.SourceMirror{
			SourceID: "src_1", Enabled: true,
			SyncStatus:         kb.SourceMirrorSyncStatusQueued,
			LastSuccessAt:      &successAt,
			LastMirroredCommit: fakeSourceMirrorCommit,
		},
	}

	initialImportJob := kb.Job{
		ID: "job_initial", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror,
		Status: KBJobStatusSucceeded, CreatedAt: successAt, UpdatedAt: successAt,
	}
	activeJob := kb.Job{
		ID: "job_sync", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror,
		Status: KBJobStatusQueued, CreatedAt: now, UpdatedAt: now,
	}

	terminalBySource := map[string]kb.Job{"src_1": initialImportJob}
	progress := sourceImportProgress(source, []kb.Job{activeJob}, terminalBySource)
	require.Equal(t, KBSourceImportProgressStateSucceeded, progress.State,
		"subsequent sync of an already-imported source must not regress to queued/importing")
}

// TestEnqueueSourceMirrorSyncDoesNotRegressSourceStatusForImportedSource
// verifies the real end-to-end path: after a successful initial import,
// calling enqueueSourceMirrorSync (as the scheduler does) must NOT change
// source.Status to "queued". If it did, importProgress would fall through
// to sourceProvisioningImportProgress and show "provisioning queued" (10%).
func TestEnqueueSourceMirrorSyncDoesNotRegressSourceStatusForImportedSource(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Simulate a previously successful initial import.
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	successAt := svc.now().Add(-time.Hour)
	mirror.LastSuccessAt = &successAt
	mirror.LastMirroredCommit = "abc123"
	mirror.SyncStatus = kb.SourceMirrorSyncStatusSuccess
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	source.Status = KBSourceStatusReady
	source.SyncStatus = KBSourceStatusReady
	source.IndexStatus = KBSourceStatusReady
	_, err = svc.store.UpdateSource(ctx, source)
	require.NoError(t, err)

	// Enqueue a background sync (scheduler path).
	_, err = svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	// source.Status must still be "ready", not "queued".
	sourceAfter, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, KBSourceStatusReady, sourceAfter.Status,
		"background sync of an already-imported source must not regress source.Status to queued")

	// Now verify importProgress through the real derivation path.
	sourceAfter2, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	sourcesWithMirrors := svc.withSourceMirrors(ctx, []kb.Source{sourceAfter2})
	require.Len(t, sourcesWithMirrors, 1)
	require.NotNil(t, sourcesWithMirrors[0].Mirror, "mirror must be loaded for import progress")
	activeJobs, err := svc.store.ListActiveSourceJobsForSources(ctx, "wsp_1", "kb_1", []string{"src_1"}, kbImportProgressJobTypes())
	require.NoError(t, err)
	terminalJobs, err := svc.store.ListLatestTerminalSourceJobsForSources(ctx, "wsp_1", "kb_1", []string{"src_1"}, kbImportProgressJobTypes())
	require.NoError(t, err)
	terminalBySource := make(map[string]kb.Job)
	for _, j := range terminalJobs {
		terminalBySource[j.SourceID] = j
	}
	progress := sourceImportProgress(sourcesWithMirrors[0], activeJobs, terminalBySource)
	require.NotEqual(t, KBSourceImportProgressStateQueued, progress.State,
		"import progress must not show queued/provisioning for a background sync of an already-imported source")
}

func TestSourceImportProgressShowsQueuedForInitialImport(t *testing.T) {
	now := time.Date(2026, 6, 12, 1, 2, 3, 0, time.UTC)

	source := kb.Source{
		ID: "src_1", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1",
		Status: KBSourceStatusQueued, SyncStatus: KBSourceStatusQueued,
		Mirror: &kb.SourceMirror{
			SourceID: "src_1", Enabled: true,
			SyncStatus: kb.SourceMirrorSyncStatusQueued,
		},
	}

	activeJob := kb.Job{
		ID: "job_sync", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror,
		Status: KBJobStatusQueued, CreatedAt: now, UpdatedAt: now,
	}

	progress := sourceImportProgress(source, []kb.Job{activeJob}, nil)
	require.Equal(t, KBSourceImportProgressStateQueued, progress.State,
		"initial import sync must show queued progress")
}

func seedAdditionalSourceMirror(t *testing.T, svc *KBService, sourceID string, repoName string) {
	t.Helper()
	now := time.Date(2026, 6, 12, 1, 2, 3, 0, time.UTC)
	_, err := svc.store.CreateSource(context.Background(), kb.Source{
		ID: sourceID, WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1",
		SourceType: "codebase", Status: KBSourceStatusReady, Name: repoName,
		DefaultRef: "main", SyncStatus: KBSourceStatusReady, IndexStatus: KBSourceStatusReady,
		Config: "{}", CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	_, err = svc.repositoryStore.Upsert(context.Background(), reposstore.ResourceRepository{
		ID: "rr_mirror_" + sourceID, WorkspaceID: "wsp_1", ResourceType: reposstore.ResourceTypeKBSource,
		ResourceID: sourceID, RepoRole: reposstore.RepoRoleMirror, OwnerKind: "workspace",
		RepoOwner: "wsp_1", RepoName: repoName, RepoFullName: "wsp_1/" + repoName,
		AGSRepoID: "repo_mirror_" + sourceID, DefaultRef: "main", Visibility: "private", UserVisible: true,
		CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	_, err = svc.repositoryStore.Upsert(context.Background(), reposstore.ResourceRepository{
		ID: "rr_insights_" + sourceID, WorkspaceID: "wsp_1", ResourceType: reposstore.ResourceTypeKBSource,
		ResourceID: sourceID, RepoRole: reposstore.RepoRoleInsights, OwnerKind: "workspace",
		RepoOwner: "wsp_1", RepoName: repoName + ".kb", RepoFullName: "wsp_1/" + repoName + ".kb",
		AGSRepoID: "repo_insights_" + sourceID, DefaultRef: "main", Visibility: "private", UserVisible: false,
		CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
	_, err = svc.store.CreateSourceMirror(context.Background(), kb.SourceMirror{
		SourceID: sourceID, WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1",
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit, RemoteURL: "https://93.184.216.34/org/" + repoName + ".git",
		DefaultRef: "main", IntervalSeconds: 60, Enabled: true, SyncStatus: kb.SourceMirrorSyncStatusIdle,
		CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)
}

func TestRunPendingSourceMirrorJobsProcessesBatchAcrossSources(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, nil)
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")

	// Enqueue a sync job for each source.
	_, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)
	_, err = svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_2", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	// Run with batch=2. Both jobs should be claimed and processed.
	result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 2, LeaseOwner: "test-worker"})
	require.NoError(t, err)
	require.Equal(t, 2, result.Claimed, "batch=2 should claim both jobs")
	require.Equal(t, 2, result.Succeeded, "both claimed jobs should succeed")
	require.Zero(t, result.Failed)
}

func TestRunPendingSourceMirrorJobsRefillsFreedWorkerSlot(t *testing.T) {
	setupCtx := context.Background()
	syncer := &blockingSourceMirrorSyncer{
		started: make(chan string, 3),
		releases: map[string]chan struct{}{
			"src_1": make(chan struct{}),
			"src_2": make(chan struct{}),
			"src_3": make(chan struct{}),
		},
		failures: map[string]error{},
	}
	svc := newSourceMirrorServiceForTest(t, syncer, nil)
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")
	seedAdditionalSourceMirror(t, svc, "src_3", "third")
	for _, sourceID := range []string{"src_1", "src_2", "src_3"} {
		_, err := svc.enqueueSourceMirrorSync(setupCtx, "wsp_1", "kb_1", sourceID, "main", "owner", KBSourceMirrorStageQueued, false)
		require.NoError(t, err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	type runResult struct {
		result SourceMirrorRunResult
		err    error
	}
	done := make(chan runResult, 1)
	go func() {
		result, err := svc.RunPendingSourceMirrorJobs(ctx, SourceMirrorRunOptions{Limit: 2, LeaseOwner: "test-worker"})
		done <- runResult{result: result, err: err}
	}()

	firstTwo := map[string]bool{}
	first := <-syncer.started
	second := <-syncer.started
	firstTwo[first] = true
	firstTwo[second] = true
	require.Len(t, firstTwo, 2)
	syncer.failures[first] = errors.New("persistent upstream failure")
	close(syncer.releases[first])
	third := <-syncer.started
	require.False(t, firstTwo[third], "queued job must start when the failed job releases one worker slot")
	close(syncer.releases[second])
	close(syncer.releases[third])

	run := <-done
	require.ErrorContains(t, run.err, "persistent upstream failure")
	require.Equal(t, SourceMirrorRunResult{Claimed: 3, Succeeded: 2, Failed: 1}, run.result)
	thirdJob, err := svc.store.GetJob(setupCtx, "wsp_1", sourceMirrorJobIDForSource(t, svc, third))
	require.NoError(t, err)
	require.Equal(t, KBJobStatusSucceeded, thirdJob.Status)
}

func sourceMirrorJobIDForSource(t *testing.T, svc *KBService, sourceID string) string {
	t.Helper()
	jobs, err := svc.store.ListSourceJobs(context.Background(), "wsp_1", "kb_1", sourceID, KBJobTypeSyncSourceMirror, 1, 1)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	return jobs[0].ID
}

func TestClaimQueuedJobsFairDistributesAcrossSources(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")

	// Enqueue 3 jobs for src_1 and 1 job for src_2. With batch=2, the fair
	// claimer should pick one from each source, not two from src_1.
	now := svc.now()
	for i := 0; i < 3; i++ {
		_, err := svc.store.CreateJob(ctx, kb.Job{
			ID: fmt.Sprintf("job_src1_%d", i), WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
			JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
			Progress:  `{"source_id":"src_1","ref":"main"}`,
			CreatedAt: now.Add(time.Duration(i) * time.Second), UpdatedAt: now.Add(time.Duration(i) * time.Second),
		})
		require.NoError(t, err)
	}
	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_src2_0", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_2",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Progress:  `{"source_id":"src_2","ref":"main"}`,
		CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second),
	})
	require.NoError(t, err)

	jobs, err := svc.store.ClaimQueuedJobsFair(ctx, KBJobTypeSyncSourceMirror, "test-worker", 45*time.Minute, 2, now)
	require.NoError(t, err)
	require.Len(t, jobs, 2, "fair claimer should claim 2 jobs across 2 sources")

	sourceIDs := map[string]bool{}
	for _, j := range jobs {
		sourceIDs[j.SourceID] = true
	}
	require.Len(t, sourceIDs, 2, "fair claimer should pick from distinct sources")
	require.True(t, sourceIDs["src_1"])
	require.True(t, sourceIDs["src_2"])
}

func TestClaimQueuedJobsFairPrioritizesInitialImports(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")
	now := svc.now()

	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_refresh", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Priority: KBSourceMirrorJobPriorityRefresh, Progress: `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: now.Add(-time.Hour), UpdatedAt: now.Add(-time.Hour),
	})
	require.NoError(t, err)
	_, err = svc.store.CreateJob(ctx, kb.Job{
		ID: "job_initial", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_2",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
		Priority: KBSourceMirrorJobPriorityInitialImport, Progress: `{"source_id":"src_2","ref":"main"}`,
		CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)

	jobs, err := svc.store.ClaimQueuedJobsFair(ctx, KBJobTypeSyncSourceMirror, "test-worker", 45*time.Minute, 1, now)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	require.Equal(t, "job_initial", jobs[0].ID, "new imports should take the next worker slot ahead of older refreshes")
}

func TestClaimQueuedJobsFairKeepsFIFOWithinPriority(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")
	now := svc.now()
	for _, job := range []kb.Job{
		{ID: "job_newer", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued, Priority: KBSourceMirrorJobPriorityRefresh, Progress: `{}`, CreatedAt: now, UpdatedAt: now},
		{ID: "job_older", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_2", JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued, Priority: KBSourceMirrorJobPriorityRefresh, Progress: `{}`, CreatedAt: now.Add(-time.Hour), UpdatedAt: now.Add(-time.Hour)},
	} {
		_, err := svc.store.CreateJob(ctx, job)
		require.NoError(t, err)
	}

	jobs, err := svc.store.ClaimQueuedJobsFair(ctx, KBJobTypeSyncSourceMirror, "test-worker", 45*time.Minute, 1, now)
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	require.Equal(t, "job_older", jobs[0].ID)
}

func TestEnqueueSourceMirrorSyncAssignsInitialImportPriority(t *testing.T) {
	ctx := context.Background()
	svc := newSourceMirrorServiceForTest(t, &fakeSourceMirrorSyncer{}, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)
	seedAdditionalSourceMirror(t, svc, "src_2", "second")

	initial, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)
	require.Equal(t, KBSourceMirrorJobPriorityInitialImport, initial.Priority)

	refreshMirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_2")
	require.NoError(t, err)
	succeededAt := svc.now().Add(-time.Hour)
	refreshMirror.LastSuccessAt = &succeededAt
	refreshMirror.LastMirroredCommit = "abc123"
	_, err = svc.store.UpdateSourceMirror(ctx, refreshMirror)
	require.NoError(t, err)

	refresh, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_2", "main", "meshx-kb-scheduler", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)
	require.Equal(t, KBSourceMirrorJobPriorityRefresh, refresh.Priority)
}

func TestReclaimStaleRunningJobsResetsExpiredLeases(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	now := svc.now()
	expiredLease := now.Add(-time.Minute)
	activeLease := now.Add(30 * time.Minute)

	// Job with expired lease — should be reclaimed.
	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_stale", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning,
		LeaseOwner: "old-worker", LeaseUntil: &expiredLease,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: now.Add(-time.Hour), UpdatedAt: now.Add(-time.Hour),
	})
	require.NoError(t, err)

	// Job with active lease — should NOT be reclaimed.
	_, err = svc.store.CreateJob(ctx, kb.Job{
		ID: "job_active", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning,
		LeaseOwner: "current-worker", LeaseUntil: &activeLease,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: now.Add(-time.Minute), UpdatedAt: now.Add(-time.Minute),
	})
	require.NoError(t, err)

	reclaimed, err := svc.ReclaimStaleSourceMirrorJobs(ctx)
	require.NoError(t, err)
	require.Equal(t, int64(1), reclaimed, "only the expired-lease job should be reclaimed")

	stale, err := svc.store.GetJob(ctx, "wsp_1", "job_stale")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusQueued, stale.Status)
	require.Empty(t, stale.LeaseOwner)
	require.Nil(t, stale.LeaseUntil)

	active, err := svc.store.GetJob(ctx, "wsp_1", "job_active")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusRunning, active.Status, "active-lease job must not be reclaimed")
}

func TestCreateUniqueSourceJobReturnsAlreadyActiveError(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Enqueue first job.
	_, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)

	// Second enqueue with failIfExisting=true should return a conflict error.
	_, err = svc.RequestSourceMirrorSync(ctx, "wsp_1", "owner", "kb_1", "src_1", "main")
	require.Error(t, err)
	require.True(t, errors.Is(err, ErrConflict), "manual sync while one is active should return 409 conflict, got: %v", err)

	// Scheduler path (failIfExisting=false) should silently reuse the existing job.
	existing, err := svc.enqueueSourceMirrorSync(ctx, "wsp_1", "kb_1", "src_1", "main", "owner", KBSourceMirrorStageQueued, false)
	require.NoError(t, err)
	require.NotEmpty(t, existing.ID, "scheduler path should return the existing job")
}

func TestUpdateSourceMirrorClearsStaleSuccessOnRemoteChange(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Simulate a previously successful sync.
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	successAt := svc.now().Add(-time.Hour)
	mirror.LastSuccessAt = &successAt
	mirror.LastMirroredCommit = "oldcommit"
	mirror.SyncStatus = kb.SourceMirrorSyncStatusSuccess
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	// Change the remote URL.
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/different-repo.git",
		DefaultRef:   "main",
	})
	require.NoError(t, err)

	updated, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Nil(t, updated.LastSuccessAt, "lastSuccessAt must be cleared when remote changes")
	require.Empty(t, updated.LastMirroredCommit, "lastMirroredCommit must be cleared when remote changes")
	require.NotEqual(t, kb.SourceMirrorSyncStatusSuccess, updated.SyncStatus, "sync status must not show success after remote change")
}

func TestUpdateSourceMirrorPreservesSuccessWhenRemoteUnchanged(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Simulate a previously successful sync.
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	successAt := svc.now().Add(-time.Hour)
	mirror.LastSuccessAt = &successAt
	mirror.LastMirroredCommit = "oldcommit"
	mirror.SyncStatus = kb.SourceMirrorSyncStatusSuccess
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	// Update only the interval, not the remote or ref.
	newInterval := 120
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType:    kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:       "https://93.184.216.34/org/repo.git",
		DefaultRef:      "main",
		IntervalSeconds: &newInterval,
	})
	require.NoError(t, err)

	updated, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.NotNil(t, updated.LastSuccessAt, "lastSuccessAt must be preserved when remote is unchanged")
	require.Equal(t, "oldcommit", updated.LastMirroredCommit, "lastMirroredCommit must be preserved when remote is unchanged")
}

func TestFailSourceMirrorSyncJobDoesNotOverwriteWhenLeaseLost(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{err: fmt.Errorf("sync failed")}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	now := svc.now()
	activeLease := now.Add(30 * time.Minute)
	// Create a job that was claimed by "old-worker" but whose lease expired.
	// A new worker has already reclaimed it as "new-worker".
	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_race", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning,
		LeaseOwner: "new-worker", LeaseUntil: &activeLease,
		AttemptCount: 2,
		Progress:     `{"source_id":"src_1","ref":"main"}`,
		CreatedAt:    now.Add(-time.Hour), UpdatedAt: now.Add(-time.Minute),
	})
	require.NoError(t, err)

	// Simulate the old worker trying to fail the job with an empty lease owner
	// (as if the lease was already cleared). This should NOT overwrite the job.
	job, err := svc.store.GetJob(ctx, "wsp_1", "job_race")
	require.NoError(t, err)
	job.LeaseOwner = "" // old worker's lease was lost
	_ = svc.failSourceMirrorSyncJob(ctx, job, sourceMirrorSyncPayload{SourceID: "src_1", Ref: "main"}, fmt.Errorf("old worker failed"))

	// The job should still be running under the new worker, not failed.
	jobAfter, err := svc.store.GetJob(ctx, "wsp_1", "job_race")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusRunning, jobAfter.Status, "job must not be overwritten when lease is lost")
	require.Equal(t, "new-worker", jobAfter.LeaseOwner, "new worker's lease must be preserved")
}

func TestCancelThenOldWorkerFailDoesNotOverwriteMirrorSource(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{err: fmt.Errorf("sync failed")}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	now := svc.now()
	activeLease := now.Add(30 * time.Minute)
	// Create a running job claimed by "old-worker".
	_, err := svc.store.CreateJob(ctx, kb.Job{
		ID: "job_running", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusRunning,
		LeaseOwner: "old-worker", LeaseUntil: &activeLease,
		AttemptCount: 1,
		Progress:     `{"source_id":"src_1","ref":"main"}`,
		CreatedAt:    now.Add(-time.Hour), UpdatedAt: now.Add(-time.Minute),
	})
	require.NoError(t, err)

	// Disable the mirror — this cancels the running job and clears its lease.
	disabled := false
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/repo.git",
		DefaultRef:   "main",
		Enabled:      &disabled,
	})
	require.NoError(t, err)

	// Verify the job was cancelled.
	job, err := svc.store.GetJob(ctx, "wsp_1", "job_running")
	require.NoError(t, err)
	require.Equal(t, KBJobStatusCancelled, job.Status)

	// Verify mirror/source are NOT failed.
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, kb.SourceMirrorSyncStatusDisabled, mirror.SyncStatus)
	require.Empty(t, mirror.LastError)

	// Now simulate the old worker's goroutine finishing with an error.
	// It still has the old lease owner in its in-memory job copy, but
	// UpdateLeasedJob will fail because the job is now "cancelled" with
	// a different status. The mirror/source must NOT be overwritten to failed.
	oldJob := job
	oldJob.LeaseOwner = "old-worker" // restore old lease owner for the fail call
	_ = svc.failSourceMirrorSyncJob(ctx, oldJob, sourceMirrorSyncPayload{SourceID: "src_1", Ref: "main"}, fmt.Errorf("old worker crashed"))

	// Mirror must still be disabled, not failed.
	mirrorAfter, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.Equal(t, kb.SourceMirrorSyncStatusDisabled, mirrorAfter.SyncStatus,
		"mirror must remain disabled, not failed, after old worker's fail call with lost lease")
	require.Empty(t, mirrorAfter.LastError, "mirror must not carry failure error after cancel")

	// Source must not be failed either.
	sourceAfter, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	require.NotEqual(t, KBSourceStatusFailed, sourceAfter.Status,
		"source must not be marked failed after old worker's fail call with lost lease")
}

func TestImportProgressNotSucceededAfterRemoteChangeWithStaleTerminalJob(t *testing.T) {
	ctx := context.Background()
	syncer := &fakeSourceMirrorSyncer{result: SourceMirrorSyncResult{MirroredCommit: fakeSourceMirrorCommit}}
	svc := newSourceMirrorServiceForTest(t, syncer, &fakeAGSRepositoryStore{})
	seedSourceMirror(t, svc, true)

	// Simulate a successful initial import: mirror has success state,
	// source is ready, and there is a succeeded terminal job.
	now := svc.now()
	successAt := now.Add(-time.Hour)
	mirror, err := svc.store.GetSourceMirror(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	mirror.LastSuccessAt = &successAt
	mirror.LastMirroredCommit = "abc123"
	mirror.SyncStatus = kb.SourceMirrorSyncStatusSuccess
	_, err = svc.store.UpdateSourceMirror(ctx, mirror)
	require.NoError(t, err)

	source, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	source.Status = KBSourceStatusReady
	source.SyncStatus = KBSourceStatusReady
	source.IndexStatus = KBSourceStatusReady
	_, err = svc.store.UpdateSource(ctx, source)
	require.NoError(t, err)

	_, err = svc.store.CreateJob(ctx, kb.Job{
		ID: "job_initial_success", WorkspaceID: "wsp_1", KnowledgeBaseID: "kb_1", SourceID: "src_1",
		JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusSucceeded,
		Progress:  `{"source_id":"src_1","ref":"main"}`,
		CreatedAt: successAt, UpdatedAt: successAt, FinishedAt: &successAt,
	})
	require.NoError(t, err)

	// Verify import progress shows succeeded before the config change.
	sourceBefore, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	sourcesBefore := svc.withSourceMirrors(ctx, []kb.Source{sourceBefore})
	terminalJobsBefore, err := svc.store.ListLatestTerminalSourceJobsForSources(ctx, "wsp_1", "kb_1", []string{"src_1"}, kbImportProgressJobTypes())
	require.NoError(t, err)
	terminalBySourceBefore := make(map[string]kb.Job)
	for _, j := range terminalJobsBefore {
		terminalBySourceBefore[j.SourceID] = j
	}
	progressBefore := sourceImportProgress(sourcesBefore[0], nil, terminalBySourceBefore)
	require.Equal(t, KBSourceImportProgressStateSucceeded, progressBefore.State,
		"before config change, import progress should show succeeded")

	// Change the remote URL — this clears LastSuccessAt/LastMirroredCommit.
	_, err = svc.UpdateSourceMirror(ctx, "wsp_1", "owner", "kb_1", "src_1", KBSourceMirrorMutation{
		UpstreamType: kb.SourceMirrorUpstreamTypeGenericGit,
		RemoteURL:    "https://93.184.216.34/org/different-repo.git",
		DefaultRef:   "main",
	})
	require.NoError(t, err)

	// No new sync job has been enqueued yet. The old terminal succeeded job
	// is still in the DB. Import progress must NOT show succeeded/100%.
	sourceAfter, err := svc.store.GetSource(ctx, "wsp_1", "kb_1", "src_1")
	require.NoError(t, err)
	sourcesAfter := svc.withSourceMirrors(ctx, []kb.Source{sourceAfter})
	terminalJobsAfter, err := svc.store.ListLatestTerminalSourceJobsForSources(ctx, "wsp_1", "kb_1", []string{"src_1"}, kbImportProgressJobTypes())
	require.NoError(t, err)
	terminalBySourceAfter := make(map[string]kb.Job)
	for _, j := range terminalJobsAfter {
		terminalBySourceAfter[j.SourceID] = j
	}
	progressAfter := sourceImportProgress(sourcesAfter[0], nil, terminalBySourceAfter)
	require.NotEqual(t, KBSourceImportProgressStateSucceeded, progressAfter.State,
		"after remote change with no new sync, import progress must not show succeeded/100%%, got state=%s", progressAfter.State)
}