package agenttask

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

func TestNoUnselectedStart(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("ready", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("unselected ready project invoked provider %d times", harness.invoker.callCount())
	}
	project := harness.store.snapshot().Projects["project"]
	if project.Status != ProjectStatusObserved || project.Intent != nil {
		t.Fatalf("unselected project = %#v", project)
	}
}

func TestManualStartFullProgression(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetDisjoint))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	state := harness.store.snapshot()
	project := state.Projects["project"]
	work := project.Works["work"]
	if project.Status != ProjectStatusCompleted || work.State != WorkStateCompleted {
		t.Fatalf("project/work = %s/%s, want completed/completed", project.Status, work.State)
	}
	if harness.invoker.callCount() != 1 || harness.reviewer.callCount() != 1 ||
		harness.integrator.callCount() != 1 {
		t.Fatalf(
			"calls invoke=%d review=%d integrate=%d",
			harness.invoker.callCount(), harness.reviewer.callCount(), harness.integrator.callCount(),
		)
	}
}

func TestInterruptedResumeDefaultsOn(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if harness.store.snapshot().Projects["project"].Status != ProjectStatusCompleted {
		t.Fatalf("default interrupted resume did not complete")
	}
	if harness.invoker.callCount() != 1 {
		t.Fatalf("default resume invocations = %d, want 1", harness.invoker.callCount())
	}
}

func TestInterruptedResumeOverrideFalseStops(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	disabled := false
	harness.start("project", "workspace", &disabled)
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if harness.store.snapshot().Projects["project"].Status != ProjectStatusStopped {
		t.Fatalf("override-off interrupted project was not stopped")
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("override-off resume invoked provider %d times", harness.invoker.callCount())
	}
}

func TestProjectIsolationKeepsIndependentProjectRunning(t *testing.T) {
	snapshots := map[ProjectID]ProjectWorkflowSnapshot{
		"broken":  testSnapshot("broken", "workspace-broken", testUnit("broken-work", WriteSetUnknown)),
		"healthy": testSnapshot("healthy", "workspace-healthy", testUnit("healthy-work", WriteSetUnknown)),
	}
	harness := newHarness(t, snapshots, 2)
	harness.workflow.errors["broken"] = errors.New("fixture workflow parse failure")
	harness.start("broken", "workspace-broken", nil)
	harness.start("healthy", "workspace-healthy", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	state := harness.store.snapshot()
	if state.Projects["broken"].Status != ProjectStatusBlocked {
		t.Fatalf("broken project status = %s", state.Projects["broken"].Status)
	}
	if state.Projects["healthy"].Status != ProjectStatusCompleted {
		t.Fatalf("healthy project status = %s", state.Projects["healthy"].Status)
	}
	if harness.invoker.callCount() != 1 {
		t.Fatalf("provider calls = %d, want healthy project only", harness.invoker.callCount())
	}
}

func TestManualStartDuplicateManagerLeasePreventsConcurrentInvocation(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		project.Lease = &LeaseRecord{
			OwnerID: "other-manager", Token: "live",
			ExpiresAt: fixedClock{now: harness.manager.clock.Now()}.Now().Add(harness.manager.config.LeaseDuration),
		}
		state.Projects["project"] = project
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("live duplicate manager lease invoked provider")
	}
}

func TestManualStartStopCancelsInvocationAndReleasesCapacity(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.invoker.delays["work"] = time.Second
	harness.invoker.locators["work"] = []LocatorRecord{{
		Kind: LocatorProcess, Opaque: "pid:77:start:100", Revision: "process-r1",
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
	}}
	harness.start("project", "workspace", nil)
	done := make(chan error, 1)
	go func() {
		done <- harness.manager.Reconcile(context.Background())
	}()
	deadline := time.Now().Add(time.Second)
	for harness.invoker.activeCount() == 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if harness.invoker.activeCount() == 0 {
		t.Fatal("provider invocation did not start")
	}
	if err := harness.manager.StopProject(context.Background(), "project"); err != nil {
		t.Fatalf("StopProject: %v", err)
	}
	select {
	case err := <-done:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("Reconcile error = %v, want project cancellation", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Reconcile did not stop cancelled invocation")
	}
	state := harness.store.snapshot()
	if state.Projects["project"].Status != ProjectStatusStopped {
		t.Fatalf("project status = %s, want stopped", state.Projects["project"].Status)
	}
	if harness.manager.scheduler.Active("provider\x00profile") != 0 {
		t.Fatal("provider scheduler capacity leaked after stop")
	}
	work := state.Projects["project"].Works["work"]
	if _, ok := work.Locators[LocatorProcess]; !ok {
		t.Fatal("cancel lost the durable process locator")
	}
	if _, ok := state.WorkspaceLeases["workspace"]; ok {
		t.Fatal("cancel retained the workspace invocation lease")
	}
}

func TestStopProjectKeepsIndependentInvocationAlive(t *testing.T) {
	snapshots := map[ProjectID]ProjectWorkflowSnapshot{
		"project-a": testSnapshot(
			"project-a",
			"workspace-a",
			testUnit("work-a", WriteSetUnknown),
		),
		"project-b": testSnapshot(
			"project-b",
			"workspace-b",
			testUnit("work-b", WriteSetUnknown),
		),
	}
	harness := newHarness(t, snapshots, 2)
	harness.start("project-a", "workspace-a", nil)
	harness.start("project-b", "workspace-b", nil)
	release := make(chan struct{})
	harness.invoker.blockAllInvocations(release)
	done := make(chan error, 1)
	go func() {
		done <- harness.manager.Reconcile(context.Background())
	}()
	waitFor(t, "two independent invocations", func() bool {
		return harness.invoker.activeCount() == 2
	})
	harness.manager.activeMu.Lock()
	delete(harness.manager.activeRuns, "project-a")
	harness.manager.activeMu.Unlock()
	if err := harness.manager.StopProject(
		context.Background(),
		"project-a",
	); err != nil {
		t.Fatalf("StopProject(project-a): %v", err)
	}
	waitFor(t, "only the selected invocation cancellation", func() bool {
		return harness.invoker.activeCount() == 1
	})
	time.Sleep(20 * time.Millisecond)
	if harness.invoker.activeCount() != 1 ||
		harness.invoker.cancelCount() != 1 {
		t.Fatalf(
			"independent invocation state = active %d, cancelled %d",
			harness.invoker.activeCount(),
			harness.invoker.cancelCount(),
		)
	}
	if harness.store.snapshot().Projects["project-a"].Lease == nil {
		t.Fatal("cross-runtime stop removed the active project lease")
	}
	close(release)
	if err := <-done; !errors.Is(err, context.Canceled) {
		t.Fatalf("Reconcile error = %v, want selected project cancellation", err)
	}
	state := harness.store.snapshot()
	if state.Projects["project-a"].Status != ProjectStatusStopped ||
		state.Projects["project-b"].Status != ProjectStatusCompleted {
		t.Fatalf(
			"project statuses = %s/%s, want stopped/completed",
			state.Projects["project-a"].Status,
			state.Projects["project-b"].Status,
		)
	}
}

func TestStopProjectPersistsUntilExplicitRestart(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)

	if err := harness.manager.StopProject(context.Background(), "project"); err != nil {
		t.Fatalf("StopProject: %v", err)
	}

	for i := 0; i < 5; i++ {
		if err := harness.manager.Reconcile(context.Background()); err != nil {
			t.Fatalf("Reconcile %d: %v", i, err)
		}
	}

	if harness.invoker.callCount() != 0 {
		t.Fatalf("stopped project was executed %d times, want 0", harness.invoker.callCount())
	}

	state := harness.store.snapshot()
	if state.Projects["project"].Status != ProjectStatusStopped {
		t.Fatalf("project status = %s, want stopped", state.Projects["project"].Status)
	}

	req := StartRequest{
		CommandID:        "start-2-project",
		ProjectID:        "project",
		WorkspaceID:      "workspace",
		MilestoneID:      "milestone",
		WorkflowRevision: "workflow-r1",
		ConfigRevision:   "config-r1",
		GrantRevision:    "grant-r1",
	}
	if err := harness.manager.StartProject(context.Background(), req); err != nil {
		t.Fatalf("StartProject restart: %v", err)
	}
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile after restart: %v", err)
	}

	state = harness.store.snapshot()
	if state.Projects["project"].Status != ProjectStatusCompleted {
		t.Fatalf("restarted project status = %s, want completed", state.Projects["project"].Status)
	}
	if harness.invoker.callCount() != 1 {
		t.Fatalf("restarted project provider calls = %d, want 1", harness.invoker.callCount())
	}
}

func TestStoppedWorkResumesFromDurableStage(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)

	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		work := project.Works["work"]
		work.Unit = testUnit("work", WriteSetUnknown)
		work.State = WorkStateReviewing
		work.ResumeStage = WorkStateReviewing
		work.Attempt = 1
		work.AttemptID = attemptID("work", 1)
		work.Submission = &Submission{
			ProjectID:  "project",
			WorkUnitID: "work",
			AttemptID:  attemptID("work", 1),
			ArtifactID: "art-1",
			Ready:      true,
		}
		project.Works["work"] = work
		state.Projects["project"] = project
	})

	if err := harness.manager.StopProject(context.Background(), "project"); err != nil {
		t.Fatalf("StopProject: %v", err)
	}

	state := harness.store.snapshot()
	work := state.Projects["project"].Works["work"]
	if work.State != WorkStateStopped || work.ResumeStage != WorkStateReviewing {
		t.Fatalf("stopped work state/resumeStage = %s/%s, want stopped/reviewing", work.State, work.ResumeStage)
	}

	req := StartRequest{
		CommandID:        "start-2-project",
		ProjectID:        "project",
		WorkspaceID:      "workspace",
		MilestoneID:      "milestone",
		WorkflowRevision: "workflow-r1",
		ConfigRevision:   "config-r1",
		GrantRevision:    "grant-r1",
	}
	if err := harness.manager.StartProject(context.Background(), req); err != nil {
		t.Fatalf("StartProject: %v", err)
	}

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}

	state = harness.store.snapshot()
	work = state.Projects["project"].Works["work"]
	if work.ResumeStage != "" {
		t.Fatalf("recovered work still has resumeStage = %s", work.ResumeStage)
	}
	if work.State != WorkStateCompleted {
		t.Fatalf("recovered work state = %s, want completed", work.State)
	}
}

func TestRecoverStoppedDispatchStartsFreshAttempt(t *testing.T) {
	work := WorkRecord{
		Unit:        testUnit("work", WriteSetUnknown),
		State:       WorkStateStopped,
		ResumeStage: WorkStateDispatching,
		Attempt:     1,
		AttemptID:   attemptID("work", 1),
		Target:      &ExecutionTarget{ProviderID: "provider"},
		Isolation:   &IsolationIdentity{ID: "overlay"},
		Locators: map[LocatorKind]LocatorRecord{
			LocatorOverlay: {Kind: LocatorOverlay, Opaque: "overlay"},
			LocatorProcess: {Kind: LocatorProcess, Opaque: "process"},
		},
	}
	recoverStoppedWork(&work)
	if work.State != WorkStateReady ||
		work.ResumeStage != "" ||
		work.Attempt != 2 ||
		work.AttemptID != attemptID("work", 2) {
		t.Fatalf("recovered dispatch identity = %#v", work)
	}
	if work.Target != nil ||
		work.Isolation != nil ||
		len(work.Locators) != 0 {
		t.Fatalf("recovered dispatch retained invocation identity = %#v", work)
	}
}

func TestAutoResumeOverrideFalseRequiresExplicitRestart(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	disabled := false
	harness.start("project", "workspace", &disabled)
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		state.Projects["project"] = project
	})

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}

	state := harness.store.snapshot()
	if state.Projects["project"].Status != ProjectStatusStopped {
		t.Fatalf("project status = %s, want stopped", state.Projects["project"].Status)
	}

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile 2: %v", err)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("provider calls = %d, want 0", harness.invoker.callCount())
	}

	req := StartRequest{
		CommandID:        "start-2-project",
		ProjectID:        "project",
		WorkspaceID:      "workspace",
		MilestoneID:      "milestone",
		WorkflowRevision: "workflow-r1",
		ConfigRevision:   "config-r1",
		GrantRevision:    "grant-r1",
	}
	if err := harness.manager.StartProject(context.Background(), req); err != nil {
		t.Fatalf("StartProject: %v", err)
	}
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile 3: %v", err)
	}
	if harness.store.snapshot().Projects["project"].Status != ProjectStatusCompleted {
		t.Fatalf("explicitly restarted project did not complete")
	}
}

func TestDeviceSingletonLeaseBlocksDuplicateOwnerUntilExpiry(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	harness.store.edit(func(state *ManagerState) {
		state.DeviceLease = &LeaseRecord{
			OwnerID: "other-daemon", Token: "device-token",
			ExpiresAt: harness.manager.clock.Now().Add(time.Minute),
		}
	})

	err := harness.manager.Reconcile(context.Background())
	if !errors.Is(err, ErrDeviceLeaseHeld) {
		t.Fatalf("Reconcile error = %v, want ErrDeviceLeaseHeld", err)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("duplicate device owner invoked provider %d times", harness.invoker.callCount())
	}

	harness.store.edit(func(state *ManagerState) {
		state.DeviceLease.ExpiresAt = harness.manager.clock.Now().Add(-time.Second)
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile after expiry: %v", err)
	}
	if harness.invoker.callCount() != 1 {
		t.Fatalf("expired device lease provider calls = %d, want 1", harness.invoker.callCount())
	}
}

func TestWorkspaceLeaseBlocksDuplicateInvocationUntilExpiry(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	harness.store.edit(func(state *ManagerState) {
		state.WorkspaceLeases["workspace"] = LeaseRecord{
			OwnerID: "other-manager", Token: "workspace-token",
			ExpiresAt: harness.manager.clock.Now().Add(time.Minute),
		}
	})

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	project := harness.store.snapshot().Projects["project"]
	if project.Status != ProjectStatusBlocked || project.Blocker == nil ||
		project.Blocker.Code != BlockerDuplicateWorkspaceCall {
		t.Fatalf("project = %#v, want duplicate workspace blocker", project)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("duplicate workspace owner invoked provider %d times", harness.invoker.callCount())
	}

	harness.store.edit(func(state *ManagerState) {
		lease := state.WorkspaceLeases["workspace"]
		lease.ExpiresAt = harness.manager.clock.Now().Add(-time.Second)
		state.WorkspaceLeases["workspace"] = lease
	})
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile after expiry: %v", err)
	}
	if harness.store.snapshot().Projects["project"].Status != ProjectStatusCompleted {
		t.Fatal("project did not recover after the foreign workspace lease expired")
	}
}

func TestWorkspaceLeaseConflictDoesNotFenceIndependentProject(t *testing.T) {
	snapshotA := testSnapshot("a-blocked", "workspace-a", testUnit("work-a", WriteSetUnknown))
	snapshotB := testSnapshot("b-independent", "workspace-b", testUnit("work-b", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{
		"a-blocked":     snapshotA,
		"b-independent": snapshotB,
	}, 1)
	harness.start("a-blocked", "workspace-a", nil)
	harness.start("b-independent", "workspace-b", nil)
	harness.store.edit(func(state *ManagerState) {
		state.WorkspaceLeases["workspace-a"] = LeaseRecord{
			OwnerID: "other-manager", Token: "foreign-token-a",
			ExpiresAt: harness.manager.clock.Now().Add(time.Minute),
		}
	})

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}

	state := harness.store.snapshot()
	projectA := state.Projects["a-blocked"]
	if projectA.Status != ProjectStatusBlocked || projectA.Blocker == nil ||
		projectA.Blocker.Code != BlockerDuplicateWorkspaceCall {
		t.Fatalf("projectA = %#v, want duplicate workspace blocker", projectA)
	}
	projectB := state.Projects["b-independent"]
	if projectB.Status != ProjectStatusCompleted {
		t.Fatalf("projectB = %#v, want completed status", projectB)
	}

	if harness.invoker.callCount() != 1 {
		t.Fatalf("invoker calls = %d, want 1 for independent project", harness.invoker.callCount())
	}

	foreignLease, ok := state.WorkspaceLeases["workspace-a"]
	if !ok || foreignLease.Token != "foreign-token-a" {
		t.Fatalf("foreign workspace lease was modified or deleted: %#v", foreignLease)
	}

	if state.DeviceLease != nil || state.Projects["a-blocked"].Lease != nil || state.Projects["b-independent"].Lease != nil {
		t.Fatalf("owned device/project claims remained: %#v", state)
	}
	if _, ok := state.WorkspaceLeases["workspace-b"]; ok {
		t.Fatalf("owned workspace claim for b remained: %#v", state)
	}
}

func TestRestartRetainsLiveChildWithoutDuplicateInvocation(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	seedDispatchingCheckpoint(harness, LocatorRecord{
		Kind: LocatorProcess, Opaque: "pid:42:start:9001", Revision: "process-r1",
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
	})
	harness.recovery.observations["work"] = RecoveryObservation{
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
		Execution: RecoveryExecutionLive,
	}

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile live child: %v", err)
	}
	work := harness.store.snapshot().Projects["project"].Works["work"]
	if work.State != WorkStateDispatching {
		t.Fatalf("live child work state = %s, want dispatching", work.State)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("live child was duplicated %d times", harness.invoker.callCount())
	}

	submission := Submission{
		ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
		ArtifactID: "artifact-recovered", Ready: true,
		Locators: []LocatorRecord{{
			Kind: LocatorProcess, Opaque: "pid:42:start:9001", Revision: "process-r1",
			ProjectID: "project", WorkspaceID: "workspace",
			WorkUnitID: "work", AttemptID: attemptID("work", 1),
		}},
	}
	harness.recovery.observations["work"] = RecoveryObservation{
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
		Execution: RecoveryExecutionSubmitted, Submission: &submission,
	}
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile submitted child: %v", err)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("recovered submission reinvoked provider %d times", harness.invoker.callCount())
	}
	if harness.store.snapshot().Projects["project"].Works["work"].State != WorkStateCompleted {
		t.Fatal("recovered submission did not finish review and integration")
	}
}

func TestRestartRecoveredSubmissionRequiresWorkflowEvidence(t *testing.T) {
	sessionLocator := LocatorRecord{
		Kind: LocatorSession, Opaque: "native-session", Revision: "session-r1",
		ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1),
	}
	processLocator := LocatorRecord{
		Kind: LocatorProcess, Opaque: "pid:42:start:9001", Revision: "process-r1",
		ProjectID: "project", WorkspaceID: "workspace", WorkUnitID: "work", AttemptID: attemptID("work", 1),
	}

	for _, test := range []struct {
		name         string
		submission   Submission
		isPi         bool
		observeFunc  func(WorkflowEvidenceRequest, int) (ArtifactEvidence, error)
		wantState    WorkState
		wantBlocker  BlockerCode
		wantReviews  int
		wantObserves int
		wantRepairs  int
	}{
		{
			name: "complete evidence recovers and reviews",
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: true,
				Locators: []LocatorRecord{processLocator},
			},
			wantState:    WorkStateCompleted,
			wantReviews:  1,
			wantObserves: 1,
			wantRepairs:  0,
		},
		{
			name: "incomplete submission blocks without evidence check or review",
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: false,
				Locators: []LocatorRecord{processLocator},
			},
			wantState:    WorkStateBlocked,
			wantBlocker:  BlockerSubmissionIncomplete,
			wantReviews:  0,
			wantObserves: 0,
			wantRepairs:  0,
		},
		{
			name: "placeholder from non-Pi provider blocks without repair or review",
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: true,
				Locators: []LocatorRecord{processLocator},
			},
			observeFunc: func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) {
				return ArtifactEvidence{
					Active: true, Identity: artifactIdentity(request.Project, request.Work, request.Submission),
					Completeness: ArtifactPlaceholder,
				}, nil
			},
			wantState:    WorkStateBlocked,
			wantBlocker:  BlockerEvidenceRepairDenied,
			wantReviews:  0,
			wantObserves: 1,
			wantRepairs:  0,
		},
		{
			name: "inactive or identity mismatched evidence blocks review",
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: true,
				Locators: []LocatorRecord{processLocator},
			},
			observeFunc: func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) {
				identity := artifactIdentity(request.Project, request.Work, request.Submission)
				identity.AttemptID = "stale-attempt"
				return ArtifactEvidence{Active: true, Identity: identity, Completeness: ArtifactComplete}, nil
			},
			wantState:    WorkStateBlocked,
			wantBlocker:  BlockerArtifactMismatch,
			wantReviews:  0,
			wantObserves: 1,
			wantRepairs:  0,
		},
		{
			name: "Pi repair rematches and reviews successfully",
			isPi: true,
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: true,
				Locators: []LocatorRecord{sessionLocator},
			},
			observeFunc: func(request WorkflowEvidenceRequest, call int) (ArtifactEvidence, error) {
				identity := artifactIdentity(request.Project, request.Work, request.Submission)
				if call == 0 {
					return ArtifactEvidence{
						Active: true, Identity: identity, Completeness: ArtifactPlaceholder,
						RepairIntent: &EvidenceRepairIntent{
							Identity: identity, DispatchOrdinal: request.Work.DispatchOrdinal,
							NativeLocator: request.Work.Locators[LocatorSession],
						},
					}, nil
				}
				return ArtifactEvidence{Active: true, Identity: identity, Completeness: ArtifactComplete}, nil
			},
			wantState:    WorkStateCompleted,
			wantReviews:  1,
			wantObserves: 2,
			wantRepairs:  1,
		},
		{
			name: "Pi repair without fresh match blocks review",
			isPi: true,
			submission: Submission{
				ProjectID: "project", WorkUnitID: "work", AttemptID: attemptID("work", 1),
				ArtifactID: "artifact-recovered", Ready: true,
				Locators: []LocatorRecord{sessionLocator},
			},
			observeFunc: func(request WorkflowEvidenceRequest, _ int) (ArtifactEvidence, error) {
				identity := artifactIdentity(request.Project, request.Work, request.Submission)
				return ArtifactEvidence{
					Active: true, Identity: identity, Completeness: ArtifactPlaceholder,
					RepairIntent: &EvidenceRepairIntent{
						Identity: identity, DispatchOrdinal: request.Work.DispatchOrdinal,
						NativeLocator: request.Work.Locators[LocatorSession],
					},
				}, nil
			},
			wantState:    WorkStateBlocked,
			wantBlocker:  BlockerSubmissionIncomplete,
			wantReviews:  0,
			wantObserves: 2,
			wantRepairs:  1,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
			harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
			harness.start("project", "workspace", nil)

			locator := processLocator
			if test.isPi {
				locator = sessionLocator
			}
			seedDispatchingCheckpoint(harness, locator)

			if test.isPi {
				harness.store.edit(func(state *ManagerState) {
					project := state.Projects["project"]
					work := project.Works["work"]
					work.Target.ProviderID = "pi"
					project.Works["work"] = work
					state.Projects["project"] = project
				})
			}

			if test.observeFunc != nil {
				harness.evidence.observeFunc = test.observeFunc
			}

			harness.recovery.observations["work"] = RecoveryObservation{
				ProjectID: "project", WorkspaceID: "workspace",
				WorkUnitID: "work", AttemptID: attemptID("work", 1),
				Execution: RecoveryExecutionSubmitted, Submission: &test.submission,
			}

			if err := harness.manager.Reconcile(context.Background()); err != nil {
				t.Fatalf("Reconcile: %v", err)
			}

			work := harness.store.snapshot().Projects["project"].Works["work"]
			if work.State != test.wantState {
				t.Fatalf("work state = %s, want %s", work.State, test.wantState)
			}

			if test.wantBlocker != "" {
				if work.Blocker == nil || work.Blocker.Code != test.wantBlocker {
					t.Fatalf("work blocker = %#v, want %q", work.Blocker, test.wantBlocker)
				}
			} else if work.Blocker != nil {
				t.Fatalf("unexpected work blocker = %#v", work.Blocker)
			}

			if harness.reviewer.callCount() != test.wantReviews {
				t.Fatalf("reviewer calls = %d, want %d", harness.reviewer.callCount(), test.wantReviews)
			}

			observes, repairs := harness.evidence.counts()
			if observes != test.wantObserves {
				t.Fatalf("observe calls = %d, want %d", observes, test.wantObserves)
			}
			if repairs != test.wantRepairs {
				t.Fatalf("repair calls = %d, want %d", repairs, test.wantRepairs)
			}
		})
	}
}

func TestCorruptLocatorIdentityStopsBeforeRecoveryOrProvider(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	seedDispatchingCheckpoint(harness, LocatorRecord{
		Kind: LocatorProcess, Opaque: "pid:42:start:9001", Revision: "process-r1",
		ProjectID: "other-project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
	})

	err := harness.manager.Reconcile(context.Background())
	if !errors.Is(err, ErrCheckpointInvalid) {
		t.Fatalf("Reconcile error = %v, want ErrCheckpointInvalid", err)
	}
	if harness.recovery.callCount() != 0 || harness.invoker.callCount() != 0 {
		t.Fatalf(
			"corrupt checkpoint calls recovery/provider = %d/%d, want 0/0",
			harness.recovery.callCount(), harness.invoker.callCount(),
		)
	}
}

func TestPartialCompletionArchiveBecomesTerminalBlocker(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		work := WorkRecord{
			Unit: testUnit("work", WriteSetUnknown), State: WorkStateCompleted,
			Attempt: 1, AttemptID: attemptID("work", 1),
			DispatchOrdinal: 1,
			Locators: map[LocatorKind]LocatorRecord{
				LocatorCompletion: {
					Kind: LocatorCompletion, Opaque: "archive:partial", Revision: "archive-r1",
					ProjectID: "project", WorkspaceID: "workspace",
					WorkUnitID: "work", AttemptID: attemptID("work", 1),
				},
			},
		}
		project.Works["work"] = work
		state.Projects["project"] = project
	})
	harness.recovery.observations["work"] = RecoveryObservation{
		ProjectID: "project", WorkspaceID: "workspace",
		WorkUnitID: "work", AttemptID: attemptID("work", 1),
		Completion: RecoveryCompletionPartial,
	}

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	project := harness.store.snapshot().Projects["project"]
	work := project.Works["work"]
	if project.Status != ProjectStatusBlocked || work.State != WorkStateBlocked ||
		work.Blocker == nil || work.Blocker.Code != BlockerPartialCompletion {
		t.Fatalf("partial completion project/work = %#v/%#v", project, work)
	}
	if harness.invoker.callCount() != 0 {
		t.Fatalf("partial completion invoked provider %d times", harness.invoker.callCount())
	}
}

func TestFailureBudgetPersistsAndStopsReviewRework(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.manager.config.MaxReworkAttempts = 5
	harness.manager.config.MaxFailureAttempts = 2
	harness.reviewer.sequences["work"] = []ReviewVerdict{
		ReviewVerdictWarn, ReviewVerdictWarn, ReviewVerdictPass,
	}
	harness.start("project", "workspace", nil)

	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	work := harness.store.snapshot().Projects["project"].Works["work"]
	budget := work.FailureBudgets[FailureStageReview]
	if work.State != WorkStateBlocked || work.Blocker == nil ||
		work.Blocker.Code != BlockerFailureBudgetExhausted {
		t.Fatalf("work = %#v, want exhausted failure budget blocker", work)
	}
	if budget.Consecutive != 2 || budget.Limit != 2 ||
		budget.LastCode != BlockerReviewFailed {
		t.Fatalf("review budget = %#v", budget)
	}
	if harness.invoker.callCount() != 2 || harness.reviewer.callCount() != 2 ||
		harness.integrator.callCount() != 0 {
		t.Fatalf(
			"calls invoke/review/integrate = %d/%d/%d",
			harness.invoker.callCount(), harness.reviewer.callCount(), harness.integrator.callCount(),
		)
	}
}

func TestSuccessfulRunPersistsOpaqueLocatorKinds(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	attempt := attemptID("work", 1)
	harness.invoker.locators["work"] = []LocatorRecord{
		{
			Kind: LocatorProcess, Opaque: "pid:42:start:9001", Revision: "process-r1",
			ProjectID: "project", WorkspaceID: "workspace",
			WorkUnitID: "work", AttemptID: attempt,
		},
		{
			Kind: LocatorSession, Opaque: "provider-session-7", Revision: "session-r1",
			ProjectID: "project", WorkspaceID: "workspace",
			WorkUnitID: "work", AttemptID: attempt,
		},
	}
	harness.start("project", "workspace", nil)
	if err := harness.manager.Reconcile(context.Background()); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	work := harness.store.snapshot().Projects["project"].Works["work"]
	for _, kind := range []LocatorKind{
		LocatorProcess, LocatorSession, LocatorOverlay, LocatorChangeSet,
	} {
		if _, ok := work.Locators[kind]; !ok {
			t.Errorf("missing persisted %s locator", kind)
		}
	}
}

func seedDispatchingCheckpoint(harness *managerHarness, locator LocatorRecord) {
	harness.t.Helper()
	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		unit := testUnit("work", WriteSetUnknown)
		work := WorkRecord{
			Unit: unit, State: WorkStateDispatching,
			Attempt: 1, AttemptID: attemptID("work", 1),
			DispatchOrdinal: 1,
			Target: &ExecutionTarget{
				ProviderID: "provider", ModelID: "model", ProfileID: "profile",
				ProfileRevision: "profile-r1", ConfigRevision: "config-r1", Capacity: 1,
			},
			Isolation: &IsolationIdentity{
				ID: "isolation-1", Revision: "isolation-r1",
				Mode: unit.IsolationMode, PinnedBaseRevision: "base-r1",
				TaskRoot: "/tmp/task-work",
			},
			Locators: map[LocatorKind]LocatorRecord{
				LocatorOverlay: {
					Kind: LocatorOverlay, Opaque: "isolation-1", Revision: "isolation-r1",
					ProjectID: "project", WorkspaceID: "workspace",
					WorkUnitID: "work", AttemptID: attemptID("work", 1),
				},
				locator.Kind: locator,
			},
		}
		project.Works["work"] = work
		state.Projects["project"] = project
	})
}

func TestClaimProjectCASConflictReturnsCommittedDecision(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)

	store := harness.store
	wrappedStore := &casConflictStore{
		store: store,
		onConflict: func() {
			store.edit(func(state *ManagerState) {
				project := state.Projects["project"]
				project.Status = ProjectStatusStopped
				state.Projects["project"] = project
			})
		},
	}
	harness.manager.store = wrappedStore

	claim, err := harness.manager.claimProject(context.Background(), "project")
	if err != nil {
		t.Fatalf("claimProject: %v", err)
	}
	if claim != nil {
		t.Fatalf("claimProject returned a claim for stopped project after CAS conflict")
	}
}

func TestClaimIntegrationCASConflictReturnsCommittedDecision(t *testing.T) {
	harness := newHarness(t, nil, 1)
	store := harness.store
	wrappedStore := &casConflictStore{
		store: store,
		onConflict: func() {
			store.edit(func(state *ManagerState) {
				state.IntegrationLeases["workspace"] = LeaseRecord{
					OwnerID:   "other-owner",
					Token:     "foreign",
					ExpiresAt: time.Now().Add(time.Hour),
				}
			})
		},
	}
	harness.manager.store = wrappedStore

	claim, err := harness.manager.claimIntegration(context.Background(), "workspace")
	if err != nil {
		t.Fatalf("claimIntegration: %v", err)
	}
	if claim != nil {
		t.Fatalf("claimIntegration returned a claim when foreign lease committed during CAS conflict")
	}
}

func TestWorkflowActivationCASConflictUsesCommittedState(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)

	store := harness.store
	wrappedStore := &casConflictStore{
		store: store,
		onConflict: func() {
			store.edit(func(state *ManagerState) {
				project := state.Projects["project"]
				project.Status = ProjectStatusStopped
				state.Projects["project"] = project
			})
		},
	}
	harness.manager.store = wrappedStore

	active, err := harness.manager.observeWorkflows(context.Background())
	if err != nil {
		t.Fatalf("observeWorkflows: %v", err)
	}
	if len(active) != 0 {
		t.Fatalf("observeWorkflows returned active projects = %v, want empty after CAS conflict to stopped", active)
	}
}

func TestWorkflowActivationCASConflictUsesCommittedEventDecision(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)

	harness.store.edit(func(state *ManagerState) {
		project := state.Projects["project"]
		project.Status = ProjectStatusRunning
		state.Projects["project"] = project
	})

	store := harness.store
	wrappedStore := &casConflictStore{
		store: store,
		onConflict: func() {
			store.edit(func(state *ManagerState) {
				project := state.Projects["project"]
				project.Status = ProjectStatusStarted
				state.Projects["project"] = project
			})
		},
	}
	harness.manager.store = wrappedStore

	active, err := harness.manager.observeWorkflows(context.Background())
	if err != nil {
		t.Fatalf("observeWorkflows: %v", err)
	}
	if len(active) != 1 || active[0] != "project" {
		t.Fatalf("observeWorkflows active = %v, want [project]", active)
	}

	eventsEmitted := harness.events.snapshot()
	var autoResumeCount int
	var observedEvent *Event
	for _, e := range eventsEmitted {
		if e.Type == EventAutoResume {
			autoResumeCount++
		}
		if e.Type == EventObserved {
			eCopy := e
			observedEvent = &eCopy
		}
	}
	if autoResumeCount != 0 {
		t.Fatalf("auto-resume event count = %d, want 0 on explicit-start winner", autoResumeCount)
	}
	if observedEvent == nil || observedEvent.CommandID == "" || observedEvent.WorkflowRevision == "" {
		t.Fatalf("observed event missing committed identity: %#v", observedEvent)
	}
}

type casConflictStore struct {
	store      *memoryStore
	mu         sync.Mutex
	injected   bool
	armed      <-chan struct{}
	onConflict func()
}

func (s *casConflictStore) Load(ctx context.Context) (ManagerState, StateRevision, error) {
	return s.store.Load(ctx)
}

func (s *casConflictStore) CompareAndSwap(ctx context.Context, expected StateRevision, next ManagerState) (StateRevision, error) {
	s.mu.Lock()
	if s.armed != nil {
		select {
		case <-s.armed:
		default:
			s.mu.Unlock()
			return s.store.CompareAndSwap(ctx, expected, next)
		}
	}
	if !s.injected {
		s.injected = true
		if s.onConflict != nil {
			s.onConflict()
		}
		s.mu.Unlock()
		return "", ErrRevisionConflict
	}
	s.mu.Unlock()
	return s.store.CompareAndSwap(ctx, expected, next)
}

// TestDeviceLeaseRenewsDuringLongInvocation verifies that the device lease
// remains held while a provider invocation is blocked, and that the guarded
// context stays alive across multiple renewal intervals.
func TestDeviceLeaseRenewsDuringLongInvocation(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	ticks, clock := harness.manualRenewals()
	harness.start("project", "workspace", nil)

	// Block the invoker so the manager stays inside invocation.Wait.
	releaseInvocation := make(chan struct{})
	harness.invoker.blockAllInvocations(releaseInvocation)

	reconcileDone := make(chan error, 1)
	go func() {
		reconcileDone <- harness.manager.Reconcile(context.Background())
	}()

	// Give the manager time to enter the blocked invocation.
	deadline := time.Now().Add(time.Second)
	for harness.invoker.activeCount() == 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if harness.invoker.activeCount() == 0 {
		t.Fatal("invocation did not enter blocked state")
	}

	// The device lease must still be held with a future expiry.
	state := harness.store.snapshot()
	if state.DeviceLease == nil {
		t.Fatal("device lease was not claimed")
	}
	if state.DeviceLease.OwnerID != harness.manager.config.OwnerID {
		t.Fatalf("device lease owner = %q, want %q", state.DeviceLease.OwnerID, harness.manager.config.OwnerID)
	}
	originalExpiry := state.DeviceLease.ExpiresAt
	if !originalExpiry.After(harness.manager.clock.Now()) {
		t.Fatalf("device lease expiry %v is not after now %v", originalExpiry, harness.manager.clock.Now())
	}

	clock.Advance(time.Minute)
	ticks <- clock.Now()
	waitFor(t, "device lease renewal", func() bool {
		return harness.store.snapshot().DeviceLease.ExpiresAt.After(originalExpiry)
	})

	// Release the invocation and verify it completes.
	close(releaseInvocation)
	err := <-reconcileDone
	if err != nil {
		t.Fatalf("Reconcile returned error after renewal: %v", err)
	}
	if harness.invoker.callCount() != 1 {
		t.Fatalf("invoker calls = %d, want 1", harness.invoker.callCount())
	}
}

// TestLeaseLossCancelsInvocationBeforeCommit verifies that replacing the
// device lease token while a provider invocation is blocked cancels the
// guarded context and prevents the late result from being committed.
func TestLeaseLossCancelsInvocationBeforeCommit(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	ticks, clock := harness.manualRenewals()
	harness.start("project", "workspace", nil)

	releaseInvocation := make(chan struct{})
	harness.invoker.blockAllInvocations(releaseInvocation)

	reconcileDone := make(chan error, 1)
	go func() {
		reconcileDone <- harness.manager.Reconcile(context.Background())
	}()

	deadline := time.Now().Add(time.Second)
	for harness.invoker.activeCount() == 0 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if harness.invoker.activeCount() == 0 {
		t.Fatal("invocation did not enter blocked state")
	}

	// Capture the original state, then replace the token with a foreign one.
	_ = harness.store.snapshot()
	harness.store.edit(func(state *ManagerState) {
		state.DeviceLease = &LeaseRecord{
			OwnerID:   "rival-manager",
			Token:     "rival-token",
			ExpiresAt: harness.manager.clock.Now().Add(time.Minute),
		}
	})
	ticks <- clock.Now()

	// Wait for Reconcile to return (it should fail with lease loss or context cancel).
	err := <-reconcileDone
	if err == nil {
		t.Fatal("Reconcile returned nil after device lease token was replaced")
	}

	// The work must not have been committed to a terminal state.
	state := harness.store.snapshot()
	work := state.Projects["project"].Works["work"]
	if work.State == WorkStateCompleted || work.State == WorkStateSubmitted {
		t.Fatalf("work state = %s, want non-terminal after lease loss", work.State)
	}

	// The foreign token must still be in the store.
	if state.DeviceLease == nil || state.DeviceLease.Token != "rival-token" {
		t.Fatalf("device lease token = %v, want rival-token", state.DeviceLease)
	}

	// The original manager's invoker cancellation count must be > 0.
	if harness.invoker.cancelCount() == 0 {
		t.Fatal("invocation was not cancelled after lease loss")
	}

	// Release the blocked invocation to clean up.
	close(releaseInvocation)
}

// TestWorkspaceProjectLeaseRenewsDuringLongReview verifies that workspace and
// project leases remain held while review is blocked, and that the guarded
// context stays alive.
func TestWorkspaceProjectLeaseRenewsDuringLongReview(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	ticks, clock := harness.manualRenewals()
	harness.start("project", "workspace", nil)

	// Block the reviewer so the manager stays inside reviewSubmission.
	releaseReview := make(chan struct{})
	harness.reviewer.blockAllReviews(releaseReview)

	reconcileDone := make(chan error, 1)
	go func() {
		reconcileDone <- harness.manager.Reconcile(context.Background())
	}()

	// Wait for the manager to enter the review stage.
	deadline := time.Now().Add(2 * time.Second)
	for {
		state := harness.store.snapshot()
		work := state.Projects["project"].Works["work"]
		if work.State == WorkStateReviewing {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("work did not enter reviewing state, state=%s", work.State)
		}
		time.Sleep(time.Millisecond)
	}

	// Workspace and project leases must be held.
	state := harness.store.snapshot()
	project := state.Projects["project"]
	if project.Lease == nil || project.Lease.OwnerID != harness.manager.config.OwnerID {
		t.Fatalf("project lease = %v, want held by manager", project.Lease)
	}
	if _, ok := state.WorkspaceLeases["workspace"]; !ok {
		t.Fatal("workspace lease was not claimed")
	}

	projectExpiry := project.Lease.ExpiresAt
	workspaceExpiry := state.WorkspaceLeases["workspace"].ExpiresAt
	clock.Advance(time.Minute)
	ticks <- clock.Now()
	waitFor(t, "project and workspace lease renewal", func() bool {
		state := harness.store.snapshot()
		return state.Projects["project"].Lease.ExpiresAt.After(projectExpiry) &&
			state.WorkspaceLeases["workspace"].ExpiresAt.After(workspaceExpiry)
	})

	// Release the review and verify the work completes.
	close(releaseReview)
	err := <-reconcileDone
	if err != nil {
		t.Fatalf("Reconcile returned error after review renewal: %v", err)
	}
	finalState := harness.store.snapshot()
	work := finalState.Projects["project"].Works["work"]
	if work.State != WorkStateCompleted {
		t.Fatalf("work state = %s, want completed", work.State)
	}
}

// TestIntegrationLeaseRenewsAndFencesLateResult verifies that the integration
// lease remains held through the external integration call, that lease loss
// rejects the late result, and that the successor token is preserved.
func TestIntegrationLeaseRenewsAndFencesLateResult(t *testing.T) {
	snapshot := testSnapshot(
		"project", "workspace",
		testUnit("work", WriteSetDisjoint),
	)
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 2)
	harness.start("project", "workspace", nil)

	// Let the work complete dispatch and review, then block integration.
	releaseIntegration := make(chan struct{})
	harness.integrator.blockAllIntegrations(releaseIntegration)

	// Run reconcile in the background.
	reconcileDone := make(chan error, 1)
	go func() {
		reconcileDone <- harness.manager.Reconcile(context.Background())
	}()

	// Wait for one snapshot that proves both the external integration phase
	// and the manager-owned integration lease are present.
	deadline := time.Now().Add(2 * time.Second)
	for {
		state := harness.store.snapshot()
		work := state.Projects["project"].Works["work"]
		lease, claimed := state.IntegrationLeases["workspace"]
		if work.State == WorkStateIntegrating && claimed &&
			lease.OwnerID == harness.manager.config.OwnerID {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"work did not enter manager-owned integration, state=%s lease=%#v",
				work.State,
				lease,
			)
		}
		time.Sleep(time.Millisecond)
	}

	// Capture the original integration lease token.
	originalState := harness.store.snapshot()
	origIntegToken, ok := originalState.IntegrationLeases["workspace"]
	if !ok {
		t.Fatalf("integration lease was not claimed; state=%s integrationLeases=%v",
			originalState.Projects["project"].Works["work"].State, originalState.IntegrationLeases)
	}

	// Replace the integration lease token (simulating ownership loss).
	harness.store.edit(func(state *ManagerState) {
		state.IntegrationLeases["workspace"] = LeaseRecord{
			OwnerID:   "rival-manager",
			Token:     "rival-integration-token",
			ExpiresAt: harness.manager.clock.Now().Add(time.Minute),
		}
	})

	// Release the integrator and verify the late result is rejected.
	close(releaseIntegration)
	err := <-reconcileDone
	if err == nil {
		t.Fatal("Reconcile returned nil after integration lease token was replaced")
	}

	// The work must not have been committed to completed.
	finalState := harness.store.snapshot()
	work := finalState.Projects["project"].Works["work"]
	if work.State == WorkStateCompleted {
		t.Fatalf("work state = %s, want non-completed after integration lease loss", work.State)
	}

	// The successor token must still be in the store.
	finalIntegLease, ok := finalState.IntegrationLeases["workspace"]
	if !ok {
		t.Fatal("integration lease disappeared after loss")
	}
	if finalIntegLease.Token != "rival-integration-token" {
		t.Fatalf("integration lease token = %q, want rival-integration-token", finalIntegLease.Token)
	}

	// The original token must not match the current lease.
	if origIntegToken.Token == finalIntegLease.Token {
		t.Fatal("original integration token was not replaced")
	}
}

func TestLeaseSupervisorRenewsAllTrackedScopesAndDeniesTakeover(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetDisjoint))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	ticks, clock := harness.manualRenewals()
	harness.start("project", "workspace", nil)
	release := make(chan struct{})
	harness.integrator.blockAllIntegrations(release)
	done := make(chan error, 1)
	go func() { done <- harness.manager.Reconcile(context.Background()) }()

	waitFor(t, "manager-owned integration lease", func() bool {
		state := harness.store.snapshot()
		work := state.Projects["project"].Works["work"]
		_, integration := state.IntegrationLeases["workspace"]
		return work.State == WorkStateIntegrating && state.DeviceLease != nil &&
			state.Projects["project"].Lease != nil && integration &&
			state.WorkspaceLeases["workspace"].Token != ""
	})
	before := harness.store.snapshot()
	deviceExpiry := before.DeviceLease.ExpiresAt
	projectExpiry := before.Projects["project"].Lease.ExpiresAt
	workspaceExpiry := before.WorkspaceLeases["workspace"].ExpiresAt
	integrationExpiry := before.IntegrationLeases["workspace"].ExpiresAt
	clock.Advance(time.Minute)
	ticks <- clock.Now()
	waitFor(t, "supervisor renewal", func() bool {
		state := harness.store.snapshot()
		return state.DeviceLease.ExpiresAt.After(deviceExpiry) &&
			state.Projects["project"].Lease.ExpiresAt.After(projectExpiry) &&
			state.WorkspaceLeases["workspace"].ExpiresAt.After(workspaceExpiry) &&
			state.IntegrationLeases["workspace"].ExpiresAt.After(integrationExpiry)
	})
	second, err := NewManager(
		ManagerConfig{OwnerID: harness.manager.config.OwnerID, LeaseDuration: time.Minute}, clock,
		harness.store, harness.workflow, fakeSelector{capacity: 1}, harness.isolation,
		harness.invoker, harness.recovery, harness.evidence, harness.reviewer, harness.integrator, harness.events,
	)
	if err != nil {
		t.Fatalf("NewManager(second): %v", err)
	}
	if _, err := second.claimDevice(context.Background()); !errors.Is(err, ErrDeviceLeaseHeld) {
		t.Fatalf("second manager device claim error = %v, want ErrDeviceLeaseHeld", err)
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
}

func TestLeaseLossImmediatelyFencesProviderAndReviewResults(t *testing.T) {
	for _, stage := range []string{"provider", "review"} {
		t.Run(stage, func(t *testing.T) {
			snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
			harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
			harness.start("project", "workspace", nil)
			release := make(chan struct{})
			if stage == "provider" {
				harness.invoker.blockAllInvocations(release)
			} else {
				harness.reviewer.blockAllReviews(release)
			}
			done := make(chan error, 1)
			go func() { done <- harness.manager.Reconcile(context.Background()) }()
			waitFor(t, stage+" external call", func() bool {
				if stage == "provider" {
					return harness.invoker.activeCount() > 0
				}
				return harness.store.snapshot().Projects["project"].Works["work"].State == WorkStateReviewing
			})
			harness.store.edit(func(state *ManagerState) {
				state.DeviceLease = &LeaseRecord{
					OwnerID: "successor", Token: "successor-device", ExpiresAt: time.Now().Add(time.Minute),
				}
			})
			close(release)
			if err := <-done; err == nil {
				t.Fatal("Reconcile returned nil after immediate ownership loss")
			}
			state := harness.store.snapshot()
			work := state.Projects["project"].Works["work"]
			if stage == "provider" && (work.State == WorkStateSubmitted || work.State == WorkStateCompleted) {
				t.Fatalf("provider result committed after lease loss: %s", work.State)
			}
			if stage == "review" && (work.State == WorkStatePendingIntegration || work.State == WorkStateCompleted) {
				t.Fatalf("review result committed after lease loss: %s", work.State)
			}
			if state.DeviceLease == nil || state.DeviceLease.Token != "successor-device" {
				t.Fatalf("device successor was not retained: %#v", state.DeviceLease)
			}
		})
	}
}

func TestLeaseLossImmediatelyFencesIntegrationResult(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetDisjoint))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	release := make(chan struct{})
	harness.integrator.blockAllIntegrations(release)
	done := make(chan error, 1)
	go func() { done <- harness.manager.Reconcile(context.Background()) }()
	waitFor(t, "integration", func() bool {
		state := harness.store.snapshot()
		_, claimed := state.IntegrationLeases["workspace"]
		return state.Projects["project"].Works["work"].State == WorkStateIntegrating && claimed
	})
	harness.store.edit(func(state *ManagerState) {
		state.IntegrationLeases["workspace"] = LeaseRecord{
			OwnerID: "successor", Token: "successor-integration", ExpiresAt: time.Now().Add(time.Minute),
		}
	})
	close(release)
	if err := <-done; err == nil {
		t.Fatal("Reconcile returned nil after immediate integration lease loss")
	}
	state := harness.store.snapshot()
	if work := state.Projects["project"].Works["work"]; work.State == WorkStateCompleted {
		t.Fatal("integration result committed after lease loss")
	}
	if lease := state.IntegrationLeases["workspace"]; lease.Token != "successor-integration" {
		t.Fatalf("integration successor token = %q", lease.Token)
	}
}

func TestLeaseFenceRevalidatesAfterCASConflict(t *testing.T) {
	snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetUnknown))
	harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
	harness.start("project", "workspace", nil)
	release := make(chan struct{})
	harness.invoker.blockAllInvocations(release)
	arm := make(chan struct{})
	harness.manager.store = &casConflictStore{
		store: harness.store,
		armed: arm,
		onConflict: func() {
			harness.store.edit(func(state *ManagerState) {
				state.DeviceLease = &LeaseRecord{
					OwnerID: "successor", Token: "successor-after-conflict", ExpiresAt: time.Now().Add(time.Minute),
				}
			})
		},
	}
	done := make(chan error, 1)
	go func() { done <- harness.manager.Reconcile(context.Background()) }()
	waitFor(t, "provider invocation", func() bool { return harness.invoker.activeCount() > 0 })
	close(arm)
	close(release)
	if err := <-done; err == nil {
		t.Fatal("Reconcile returned nil after CAS-conflict ownership loss")
	}
	state := harness.store.snapshot()
	if work := state.Projects["project"].Works["work"]; work.State == WorkStateSubmitted {
		t.Fatal("provider result committed after conflict retry observed a successor")
	}
	if state.DeviceLease == nil || state.DeviceLease.Token != "successor-after-conflict" {
		t.Fatalf("successor device lease was not retained: %#v", state.DeviceLease)
	}
}

func TestLeaseCleanupReleasesOwnedClaimsAndPreservesSuccessors(t *testing.T) {
	t.Run("normal", func(t *testing.T) {
		snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetDisjoint))
		harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
		harness.start("project", "workspace", nil)
		if err := harness.manager.Reconcile(context.Background()); err != nil {
			t.Fatalf("Reconcile: %v", err)
		}
		state := harness.store.snapshot()
		if state.DeviceLease != nil || state.Projects["project"].Lease != nil {
			t.Fatalf("owned device/project claims remained: %#v %#v", state.DeviceLease, state.Projects["project"].Lease)
		}
		if _, ok := state.WorkspaceLeases["workspace"]; ok {
			t.Fatal("owned workspace claim remained")
		}
		if _, ok := state.IntegrationLeases["workspace"]; ok {
			t.Fatal("owned integration claim remained")
		}
	})
	t.Run("successors", func(t *testing.T) {
		snapshot := testSnapshot("project", "workspace", testUnit("work", WriteSetDisjoint))
		harness := newHarness(t, map[ProjectID]ProjectWorkflowSnapshot{"project": snapshot}, 1)
		harness.start("project", "workspace", nil)
		release := make(chan struct{})
		harness.integrator.blockAllIntegrations(release)
		done := make(chan error, 1)
		go func() { done <- harness.manager.Reconcile(context.Background()) }()
		waitFor(t, "integration claim", func() bool {
			state := harness.store.snapshot()
			_, claimed := state.IntegrationLeases["workspace"]
			return state.Projects["project"].Works["work"].State == WorkStateIntegrating && claimed
		})
		harness.store.edit(func(state *ManagerState) {
			expires := time.Now().Add(time.Minute)
			project := state.Projects["project"]
			project.Lease = &LeaseRecord{OwnerID: "successor", Token: "successor-project", ExpiresAt: expires}
			state.Projects["project"] = project
			state.WorkspaceLeases["workspace"] = LeaseRecord{OwnerID: "successor", Token: "successor-workspace", ExpiresAt: expires}
			state.IntegrationLeases["workspace"] = LeaseRecord{OwnerID: "successor", Token: "successor-integration", ExpiresAt: expires}
		})
		close(release)
		if err := <-done; err == nil {
			t.Fatal("Reconcile returned nil after successor replacement")
		}
		state := harness.store.snapshot()
		if state.DeviceLease != nil {
			t.Fatalf("owned device claim was not released: %#v", state.DeviceLease)
		}
		if state.Projects["project"].Lease.Token != "successor-project" ||
			state.WorkspaceLeases["workspace"].Token != "successor-workspace" ||
			state.IntegrationLeases["workspace"].Token != "successor-integration" {
			t.Fatalf("successor claims were not retained: %#v", state)
		}
	})
}