package migration

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	contentservice "github.com/tidbcloud/AgentTeam/internal/service/content"
	contentstore "github.com/tidbcloud/AgentTeam/internal/store/content"
	migrationstore "github.com/tidbcloud/AgentTeam/internal/store/content/migration"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type checkpointFailingMigrationStore struct {
	MigrationStore
	fail atomic.Bool
}

type completionFailingMigrationStore struct {
	MigrationStore
	fail atomic.Bool
}

func (failing *completionFailingMigrationStore) CompleteJob(ctx context.Context, jobID, leaseOwner string, now time.Time) (migrationstore.Job, error) {
	if failing.fail.CompareAndSwap(true, false) {
		return migrationstore.Job{}, errors.New("injected completion checkpoint failure")
	}
	return failing.MigrationStore.CompleteJob(ctx, jobID, leaseOwner, now)
}

func (failing *checkpointFailingMigrationStore) CheckpointPublished(ctx context.Context, params migrationstore.CheckpointPublishedParams) (migrationstore.Job, error) {
	if failing.fail.CompareAndSwap(true, false) {
		return migrationstore.Job{}, errors.New("injected publication checkpoint failure")
	}
	return failing.MigrationStore.CheckpointPublished(ctx, params)
}

func TestWorkerResumesMappedProjectionWithoutRepublish(t *testing.T) {
	ctx := context.Background()
	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
	require.NoError(t, err)
	closeWorkerTestDB(t, db)
	require.NoError(t, db.AutoMigrate(
		&contentstore.VersionedContent{}, &contentstore.ContentVersion{}, &contentstore.VersionOperation{}, &contentstore.VersionPath{},
		&contentstore.VersionPin{}, &contentstore.Outbox{}, &contentstore.StagedPayloadGC{}, &migrationstore.Mapping{}, &migrationstore.Job{},
	))
	tree, err := contentpayload.NewFilesystemTreeBackend(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, tree.Close()) })
	var ids atomic.Int64
	newID := func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, ids.Add(1)), nil }
	core, err := contentservice.NewTree(contentstore.New(db), tree, newID)
	require.NoError(t, err)
	persistence := &completionFailingMigrationStore{MigrationStore: migrationstore.New(db)}
	worker, err := NewWorker(persistence, core, newID, "skill", "worker_1", 3*time.Minute)
	require.NoError(t, err)
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "skill_1", SourceCursor: "head",
		RepositoryID: "42", CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: 10, MaxFileBytes: 1024, MaxTotalBytes: 4096, MaxManifestBytes: 4096},
	}
	var exports, projections atomic.Int64
	exporter := func(context.Context, contentpayload.Limits) (Export, error) {
		exports.Add(1)
		return Export{Files: []File{
			{Path: "README.md", Data: []byte("index")},
			{Path: "skills/reviewer/SKILL.md", Data: []byte("---\nname: Reviewer\ndescription: Review\n---\n")},
		}}, nil
	}
	projectErr := errors.New("injected product projection failure")
	first, err := worker.RunExact(ctx, request, exporter, func(context.Context, Result) error {
		projections.Add(1)
		return projectErr
	})
	require.ErrorIs(t, err, projectErr)
	require.Equal(t, migrationstore.MigrationCursorMapped, first.Cursor)
	require.EqualValues(t, 1, exports.Load())
	require.NoError(t, db.Model(&contentstore.VersionedContent{}).Where("id = ?", first.ContentID).Update("head_version_id", "ctv_newer").Error)
	_, err = worker.RunExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("mapped retry must not re-export legacy source")
		return Export{}, nil
	}, func(context.Context, Result) error {
		t.Fatal("a stale mapped version must not reach product projection")
		return nil
	})
	require.ErrorIs(t, err, contentstore.ErrHeadConflict)
	require.NoError(t, db.Model(&contentstore.VersionedContent{}).Where("id = ?", first.ContentID).Update("head_version_id", first.VersionID).Error)

	persistence.fail.Store(true)
	second, err := worker.RunExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("mapped retry must not re-export legacy source")
		return Export{}, nil
	}, func(_ context.Context, result Result) error {
		projections.Add(1)
		require.Equal(t, first.ContentID, result.ContentID)
		require.Equal(t, first.VersionID, result.VersionID)
		require.Len(t, result.Paths, 2)
		return nil
	})
	require.ErrorContains(t, err, "injected completion checkpoint failure")
	require.Equal(t, migrationstore.MigrationCursorMapped, second.Cursor)
	require.EqualValues(t, 2, projections.Load())

	third, err := worker.RunExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("mapped completion retry must not export legacy source")
		return Export{}, nil
	}, func(context.Context, Result) error {
		projections.Add(1)
		return nil
	})
	require.NoError(t, err)
	require.Equal(t, migrationstore.MigrationCursorCompleted, third.Cursor)
	require.Equal(t, second.VersionID, third.VersionID)
	require.EqualValues(t, 3, projections.Load())

	fourth, err := worker.RunExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("completed replay must not export legacy source")
		return Export{}, nil
	}, func(context.Context, Result) error {
		t.Fatal("completed replay must not repeat product projection")
		return nil
	})
	require.NoError(t, err)
	require.True(t, fourth.Replayed)
	require.Equal(t, third.VersionID, fourth.VersionID)

	var versions, operations, mappings, jobs int64
	require.NoError(t, db.Model(&contentstore.ContentVersion{}).Count(&versions).Error)
	require.NoError(t, db.Model(&contentstore.VersionOperation{}).Count(&operations).Error)
	require.NoError(t, db.Model(&migrationstore.Mapping{}).Count(&mappings).Error)
	require.NoError(t, db.Model(&migrationstore.Job{}).Count(&jobs).Error)
	require.EqualValues(t, 1, versions)
	require.EqualValues(t, 1, operations)
	require.EqualValues(t, 1, mappings)
	require.EqualValues(t, 1, jobs)
}

func TestWorkerUsesReceiptBeforeRestagingAfterLostCheckpoint(t *testing.T) {
	ctx := context.Background()
	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
	require.NoError(t, err)
	closeWorkerTestDB(t, db)
	require.NoError(t, db.AutoMigrate(
		&contentstore.VersionedContent{}, &contentstore.ContentVersion{}, &contentstore.VersionOperation{}, &contentstore.VersionPath{},
		&contentstore.VersionPin{}, &contentstore.Outbox{}, &contentstore.StagedPayloadGC{}, &migrationstore.Mapping{}, &migrationstore.Job{},
	))
	tree, err := contentpayload.NewFilesystemTreeBackend(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, tree.Close()) })
	var ids atomic.Int64
	newID := func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, ids.Add(1)), nil }
	core, err := contentservice.NewTree(contentstore.New(db), tree, newID)
	require.NoError(t, err)
	persistence := &checkpointFailingMigrationStore{MigrationStore: migrationstore.New(db)}
	persistence.fail.Store(true)
	worker, err := NewWorker(persistence, core, newID, "role", "worker_checkpoint", 3*time.Minute)
	require.NoError(t, err)
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "role_1", SourceCursor: "rlv_1",
		RepositoryID: "42", CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: 10, MaxFileBytes: 1024, MaxTotalBytes: 4096, MaxManifestBytes: 4096},
	}
	var exports atomic.Int64
	exporter := func(context.Context, contentpayload.Limits) (Export, error) {
		exports.Add(1)
		return Export{Files: []File{{Path: "IDENTITY.md", Data: []byte("review")}}}, nil
	}
	_, err = worker.RunExact(ctx, request, exporter, func(context.Context, Result) error { return nil })
	require.ErrorContains(t, err, "injected publication checkpoint failure")
	var published contentstore.ContentVersion
	require.NoError(t, db.First(&published).Error)
	require.NoError(t, db.Model(&contentstore.VersionOperation{}).Where("version_id = ?", published.ID).Update("root_digest", strings.Repeat("f", 64)).Error)
	_, err = worker.RunExact(ctx, request, exporter, func(context.Context, Result) error { return nil })
	require.ErrorIs(t, err, migrationstore.ErrMigrationConflict, "a mismatched durable receipt must fail closed")
	require.NoError(t, db.Model(&contentstore.VersionOperation{}).Where("version_id = ?", published.ID).Update("root_digest", published.RootDigest).Error)

	result, err := worker.RunExact(ctx, request, exporter, func(context.Context, Result) error { return nil })
	require.NoError(t, err)
	require.True(t, result.Replayed, "retry must resolve the durable publication receipt")
	require.EqualValues(t, 3, exports.Load(), "a pending cursor may re-export the bounded source before receipt validation")
	var versions, operations int64
	require.NoError(t, db.Model(&contentstore.ContentVersion{}).Count(&versions).Error)
	require.NoError(t, db.Model(&contentstore.VersionOperation{}).Count(&operations).Error)
	require.EqualValues(t, 1, versions)
	require.EqualValues(t, 1, operations)
}

func TestWorkerDetachedMigrationKeepsHeadAndValidatesCompletedResult(t *testing.T) {
	ctx := context.Background()
	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})
	require.NoError(t, err)
	closeWorkerTestDB(t, db)
	require.NoError(t, db.AutoMigrate(
		&contentstore.VersionedContent{}, &contentstore.ContentVersion{}, &contentstore.VersionOperation{}, &contentstore.VersionPath{},
		&contentstore.VersionPin{}, &contentstore.Outbox{}, &contentstore.StagedPayloadGC{}, &migrationstore.Mapping{}, &migrationstore.Job{},
	))
	tree, err := contentpayload.NewFilesystemTreeBackend(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, tree.Close()) })
	var ids atomic.Int64
	newID := func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, ids.Add(1)), nil }
	core, err := contentservice.NewTree(contentstore.New(db), tree, newID)
	require.NoError(t, err)
	quota := contentservice.QuotaProfile{MaxRetainedVersions: 512, MaxStoredPayloadBytes: 1 << 30, MaxPins: 64}
	content, err := core.CreateContent(ctx, contentservice.CreateContentRequest{
		WorkspaceID: "wsp_1", Domain: "memory", BindingKey: "agent_1", Model: contentservice.TreeModel, Actor: "usr_1", Quota: quota,
	})
	require.NoError(t, err)
	parent, err := core.CommitVersion(ctx, contentservice.CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedGeneration: 1, Actor: "usr_1", Model: contentservice.TreeModel,
		Files: []contentpayload.File{{Path: "MEMORY.md", Size: 6, Content: strings.NewReader("parent")}},
	})
	require.NoError(t, err)
	live, err := core.CommitVersion(ctx, contentservice.CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: content.ID, ExpectedHeadVersionID: parent.Version.ID, ExpectedGeneration: 1,
		Actor: "usr_1", Model: contentservice.TreeModel,
		Files: []contentpayload.File{{Path: "MEMORY.md", Size: 4, Content: strings.NewReader("live")}},
	})
	require.NoError(t, err)

	persistence := &checkpointFailingMigrationStore{MigrationStore: migrationstore.New(db)}
	persistence.fail.Store(true)
	authorized, err := contentservice.NewAuthorizedService(core, contentservice.ScopedAccessPolicy{})
	require.NoError(t, err)
	worker, err := NewWorker(persistence, authorized, newID, "memory", "memory_worker", 3*time.Minute)
	require.NoError(t, err)
	request := DetachedRequest{
		Request: Request{
			WorkspaceID: "wsp_1", BindingKey: "agent_1", SourceCursor: "legacy-1", RepositoryID: "42",
			CommitSHA: strings.Repeat("a", 40), ExpectedHeadVersionID: live.Version.ID, Actor: "usr_1",
			Limits: contentpayload.Limits{MaxFiles: 10, MaxFileBytes: 1024, MaxTotalBytes: 4096, MaxManifestBytes: 4096},
		},
		ParentVersionID: parent.Version.ID, ExpectedGeneration: 1, Quota: quota,
	}
	var exports, projections atomic.Int64
	exporter := func(context.Context, contentpayload.Limits) (Export, error) {
		exports.Add(1)
		return Export{Files: []File{{Path: "archive.md", Data: []byte("detached")}}}, nil
	}
	_, err = worker.RunDetachedExact(ctx, request, exporter, func(context.Context, Result) error {
		projections.Add(1)
		return nil
	})
	require.ErrorContains(t, err, "injected publication checkpoint failure")

	result, err := worker.RunDetachedExact(ctx, request, exporter, func(_ context.Context, projected Result) error {
		projections.Add(1)
		require.Equal(t, content.ID, projected.ContentID)
		require.Len(t, projected.Paths, 1)
		return nil
	})
	require.NoError(t, err)
	require.True(t, result.Replayed, "lost job checkpoint resumes from the durable publication receipt")
	require.Equal(t, migrationstore.MigrationCursorCompleted, result.Cursor)
	require.EqualValues(t, 2, exports.Load())
	require.EqualValues(t, 1, projections.Load())

	current, err := core.GetContent(ctx, "wsp_1", content.ID)
	require.NoError(t, err)
	require.Equal(t, live.Version.ID, current.HeadVersionID)
	detached, err := core.GetVersion(ctx, "wsp_1", content.ID, result.VersionID)
	require.NoError(t, err)
	require.Equal(t, parent.Version.ID, detached.ParentVersionID)
	require.Equal(t, uint64(1), detached.Generation)

	replayed, err := worker.RunDetachedExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("completed detached replay must not export")
		return Export{}, nil
	}, func(context.Context, Result) error {
		t.Fatal("completed detached replay must not project")
		return nil
	})
	require.NoError(t, err)
	require.True(t, replayed.Replayed)
	_, err = worker.RunExact(ctx, request.Request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("completed detached job must not export through the ordinary entrypoint")
		return Export{}, nil
	}, func(context.Context, Result) error {
		t.Fatal("completed detached job must not project through the ordinary entrypoint")
		return nil
	})
	require.ErrorIs(t, err, contentstore.ErrHeadConflict, "ordinary replay must prove the completed version is the live head")

	var versions, outbox, operations, retainedGC, mappings int64
	require.NoError(t, db.Model(&contentstore.ContentVersion{}).Where("content_id = ?", content.ID).Count(&versions).Error)
	require.NoError(t, db.Model(&contentstore.Outbox{}).Where("content_id = ?", content.ID).Count(&outbox).Error)
	require.NoError(t, db.Model(&contentstore.VersionOperation{}).Where("content_id = ?", content.ID).Count(&operations).Error)
	require.NoError(t, db.Model(&contentstore.StagedPayloadGC{}).Where("content_id = ? AND status = ?", content.ID, contentstore.GCStatusRetained).Count(&retainedGC).Error)
	require.NoError(t, db.Model(&migrationstore.Mapping{}).Count(&mappings).Error)
	require.EqualValues(t, 3, versions)
	require.EqualValues(t, 3, outbox)
	require.EqualValues(t, 1, operations)
	require.EqualValues(t, 3, retainedGC)
	require.EqualValues(t, 1, mappings)

	require.NoError(t, db.Model(&contentstore.VersionedContent{}).Where("id = ?", content.ID).Update("head_version_id", result.VersionID).Error)
	_, err = worker.RunDetachedExact(ctx, request, exporter, func(context.Context, Result) error { return nil })
	require.ErrorIs(t, err, contentstore.ErrHeadConflict, "completed replay must revalidate the live-head fence")
	require.NoError(t, db.Model(&contentstore.VersionedContent{}).Where("id = ?", content.ID).Update("head_version_id", live.Version.ID).Error)
	require.NoError(t, db.Model(&migrationstore.Mapping{}).Where("ags_repository_id = ?", request.RepositoryID).Update("root_digest", strings.Repeat("f", 64)).Error)
	_, err = worker.RunDetachedExact(ctx, request, exporter, func(context.Context, Result) error { return nil })
	require.ErrorIs(t, err, migrationstore.ErrMigrationConflict, "completed replay must revalidate the exact mapping")

	ordinary := request.Request
	ordinary.SourceCursor = "ordinary-1"
	ordinary.CommitSHA = strings.Repeat("b", 40)
	ordinaryResult, err := worker.RunExact(ctx, ordinary, func(context.Context, contentpayload.Limits) (Export, error) {
		return Export{Files: []File{{Path: "ordinary.md", Data: []byte("ordinary")}}}, nil
	}, func(context.Context, Result) error { return nil })
	require.NoError(t, err)
	require.Equal(t, migrationstore.MigrationCursorCompleted, ordinaryResult.Cursor)
	_, err = worker.RunDetachedExact(ctx, DetachedRequest{
		Request: ordinary, ParentVersionID: live.Version.ID, ExpectedGeneration: 1, Quota: quota,
	}, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("completed ordinary job must not export through the detached entrypoint")
		return Export{}, nil
	}, func(context.Context, Result) error {
		t.Fatal("completed ordinary job must not project through the detached entrypoint")
		return nil
	})
	require.ErrorIs(t, err, contentstore.ErrHeadConflict, "detached replay must preserve its live-head fence")
}

func TestWorkerRejectsBoundsBeforeStoreOrContentIO(t *testing.T) {
	var ids atomic.Int64
	worker := &Worker{
		domain:    "skill",
		newID:     func(prefix string) (string, error) { ids.Add(1); return prefix + "_1", nil },
		semaphore: make(chan struct{}, 1),
	}
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "skill_1", SourceCursor: "head",
		RepositoryID: "42", CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: MigrationMaxFiles + 1, MaxFileBytes: 1, MaxTotalBytes: 1, MaxManifestBytes: 1},
	}
	_, err := worker.RunExact(context.Background(), request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("invalid bounds must not export")
		return Export{}, nil
	}, func(context.Context, Result) error { return nil })
	require.Error(t, err)
	require.Zero(t, ids.Load())
}

func TestWorkerRejectsPartialDetachedQuotaBeforeStoreOrExport(t *testing.T) {
	worker := &Worker{}
	request := DetachedRequest{
		Request: Request{
			WorkspaceID: "wsp_1", BindingKey: "agent_1", SourceCursor: "legacy-1", RepositoryID: "42",
			CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
			Limits: contentpayload.Limits{MaxFiles: 1, MaxFileBytes: 1, MaxTotalBytes: 1, MaxManifestBytes: 1024},
		},
		ExpectedGeneration: 1,
		Quota:              contentservice.QuotaProfile{MaxRetainedVersions: 512, MaxStoredPayloadBytes: 1 << 30},
	}
	_, err := worker.RunDetachedExact(context.Background(), request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("invalid detached quota must fail before export")
		return Export{}, nil
	}, func(context.Context, Result) error { return nil })
	require.ErrorContains(t, err, "generation and quota are invalid")
}

type completedMigrationStore struct {
	MigrationStore
	domains []string
}

type completedContentStore struct{ ContentStore }

func (completedContentStore) FindContentByBinding(context.Context, string, string, string) (contentservice.Content, bool, error) {
	return contentservice.Content{ID: "ctn_1", HeadVersionID: "ctv_1"}, true, nil
}

type pendingMigrationStore struct{ MigrationStore }

func (store *pendingMigrationStore) AcquireExactJob(_ context.Context, params migrationstore.AcquireParams) (migrationstore.Job, error) {
	return migrationstore.Job{ID: params.ID, WorkspaceID: params.WorkspaceID, Domain: params.Domain, BindingKey: params.BindingKey, Cursor: migrationstore.MigrationCursorPending}, nil
}

func (store *pendingMigrationStore) ReleaseFailure(context.Context, string, string, error, time.Time) error {
	return nil
}

func (store *completedMigrationStore) AcquireExactJob(_ context.Context, params migrationstore.AcquireParams) (migrationstore.Job, error) {
	store.domains = append(store.domains, params.Domain)
	return migrationstore.Job{
		ID: params.ID, WorkspaceID: params.WorkspaceID, Domain: params.Domain, BindingKey: params.BindingKey,
		ContentID: "ctn_1", ContentVersionID: "ctv_1", Cursor: migrationstore.MigrationCursorCompleted,
	}, nil
}

func TestWorkerDomainIsFixedByOwner(t *testing.T) {
	store := &completedMigrationStore{}
	sequence := atomic.Int64{}
	newID := func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, sequence.Add(1)), nil }
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "same_binding", SourceCursor: "same_cursor",
		RepositoryID: "42", CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: 1, MaxFileBytes: 1, MaxTotalBytes: 1, MaxManifestBytes: 1024},
	}
	for _, domain := range []string{"role", "skill"} {
		worker := &Worker{
			store: store, content: completedContentStore{}, newID: newID, domain: domain, leaseOwner: domain + "_worker", leaseDuration: 3 * time.Minute,
			now: func() time.Time { return time.Date(2026, 8, 21, 9, 0, 0, 0, time.UTC) }, semaphore: make(chan struct{}, 1),
		}
		_, err := worker.RunExact(context.Background(), request, func(context.Context, contentpayload.Limits) (Export, error) {
			t.Fatal("completed exact target must not export")
			return Export{}, nil
		}, func(context.Context, Result) error {
			t.Fatal("completed exact target must not project")
			return nil
		})
		require.NoError(t, err)
	}
	require.Equal(t, []string{"role", "skill"}, store.domains)
}

func TestWorkerRejectsManifestBeforeContentStoreIO(t *testing.T) {
	sequence := atomic.Int64{}
	worker := &Worker{
		store: &pendingMigrationStore{}, domain: "skill", leaseOwner: "skill_worker", leaseDuration: 3 * time.Minute,
		newID: func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, sequence.Add(1)), nil },
		now:   func() time.Time { return time.Date(2026, 8, 21, 10, 0, 0, 0, time.UTC) }, semaphore: make(chan struct{}, 1),
	}
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "skill_1", SourceCursor: "head", RepositoryID: "42",
		CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: 1, MaxFileBytes: 1024, MaxTotalBytes: 1024, MaxManifestBytes: 100},
	}
	_, err := worker.RunExact(context.Background(), request, func(context.Context, contentpayload.Limits) (Export, error) {
		return Export{Files: []File{{Path: strings.Repeat("x", 80), Data: []byte("x")}}}, nil
	}, func(context.Context, Result) error { return nil })
	require.ErrorContains(t, err, "manifest exceeds bounds")
}

func TestWorkerRejectsExpiredDeadlineBeforeExport(t *testing.T) {
	sequence := atomic.Int64{}
	worker := &Worker{
		store: &pendingMigrationStore{}, domain: "role", leaseOwner: "role_worker", leaseDuration: 3 * time.Minute,
		newID: func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, sequence.Add(1)), nil },
		now:   func() time.Time { return time.Date(2026, 8, 21, 10, 0, 0, 0, time.UTC) }, semaphore: make(chan struct{}, 1),
	}
	request := Request{
		WorkspaceID: "wsp_1", BindingKey: "role_1", SourceCursor: "rlv_1", RepositoryID: "42",
		CommitSHA: strings.Repeat("a", 40), Actor: "usr_1",
		Limits: contentpayload.Limits{MaxFiles: 1, MaxFileBytes: 1024, MaxTotalBytes: 1024, MaxManifestBytes: 1024},
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	_, err := worker.RunExact(ctx, request, func(context.Context, contentpayload.Limits) (Export, error) {
		t.Fatal("expired deadline must fail before source export")
		return Export{}, nil
	}, func(context.Context, Result) error { return nil })
	require.ErrorIs(t, err, context.Canceled)
}

func closeWorkerTestDB(t *testing.T, db *gorm.DB) {
	t.Helper()
	sqlDB, err := db.DB()
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
}