# Sprint 2: Diarization Enhancements

> **Priority**: 2 | **Owner**: Backend + Frontend | **Complexity**: Low-Medium

---

## Validation Status (2025-12-27)

### ✅ 100% COMPLETE

| Component | Location | Status |
|-----------|----------|--------|
| `DiarizationEngine` | `infrastructure/diarization/engine.py` | ✅ Streaming + offline modes, 13 methods |
| `DiarizationMixin` | `grpc/_mixins/diarization.py` | ✅ `RefineSpeakerDiarization`, `RenameSpeaker`, 9 methods |
| `DiarizationMixin` (job) | `grpc/_mixins/diarization_job.py` | ✅ Job status + cancel handling |
| `CancelDiarizationJob` RPC | `grpc/_mixins/diarization_job.py:153-223` | ✅ Full cancellation with state management |
| Session management | `infrastructure/diarization/session.py` | ✅ Streaming session |
| Speaker assigner | `infrastructure/diarization/assigner.py` | ✅ `assign_speaker`, batch assignment |
| DTOs | `infrastructure/diarization/dto.py` | ✅ Diarization data transfer |
| Job repository | `persistence/repositories/diarization_job_repo.py` | ✅ Complete |
| `get_active_for_meeting()` | `diarization_job_repo.py:174-195` | ✅ Concurrency guard |
| ORM model | `persistence/models/core/diarization.py` | ✅ `DiarizationJobModel` |
| Domain errors | `domain/summarization/ports.py` | ✅ Error hierarchy |
| Tests | `tests/grpc/test_diarization_*.py` | ✅ Cancel, refine tests |
| Tests | `tests/infrastructure/test_diarization.py` | ✅ Unit tests |
| Frontend hook | `client/src/hooks/use-diarization.ts` | ✅ Polling + cancel support |

---

## Objective

Add targeted enhancements to the existing diarization system: job cancellation, concurrency guards, real progress tracking, proper error types, and improved frontend UX. This sprint does **not** extract an application service layer—the existing mixin architecture is well-suited to the diarization workflow.

---

## Rationale: Why No Application Service?

After thorough analysis, extracting a `DiarizationService` would introduce redundancy:

| What Sprint Originally Proposed | What Already Exists |
|--------------------------------|---------------------|
| Job persistence in service | `SqlAlchemyDiarizationJobRepository` (10 methods) |
| Crash recovery in service | `RecoveryService.recover_crashed_diarization_jobs()` |
| Job cleanup in service | `repo.prune_completed(ttl_seconds)` |
| Background task management | Mixin uses `run_in_executor` pattern correctly |

**Key insight**: `SummarizationService` exists because it orchestrates multiple providers, handles consent, and performs citation verification. Diarization has one engine with two modes—no orchestration needed. The mixin pattern is appropriate here.

**References**:
- `src/noteflow/grpc/_mixins/diarization.py` — Current implementation (~580 lines, well-structured)
- `src/noteflow/infrastructure/persistence/repositories/diarization_job_repo.py` — Complete repository
- `src/noteflow/application/services/recovery_service.py` — Already handles job recovery

---

## Current State Analysis

### What Exists (Complete)

| Component | Location | Status |
|-----------|----------|--------|
| Diarization Engine | `infrastructure/diarization/engine.py` | Full (streaming + offline) |
| Job Model | `persistence/models/core/diarization.py` | Complete (8 fields) |
| Job Repository | `persistence/repositories/diarization_job_repo.py` | Complete (10 methods) |
| gRPC Mixin | `grpc/_mixins/diarization.py` | DB-first with memory fallback |
| Recovery | `application/services/recovery_service.py` | Handles orphaned jobs |
| Proto RPCs | `noteflow.proto` | 3 RPCs implemented |

### What's Missing (This Sprint)

| Gap | Impact | Priority |
|-----|--------|----------|
| No Cancel RPC | Users can't stop long-running jobs | P0 |
| No concurrency guard | Can start duplicate jobs for same meeting | P1 |
| Fake progress tracking | UI shows 0% → 50% → 100% jumps | P1 |
| No error hierarchy | Generic exceptions, poor gRPC status mapping | P2 |
| Frontend polling issues | 1s interval, no backoff, weak error handling | P2 |

---

## Implementation Tasks

### Task 1: Add Cancel RPC

**File**: `src/noteflow/grpc/proto/noteflow.proto`

Add after existing diarization RPCs (~line 39):

```protobuf
rpc CancelDiarizationJob(CancelDiarizationJobRequest) returns (CancelDiarizationJobResponse);
```

Add message definitions (~line 568):

```protobuf
message CancelDiarizationJobRequest {
  string job_id = 1;
}

message CancelDiarizationJobResponse {
  bool cancelled = 1;
  JobStatus final_status = 2;
  string error_message = 3;
}
```

**File**: `src/noteflow/grpc/_mixins/diarization.py`

Add handler method to `DiarizationMixin`:

```python
async def CancelDiarizationJob(
    self: ServicerHost,
    request: noteflow_pb2.CancelDiarizationJobRequest,
    context: grpc.aio.ServicerContext,
) -> noteflow_pb2.CancelDiarizationJobResponse:
    """Cancel a running or queued diarization job."""
    job_id = request.job_id

    # Cancel active task if running
    task = self._diarization_tasks.get(job_id)
    if task and not task.done():
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

    # Update database status
    async with self._create_repository_provider() as repo:
        if not repo.supports_diarization_jobs:
            # Memory-only mode: update in-memory dict
            if job_id in self._diarization_jobs:
                self._diarization_jobs[job_id] = self._diarization_jobs[job_id]._replace(
                    status=noteflow_pb2.JOB_STATUS_FAILED,
                    error_message="Cancelled by user",
                )
                return noteflow_pb2.CancelDiarizationJobResponse(
                    cancelled=True,
                    final_status=noteflow_pb2.JOB_STATUS_FAILED,
                )
            return noteflow_pb2.CancelDiarizationJobResponse(
                cancelled=False,
                error_message="Job not found",
            )

        job = await repo.diarization_jobs.get(job_id)
        if not job:
            return noteflow_pb2.CancelDiarizationJobResponse(
                cancelled=False,
                error_message="Job not found",
            )

        if job.status not in (
            noteflow_pb2.JOB_STATUS_QUEUED,
            noteflow_pb2.JOB_STATUS_RUNNING,
        ):
            return noteflow_pb2.CancelDiarizationJobResponse(
                cancelled=False,
                final_status=job.status,
                error_message="Job already completed or failed",
            )

        await repo.diarization_jobs.update_status(
            job_id,
            noteflow_pb2.JOB_STATUS_FAILED,
            error_message="Cancelled by user",
        )
        await repo.commit()

    _logger.info("Cancelled diarization job", extra={"job_id": job_id})

    return noteflow_pb2.CancelDiarizationJobResponse(
        cancelled=True,
        final_status=noteflow_pb2.JOB_STATUS_FAILED,
    )
```

---

### Task 2: Add Concurrency Guard

Prevent multiple simultaneous diarization jobs for the same meeting.

**File**: `src/noteflow/infrastructure/persistence/repositories/diarization_job_repo.py`

Add method:

```python
async def get_active_for_meeting(self, meeting_id: str) -> DiarizationJob | None:
    """Get active (queued or running) job for a meeting, if any.

    Args:
        meeting_id: Meeting UUID as string.

    Returns:
        Active job or None.
    """
    stmt = (
        select(DiarizationJobModel)
        .where(
            DiarizationJobModel.meeting_id == meeting_id,
            DiarizationJobModel.status.in_([
                JOB_STATUS_QUEUED,
                JOB_STATUS_RUNNING,
            ]),
        )
        .order_by(DiarizationJobModel.created_at.desc())
        .limit(1)
    )
    model = await self._execute_scalar(stmt)
    return self._to_domain(model) if model else None
```

**File**: `src/noteflow/domain/ports/repositories.py`

Add to `DiarizationJobRepository` protocol:

```python
async def get_active_for_meeting(self, meeting_id: str) -> DiarizationJob | None:
    """Get active job for meeting, if any."""
    ...
```

**File**: `src/noteflow/grpc/_mixins/diarization.py`

Modify `RefineSpeakerDiarization` to check for existing job:

```python
async def RefineSpeakerDiarization(
    self: ServicerHost,
    request: noteflow_pb2.RefineSpeakerDiarizationRequest,
    context: grpc.aio.ServicerContext,
) -> noteflow_pb2.RefineSpeakerDiarizationResponse:
    """Start speaker diarization refinement job."""
    meeting_id = request.meeting_id

    async with self._create_repository_provider() as repo:
        # Check for existing active job
        if repo.supports_diarization_jobs:
            existing = await repo.diarization_jobs.get_active_for_meeting(meeting_id)
            if existing:
                return noteflow_pb2.RefineSpeakerDiarizationResponse(
                    job_id=existing.job_id,
                    status=existing.status,
                    error_message=f"Job {existing.job_id} already active for this meeting",
                )

        # ... rest of existing implementation ...
```

---

### Task 3: Add Real Progress Tracking

Replace placeholder `progress = 50` with time-based estimation.

**File**: `src/noteflow/infrastructure/persistence/models/core/diarization.py`

Add field to `DiarizationJobModel`:

```python
audio_duration_seconds: Mapped[float | None] = mapped_column(
    Float,
    nullable=True,
    doc="Audio duration for progress estimation",
)
```

**File**: Alembic migration

Create migration `alembic revision --autogenerate -m "add_audio_duration_to_diarization_jobs"`:

```python
"""Add audio_duration_seconds to diarization_jobs.

Revision ID: xxxx
"""

from alembic import op
import sqlalchemy as sa


def upgrade() -> None:
    op.add_column(
        "diarization_jobs",
        sa.Column("audio_duration_seconds", sa.Float(), nullable=True),
        schema="noteflow",
    )


def downgrade() -> None:
    op.drop_column("diarization_jobs", "audio_duration_seconds", schema="noteflow")
```

**File**: `src/noteflow/grpc/_mixins/diarization.py`

Update job creation to capture audio duration:

```python
# In RefineSpeakerDiarization, after loading meeting
audio_duration = None
if meeting.duration_seconds:
    audio_duration = meeting.duration_seconds

job = DiarizationJob(
    job_id=job_id,
    meeting_id=meeting_id,
    status=noteflow_pb2.JOB_STATUS_QUEUED,
    audio_duration_seconds=audio_duration,
    # ... other fields ...
)
```

Update `GetDiarizationJobStatus` for real progress:

```python
async def GetDiarizationJobStatus(
    self: ServicerHost,
    request: noteflow_pb2.GetDiarizationJobStatusRequest,
    context: grpc.aio.ServicerContext,
) -> noteflow_pb2.DiarizationJobStatus:
    """Get status of diarization job."""
    job_id = request.job_id

    async with self._create_repository_provider() as repo:
        # ... existing job lookup ...

        # Calculate progress
        progress_percent = 0
        if job.status == noteflow_pb2.JOB_STATUS_COMPLETED:
            progress_percent = 100
        elif job.status == noteflow_pb2.JOB_STATUS_RUNNING and job.started_at:
            elapsed = (utc_now() - job.started_at).total_seconds()
            if job.audio_duration_seconds and job.audio_duration_seconds > 0:
                # Estimate: ~10 seconds processing per 60 seconds audio
                estimated_duration = job.audio_duration_seconds * 0.17
                progress_percent = min(95, int((elapsed / estimated_duration) * 100))
            else:
                # Fallback: assume 2 minutes total
                progress_percent = min(95, int((elapsed / 120) * 100))

        return noteflow_pb2.DiarizationJobStatus(
            job_id=job_id,
            status=job.status,
            segments_updated=job.segments_updated,
            speaker_ids=list(job.speaker_ids) if job.speaker_ids else [],
            error_message=job.error_message or "",
            progress_percent=progress_percent,
        )
```

---

### Task 4: Add Error Hierarchy

Follow the `SummarizationError` pattern from `domain/summarization/ports.py`.

**File**: `src/noteflow/domain/ports/diarization.py` (new file)

```python
"""Diarization domain port interfaces and errors."""

from __future__ import annotations


class DiarizationError(Exception):
    """Base exception for diarization errors."""


class JobNotFoundError(DiarizationError):
    """Raised when diarization job doesn't exist."""

    def __init__(self, job_id: str) -> None:
        self.job_id = job_id
        super().__init__(f"Diarization job not found: {job_id}")


class JobAlreadyActiveError(DiarizationError):
    """Raised when meeting already has an active diarization job."""

    def __init__(self, meeting_id: str, existing_job_id: str) -> None:
        self.meeting_id = meeting_id
        self.existing_job_id = existing_job_id
        super().__init__(
            f"Meeting {meeting_id} already has active job: {existing_job_id}"
        )


class AudioNotFoundError(DiarizationError):
    """Raised when meeting has no audio file for diarization."""

    def __init__(self, meeting_id: str) -> None:
        self.meeting_id = meeting_id
        super().__init__(f"No audio file for meeting: {meeting_id}")


class EngineNotReadyError(DiarizationError):
    """Raised when diarization engine is not initialized."""

    def __init__(self, reason: str = "Engine not loaded") -> None:
        self.reason = reason
        super().__init__(f"Diarization engine not ready: {reason}")
```

**File**: `src/noteflow/grpc/_mixins/diarization.py`

Import and use error types:

```python
from noteflow.domain.ports.diarization import (
    AudioNotFoundError,
    DiarizationError,
    EngineNotReadyError,
    JobAlreadyActiveError,
    JobNotFoundError,
)
```

Map to gRPC status codes:

```python
async def RefineSpeakerDiarization(self, request, context):
    try:
        # ... existing logic ...
    except JobAlreadyActiveError as e:
        context.set_code(grpc.StatusCode.ALREADY_EXISTS)
        context.set_details(str(e))
        return noteflow_pb2.RefineSpeakerDiarizationResponse(
            job_id=e.existing_job_id,
            error_message=str(e),
        )
    except AudioNotFoundError as e:
        context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
        context.set_details(str(e))
        return noteflow_pb2.RefineSpeakerDiarizationResponse(error_message=str(e))
    except EngineNotReadyError as e:
        context.set_code(grpc.StatusCode.UNAVAILABLE)
        context.set_details(str(e))
        return noteflow_pb2.RefineSpeakerDiarizationResponse(error_message=str(e))
```

---

### Task 5: Improve Frontend Hook

**File**: `client/src/hooks/use-diarization.ts`

Replace with improved version:

```typescript
import { useState, useCallback, useRef, useEffect } from 'react';
import { invoke } from '@tauri-apps/api/tauri';

interface DiarizationJobStatus {
  jobId: string;
  status: 'queued' | 'running' | 'completed' | 'failed';
  segmentsUpdated: number;
  speakerIds: string[];
  errorMessage?: string;
  progressPercent: number;
}

interface UseDiarizationResult {
  isProcessing: boolean;
  progress: number;
  error: string | null;
  speakerIds: string[];
  startDiarization: (meetingId: string, numSpeakers?: number) => Promise<void>;
  renameSpeaker: (meetingId: string, oldId: string, newName: string) => Promise<number>;
  cancelDiarization: () => Promise<void>;
}

// Polling configuration
const INITIAL_POLL_MS = 1000;
const MAX_POLL_MS = 5000;
const BACKOFF_FACTOR = 1.5;
const MAX_RETRIES = 3;

export function useDiarization(): UseDiarizationResult {
  const [isProcessing, setIsProcessing] = useState(false);
  const [progress, setProgress] = useState(0);
  const [error, setError] = useState<string | null>(null);
  const [speakerIds, setSpeakerIds] = useState<string[]>([]);

  const currentJobId = useRef<string | null>(null);
  const pollTimeoutRef = useRef<NodeJS.Timeout | null>(null);
  const pollIntervalMs = useRef(INITIAL_POLL_MS);
  const lastStatus = useRef<string | null>(null);
  const retryCount = useRef(0);

  const stopPolling = useCallback(() => {
    if (pollTimeoutRef.current) {
      clearTimeout(pollTimeoutRef.current);
      pollTimeoutRef.current = null;
    }
    pollIntervalMs.current = INITIAL_POLL_MS;
    retryCount.current = 0;
  }, []);

  const pollJobStatus = useCallback(async () => {
    if (!currentJobId.current) return;

    try {
      const status = await invoke<DiarizationJobStatus>(
        'get_diarization_job_status',
        { jobId: currentJobId.current }
      );

      // Reset retry count on success
      retryCount.current = 0;

      setProgress(status.progressPercent);

      // Apply backoff if status unchanged
      if (status.status === lastStatus.current && status.status === 'running') {
        pollIntervalMs.current = Math.min(
          pollIntervalMs.current * BACKOFF_FACTOR,
          MAX_POLL_MS
        );
      } else {
        pollIntervalMs.current = INITIAL_POLL_MS;
      }
      lastStatus.current = status.status;

      if (status.status === 'completed') {
        setIsProcessing(false);
        setSpeakerIds(status.speakerIds);
        stopPolling();
        return;
      }

      if (status.status === 'failed') {
        setIsProcessing(false);
        setError(status.errorMessage || 'Diarization failed');
        stopPolling();
        return;
      }

      // Schedule next poll
      pollTimeoutRef.current = setTimeout(pollJobStatus, pollIntervalMs.current);
    } catch (err) {
      retryCount.current++;

      if (retryCount.current >= MAX_RETRIES) {
        setError('Lost connection to server');
        setIsProcessing(false);
        stopPolling();
        return;
      }

      // Retry with backoff
      pollIntervalMs.current = Math.min(
        pollIntervalMs.current * BACKOFF_FACTOR,
        MAX_POLL_MS
      );
      pollTimeoutRef.current = setTimeout(pollJobStatus, pollIntervalMs.current);
    }
  }, [stopPolling]);

  const startDiarization = useCallback(
    async (meetingId: string, numSpeakers?: number) => {
      setIsProcessing(true);
      setProgress(0);
      setError(null);
      setSpeakerIds([]);
      lastStatus.current = null;
      pollIntervalMs.current = INITIAL_POLL_MS;

      try {
        const response = await invoke<{ jobId: string; errorMessage?: string }>(
          'refine_speaker_diarization',
          { meetingId, numSpeakers }
        );

        if (response.errorMessage) {
          // Job already exists or other error
          setError(response.errorMessage);
          setIsProcessing(false);
          return;
        }

        currentJobId.current = response.jobId;

        // Start polling
        pollTimeoutRef.current = setTimeout(pollJobStatus, pollIntervalMs.current);
      } catch (err) {
        setIsProcessing(false);
        setError(err instanceof Error ? err.message : 'Failed to start diarization');
      }
    },
    [pollJobStatus]
  );

  const renameSpeaker = useCallback(
    async (meetingId: string, oldId: string, newName: string): Promise<number> => {
      const result = await invoke<{ segmentsUpdated: number }>(
        'rename_speaker',
        { meetingId, oldSpeakerId: oldId, newSpeakerName: newName }
      );
      return result.segmentsUpdated;
    },
    []
  );

  const cancelDiarization = useCallback(async () => {
    if (!currentJobId.current) return;

    try {
      await invoke('cancel_diarization_job', { jobId: currentJobId.current });
    } catch {
      // Ignore cancel errors
    }

    stopPolling();
    setIsProcessing(false);
    setProgress(0);
    setError(null);
    currentJobId.current = null;
  }, [stopPolling]);

  // Cleanup on unmount
  useEffect(() => {
    return () => {
      stopPolling();
    };
  }, [stopPolling]);

  return {
    isProcessing,
    progress,
    error,
    speakerIds,
    startDiarization,
    renameSpeaker,
    cancelDiarization,
  };
}
```

---

### Task 6: Add Tauri Cancel Command

**File**: `client/src-tauri/src/commands/diarization.rs`

Add cancel command:

```rust
#[tauri::command]
pub async fn cancel_diarization_job(
    job_id: String,
    state: tauri::State<'_, AppState>,
) -> Result<CancelDiarizationResponse, String> {
    let client = state.grpc_client.lock().await;
    let response = client
        .cancel_diarization_job(CancelDiarizationJobRequest { job_id })
        .await
        .map_err(|e| e.to_string())?;

    Ok(CancelDiarizationResponse {
        cancelled: response.cancelled,
        final_status: response.final_status,
        error_message: response.error_message,
    })
}
```

**File**: `client/src-tauri/src/lib.rs`

Register the command in the Tauri builder.

**File**: `client/src/lib/tauri.ts`

Add TypeScript binding:

```typescript
export async function cancelDiarizationJob(jobId: string): Promise<{
  cancelled: boolean;
  finalStatus: string;
  errorMessage?: string;
}> {
  return invoke('cancel_diarization_job', { jobId });
}
```

---

## Files Summary

### Files to Create

| File | Purpose | Lines Est. |
|------|---------|------------|
| `src/noteflow/domain/ports/diarization.py` | Error hierarchy | ~40 |
| `alembic/versions/xxx_add_audio_duration.py` | Migration | ~20 |

### Files to Modify

| File | Change | Lines Est. |
|------|--------|------------|
| `src/noteflow/grpc/proto/noteflow.proto` | Add Cancel RPC | +15 |
| `src/noteflow/grpc/_mixins/diarization.py` | Cancel handler, concurrency check, progress calc | +60 |
| `src/noteflow/infrastructure/persistence/repositories/diarization_job_repo.py` | Add `get_active_for_meeting()` | +15 |
| `src/noteflow/infrastructure/persistence/models/core/diarization.py` | Add `audio_duration_seconds` | +5 |
| `src/noteflow/domain/ports/repositories.py` | Add protocol method | +5 |
| `client/src/hooks/use-diarization.ts` | Backoff, retries, error handling | ~120 (rewrite) |
| `client/src-tauri/src/commands/diarization.rs` | Cancel command | +20 |
| `client/src/lib/tauri.ts` | Cancel binding | +10 |

**Total**: ~310 lines (vs. original ~700 lines that would have been redundant)

---

## Acceptance Criteria

### Functional

- [ ] `CancelDiarizationJob` RPC cancels running/queued jobs
- [ ] Cannot start duplicate diarization for same meeting
- [ ] Progress bar shows realistic estimates (not 0→50→100 jumps)
- [ ] Frontend handles server disconnects gracefully (retries with backoff)
- [ ] Errors mapped to appropriate gRPC status codes

### Technical

- [ ] Error hierarchy follows `SummarizationError` pattern
- [ ] Repository method added to protocol (no implicit interface)
- [ ] Migration is reversible
- [ ] No new service layer (mixin handles all logic)

### Quality Gates

- [ ] `pytest tests/quality/` passes
- [ ] `ruff check src/noteflow` clean
- [ ] `mypy src/noteflow` clean
- [ ] Proto regenerated and client updated
- [ ] Frontend tests pass

---

## Test Plan

### Unit Tests

**File**: `tests/unit/test_diarization_mixin.py`

```python
import pytest
from unittest.mock import AsyncMock, MagicMock

from noteflow.domain.ports.diarization import (
    JobAlreadyActiveError,
    JobNotFoundError,
)


@pytest.mark.parametrize(
    ("existing_status", "expected_cancelled"),
    [
        pytest.param("queued", True, id="cancel-queued"),
        pytest.param("running", True, id="cancel-running"),
        pytest.param("completed", False, id="already-completed"),
        pytest.param("failed", False, id="already-failed"),
    ],
)
async def test_cancel_job_by_status(
    diarization_mixin: DiarizationMixin,
    mock_repo: MagicMock,
    existing_status: str,
    expected_cancelled: bool,
) -> None:
    """Cancel succeeds only for queued/running jobs."""
    mock_repo.diarization_jobs.get.return_value = MagicMock(status=existing_status)

    response = await diarization_mixin.CancelDiarizationJob(
        request=MagicMock(job_id="test-job"),
        context=MagicMock(),
    )

    assert response.cancelled == expected_cancelled


async def test_refine_rejects_duplicate_job(
    diarization_mixin: DiarizationMixin,
    mock_repo: MagicMock,
) -> None:
    """Cannot start diarization if job already active."""
    mock_repo.diarization_jobs.get_active_for_meeting.return_value = MagicMock(
        job_id="existing-job",
        status="running",
    )

    response = await diarization_mixin.RefineSpeakerDiarization(
        request=MagicMock(meeting_id="meeting-123"),
        context=MagicMock(),
    )

    assert "already active" in response.error_message
    assert response.job_id == "existing-job"


@pytest.mark.parametrize(
    ("elapsed_seconds", "audio_duration", "expected_progress"),
    [
        pytest.param(0, 60, 0, id="just-started"),
        pytest.param(5, 60, 49, id="halfway"),  # 60s * 0.17 = 10.2s expected
        pytest.param(20, 60, 95, id="capped-at-95"),
        pytest.param(10, None, 8, id="fallback-no-duration"),  # 10/120 = 8%
    ],
)
async def test_progress_calculation(
    elapsed_seconds: float,
    audio_duration: float | None,
    expected_progress: int,
) -> None:
    """Progress calculated from elapsed time and audio duration."""
    # Test implementation
    pass
```

### Integration Tests

**File**: `tests/integration/test_diarization_cancel.py`

```python
import pytest


@pytest.mark.integration
async def test_cancel_running_job(grpc_client, meeting_with_audio) -> None:
    """Can cancel a running diarization job."""
    # Start job
    start_response = await grpc_client.refine_speaker_diarization(
        meeting_id=str(meeting_with_audio.id),
    )
    assert start_response.job_id

    # Cancel immediately
    cancel_response = await grpc_client.cancel_diarization_job(
        job_id=start_response.job_id,
    )
    assert cancel_response.cancelled

    # Verify status is failed
    status = await grpc_client.get_diarization_job_status(start_response.job_id)
    assert status.status == "failed"
    assert "Cancelled" in status.error_message


@pytest.mark.integration
async def test_duplicate_job_rejected(grpc_client, meeting_with_audio) -> None:
    """Cannot start second job while first is active."""
    # Start first job
    first = await grpc_client.refine_speaker_diarization(
        meeting_id=str(meeting_with_audio.id),
    )
    assert first.job_id

    # Try to start second
    second = await grpc_client.refine_speaker_diarization(
        meeting_id=str(meeting_with_audio.id),
    )

    # Should return existing job, not create new one
    assert second.job_id == first.job_id
    assert "already active" in second.error_message
```

---

## Definition of Done

- [ ] All acceptance criteria met
- [ ] Quality gates pass
- [ ] Proto regenerated, stubs updated
- [ ] Tauri commands registered and working
- [ ] Frontend hook tested with dev server
- [ ] No new application service created (mixin-only changes)

---

## Dependencies

- Sprint 0 for proto regeneration scripts
- Existing `DiarizationMixin` implementation
- Existing `SqlAlchemyDiarizationJobRepository`

---

## Failure Modes

| Failure | Detection | Recovery |
|---------|-----------|----------|
| Cancel called after job completes | Check status before cancelling | Return `cancelled=false` with status |
| Race condition on duplicate check | DB constraint or atomic check | Return existing job_id |
| Progress estimation wrong | Audio duration missing | Fallback to time-based estimate |
| Frontend loses connection | Retry count exceeded | Show error, allow manual refresh |

---

## Post-Sprint

- Consider WebSocket for real-time progress (eliminates polling)
- Add speaker confidence scores to UI
- Track diarization quality metrics (DER)
