diff --git a/deployments/entity-linkage/src/stitch/entity_linkage/client.py b/deployments/entity-linkage/src/stitch/entity_linkage/client.py index fb163927..f7c78ae4 100644 --- a/deployments/entity-linkage/src/stitch/entity_linkage/client.py +++ b/deployments/entity-linkage/src/stitch/entity_linkage/client.py @@ -75,6 +75,18 @@ async def iter_oil_gas_fields( ): yield self._to_candidate(item) + async def get_oil_gas_fields_total(self) -> int | None: + """Total resource count, for use as a linkage-progress denominator. + + Fetches a single-item page purely to read ``total_count``; the streaming + iterator used by the pass itself discards that field. Returns ``None`` if + the payload omits an integer count, so progress can still report a + numerator without a denominator. + """ + payload = await self._client.list_oil_gas_fields_page(page=1, page_size=1) + total = payload.get("total_count") + return total if isinstance(total, int) else None + async def list_merge_candidates(self) -> list[dict[str, Any]]: return await self._client.list_merge_candidates() diff --git a/deployments/entity-linkage/src/stitch/entity_linkage/entities.py b/deployments/entity-linkage/src/stitch/entity_linkage/entities.py index 085dad34..e2b7c5d9 100644 --- a/deployments/entity-linkage/src/stitch/entity_linkage/entities.py +++ b/deployments/entity-linkage/src/stitch/entity_linkage/entities.py @@ -99,6 +99,24 @@ class BulkLinkResponse(BaseModel): resources_failed: int = 0 +class LinkProgress(BaseModel): + """In-flight progress of a running linkage pass. + + Written onto the live job record as the pass streams resources, so a poller + can see how far along a multi-hour run is rather than only "running". Field + names mirror :class:`BulkLinkResponse` so the running view and the final + result read consistently. ``total_resources`` is ``None`` when the + denominator could not be fetched; percent is derived by the caller. + """ + + resources_scanned: int + total_resources: int | None + merge_candidates_created: int + merge_candidates_skipped: int + resources_failed: int + updated_at: datetime + + class PaginationParams(BaseModel): page: int = Field(1, ge=1) page_size: int = Field(50, ge=1, le=200) diff --git a/deployments/entity-linkage/src/stitch/entity_linkage/jobs.py b/deployments/entity-linkage/src/stitch/entity_linkage/jobs.py index 6ecdf534..c5f225de 100644 --- a/deployments/entity-linkage/src/stitch/entity_linkage/jobs.py +++ b/deployments/entity-linkage/src/stitch/entity_linkage/jobs.py @@ -24,7 +24,7 @@ logger = logging.getLogger("stitch.entity_linkage") -RunThunk = Callable[[], Awaitable[BaseModel]] +RunThunk = Callable[["JobRecord"], Awaitable[BaseModel]] class JobState(str, Enum): @@ -39,6 +39,7 @@ class JobRecord(BaseModel): params: SerializeAsAny[BaseModel] started_at: datetime finished_at: datetime | None = None + progress: SerializeAsAny[BaseModel] | None = None result: SerializeAsAny[BaseModel] | None = None error: str | None = None @@ -53,7 +54,8 @@ class JobManager: """Single-job, in-memory run manager. State is lost on restart and concurrent runs are rejected. The run body is - supplied per start as a zero-arg coroutine, so this manager is generic. + supplied per start as a coroutine that receives the live ``JobRecord``, so it + can publish progress onto the record while it runs; this manager stays generic. """ def __init__(self) -> None: @@ -81,7 +83,7 @@ async def start(self, params: BaseModel, run: RunThunk) -> JobRecord: async def _run(self, record: JobRecord, run: RunThunk) -> None: try: - record.result = await run() + record.result = await run(record) record.state = JobState.succeeded except Exception as exc: logger.exception("Linkage run %s failed", record.job_id) diff --git a/deployments/entity-linkage/src/stitch/entity_linkage/matching.py b/deployments/entity-linkage/src/stitch/entity_linkage/matching.py index 27b32b59..b2f01470 100644 --- a/deployments/entity-linkage/src/stitch/entity_linkage/matching.py +++ b/deployments/entity-linkage/src/stitch/entity_linkage/matching.py @@ -21,13 +21,15 @@ from __future__ import annotations import logging -from collections.abc import Sequence +from collections.abc import Callable, Sequence +from datetime import UTC, datetime import httpx from stitch.entity_linkage.client import StitchApiClient from stitch.entity_linkage.entities import ( BulkLinkResponse, + LinkProgress, ResourceLinkResult, normalize_country, normalize_name, @@ -36,6 +38,11 @@ logger = logging.getLogger(__name__) +# How often the bulk pass publishes a progress snapshot, in resources scanned. +# Small enough that a 2s poller sees frequent movement, large enough to avoid +# building a progress model on every one of hundreds of thousands of resources. +PROGRESS_UPDATE_EVERY = 100 + # A 4xx from create-merge-candidate is an expected, non-fatal outcome during a # run: the API rejects a duplicate fingerprint (a candidate already exists) or a # resource that has already been merged. We skip those rather than aborting. @@ -171,17 +178,33 @@ async def link_all( apply_merges: bool, page_size: int, initiated_by: str, + on_progress: Callable[[LinkProgress], None] | None = None, ) -> BulkLinkResponse: """Run the bounded matcher over every resource, streaming ids page by page. Groups are de-duplicated by fingerprint across the run, so each block is submitted at most once even though every member rediscovers it. Members of an already-formed block are skipped without re-searching. + + ``on_progress``, when supplied, is called with a :class:`LinkProgress` + snapshot periodically (every ``PROGRESS_UPDATE_EVERY`` resources) and once + more at the end, so a poller can track a long run's advance. """ # Only needed when we will actually POST; skip the (currently unpaginated) # candidate-list fetch entirely on a dry run. known_existing = await _existing_fingerprints(client) if apply_merges else None + # Denominator for progress; only worth an extra request when a progress + # consumer is listening. None if unavailable -- a failure here must not abort + # the pass, so fall back to an unknown total. + total_resources: int | None = None + if on_progress is not None: + try: + total_resources = await client.get_oil_gas_fields_total() + except (StitchAPIError, httpx.HTTPError, OSError) as exc: + logger.warning("Could not fetch resource total for progress: %s", exc) + total_resources = None + groups_by_fingerprint: dict[str, list[int]] = {} processed_ids: set[int] = set() resources_scanned = 0 @@ -189,8 +212,30 @@ async def link_all( skipped = 0 failed = 0 + def emit_progress() -> None: + if on_progress is None: + return + on_progress( + LinkProgress( + resources_scanned=resources_scanned, + total_resources=total_resources, + merge_candidates_created=created, + merge_candidates_skipped=skipped, + resources_failed=failed, + updated_at=datetime.now(UTC), + ) + ) + + # Publish a 0/total snapshot up front so a poller sees a denominator (and any + # progress at all) before the 100th resource -- and, for a run shorter than + # one throttle window, at all, since the state flips to succeeded right after + # the final snapshot with no yield in between. + emit_progress() + async for candidate in client.iter_oil_gas_fields(page_size=page_size): resources_scanned += 1 + if resources_scanned % PROGRESS_UPDATE_EVERY == 0: + emit_progress() if candidate.id in processed_ids: continue @@ -232,6 +277,9 @@ async def link_all( elif was_skipped: skipped += 1 + # Final snapshot so the last poll before completion reflects the exact totals. + emit_progress() + return BulkLinkResponse( initiated_by=initiated_by, apply_merges=apply_merges, diff --git a/deployments/entity-linkage/src/stitch/entity_linkage/routers/link.py b/deployments/entity-linkage/src/stitch/entity_linkage/routers/link.py index 5285444c..b29b5444 100644 --- a/deployments/entity-linkage/src/stitch/entity_linkage/routers/link.py +++ b/deployments/entity-linkage/src/stitch/entity_linkage/routers/link.py @@ -83,13 +83,14 @@ async def start_link_all( """ initiated_by = user_label(auth_context.user) - async def run() -> BulkLinkResponse: + async def run(record: JobRecord) -> BulkLinkResponse: async with StitchApiClient() as client: return await matching.link_all( client, apply_merges=request.apply_merges, page_size=request.page_size, initiated_by=initiated_by, + on_progress=lambda progress: setattr(record, "progress", progress), ) try: diff --git a/deployments/entity-linkage/tests/test_jobs.py b/deployments/entity-linkage/tests/test_jobs.py index 154cba58..617c0f3b 100644 --- a/deployments/entity-linkage/tests/test_jobs.py +++ b/deployments/entity-linkage/tests/test_jobs.py @@ -7,6 +7,7 @@ from stitch.entity_linkage.jobs import ( JobAlreadyRunningError, + JobRecord, JobState, get_job_manager, reset_manager, @@ -21,6 +22,10 @@ class _Result(BaseModel): doubled: int +class _Progress(BaseModel): + scanned: int + + @pytest.fixture(autouse=True) def _reset(): reset_manager() @@ -42,7 +47,7 @@ def test_run_thunk_success_records_result() -> None: async def scenario() -> None: mgr = get_job_manager() - async def run() -> _Result: + async def run(_record: JobRecord) -> _Result: return _Result(doubled=6) record = await mgr.start(_Params(n=3), run) @@ -63,7 +68,7 @@ def test_run_thunk_failure_records_error() -> None: async def scenario() -> None: mgr = get_job_manager() - async def run() -> _Result: + async def run(_record: JobRecord) -> _Result: raise RuntimeError("kaboom") await mgr.start(_Params(), run) @@ -82,7 +87,7 @@ def test_manager_rejects_concurrent_start() -> None: async def scenario() -> None: mgr = get_job_manager() - async def slow() -> _Result: + async def slow(_record: JobRecord) -> _Result: await asyncio.sleep(0.5) return _Result(doubled=0) @@ -91,3 +96,23 @@ async def slow() -> _Result: await mgr.start(_Params(), slow) asyncio.run(scenario()) + + +def test_run_thunk_can_write_progress_onto_record() -> None: + async def scenario() -> None: + mgr = get_job_manager() + + async def run(record: JobRecord) -> _Result: + # The run body writes progress onto the live record; a poller reading + # mgr.current() must see it while the run is still in flight. + record.progress = _Progress(scanned=42) + return _Result(doubled=0) + + await mgr.start(_Params(), run) + for _ in range(200): + if mgr.current().state != JobState.running: + break + await asyncio.sleep(0.01) + assert mgr.current().progress.model_dump() == {"scanned": 42} + + asyncio.run(scenario()) diff --git a/deployments/entity-linkage/tests/test_link_api.py b/deployments/entity-linkage/tests/test_link_api.py index 88591641..35b15174 100644 --- a/deployments/entity-linkage/tests/test_link_api.py +++ b/deployments/entity-linkage/tests/test_link_api.py @@ -71,6 +71,9 @@ async def get_oil_gas_field_detail(self, resource_id: int) -> FieldDetailCandida raise self.detail_error return self.details_by_id[resource_id] + async def get_oil_gas_fields_total(self) -> int | None: + return len(self.items) + async def iter_oil_gas_fields( self, *, @@ -282,6 +285,14 @@ def test_link_all_launches_job_and_status_succeeds(test_client, install_client) assert result["merge_candidates_skipped"] == 0 assert fake.create_calls == [[1, 2]] + # The run publishes a final progress snapshot onto the record. + progress = final["progress"] + assert progress is not None + assert progress["resources_scanned"] == 3 + assert progress["total_resources"] == 3 + assert progress["merge_candidates_created"] == 1 + assert progress["updated_at"] is not None + def test_link_all_records_downstream_failure_in_status( test_client, install_client @@ -311,7 +322,9 @@ def test_link_all_rejects_concurrent_run_with_409( ) -> None: install_client() - async def slow_link_all(client, *, apply_merges, page_size, initiated_by): + async def slow_link_all( + client, *, apply_merges, page_size, initiated_by, on_progress=None + ): import asyncio await asyncio.sleep(0.5) diff --git a/deployments/entity-linkage/tests/test_matching.py b/deployments/entity-linkage/tests/test_matching.py index 617256d8..5a508974 100644 --- a/deployments/entity-linkage/tests/test_matching.py +++ b/deployments/entity-linkage/tests/test_matching.py @@ -34,6 +34,7 @@ def __init__( self.iter_q: list[str | None] = [] self.create_calls: list[list[int]] = [] self.list_candidates_calls = 0 + self.total_calls = 0 async def __aenter__(self) -> "FakeMatchingClient": return self @@ -45,6 +46,10 @@ async def get_oil_gas_field_detail(self, resource_id: int) -> FieldDetailCandida self.detail_calls.append(resource_id) return self.details_by_id[resource_id] + async def get_oil_gas_fields_total(self) -> int | None: + self.total_calls += 1 + return len(self.items) + async def iter_oil_gas_fields( self, *, @@ -407,3 +412,105 @@ async def test_link_all_does_not_swallow_programming_errors() -> None: await matching.link_all( client, apply_merges=True, page_size=200, initiated_by="Tester" ) + + +@pytest.mark.anyio +async def test_link_all_reports_progress_via_callback() -> None: + # With fewer resources than PROGRESS_UPDATE_EVERY, an initial 0 snapshot and a + # final snapshot are emitted; the final must carry the exact totals and the + # fetched denominator. + client = FakeMatchingClient( + items=[ + FieldCandidate(id=1, name="Alpha", country="US"), + FieldCandidate(id=2, name="alpha", country="US"), + FieldCandidate(id=3, name="Beta", country="CA"), + ], + details_by_id={ + 1: FieldDetailCandidate(id=1, name="Alpha", country="US"), + 2: FieldDetailCandidate(id=2, name="alpha", country="US"), + 3: FieldDetailCandidate(id=3, name="Beta", country="CA"), + }, + ) + + snapshots: list = [] + await matching.link_all( + client, + apply_merges=True, + page_size=200, + initiated_by="Tester", + on_progress=snapshots.append, + ) + + assert snapshots, "expected at least a final progress snapshot" + # An up-front 0/total snapshot gives pollers a denominator immediately. + assert snapshots[0].resources_scanned == 0 + assert snapshots[0].total_resources == 3 + final = snapshots[-1] + assert final.resources_scanned == 3 + assert final.total_resources == 3 + assert final.merge_candidates_created == 1 + assert final.merge_candidates_skipped == 0 + assert final.resources_failed == 0 + assert final.updated_at is not None + # The denominator is fetched when a progress consumer is listening. + assert client.total_calls == 1 + + +@pytest.mark.anyio +async def test_link_all_emits_progress_mid_run() -> None: + # Scan more than PROGRESS_UPDATE_EVERY resources so the throttled heartbeat + # fires during the run, not just the final snapshot. Unique names keep every + # resource a singleton (no match groups), isolating the progress cadence. + count = matching.PROGRESS_UPDATE_EVERY * 2 + 5 + items = [ + FieldCandidate(id=i, name=f"Field {i}", country="US") + for i in range(1, count + 1) + ] + details_by_id = { + i: FieldDetailCandidate(id=i, name=f"Field {i}", country="US") + for i in range(1, count + 1) + } + client = FakeMatchingClient(items=items, details_by_id=details_by_id) + + snapshots: list = [] + await matching.link_all( + client, + apply_merges=False, + page_size=200, + initiated_by="Tester", + on_progress=snapshots.append, + ) + + # Initial 0 snapshot, two throttled emits (100, 200), and the final one (205). + assert len(snapshots) >= 4 + scanned_values = [s.resources_scanned for s in snapshots] + assert scanned_values == sorted(scanned_values) + assert len(set(scanned_values)) == len(scanned_values) + assert scanned_values[0] == 0 + assert matching.PROGRESS_UPDATE_EVERY in scanned_values + assert snapshots[-1].resources_scanned == count + # Denominator is fetched once and carried on every snapshot. + assert client.total_calls == 1 + assert all(s.total_resources == count for s in snapshots) + + +@pytest.mark.anyio +async def test_link_all_skips_total_fetch_without_progress_consumer() -> None: + # No on_progress: the denominator is never used, so the extra request that + # fetches it must be skipped. + client = FakeMatchingClient( + items=[ + FieldCandidate(id=1, name="Alpha", country="US"), + FieldCandidate(id=2, name="alpha", country="US"), + ], + details_by_id={ + 1: FieldDetailCandidate(id=1, name="Alpha", country="US"), + 2: FieldDetailCandidate(id=2, name="alpha", country="US"), + }, + ) + + await matching.link_all( + client, apply_merges=True, page_size=200, initiated_by="Tester" + ) + + assert client.total_calls == 0 diff --git a/deployments/stitch-frontend/src/pages/EntityLinkagePage.jsx b/deployments/stitch-frontend/src/pages/EntityLinkagePage.jsx index 89dfaa75..c23e32a7 100644 --- a/deployments/stitch-frontend/src/pages/EntityLinkagePage.jsx +++ b/deployments/stitch-frontend/src/pages/EntityLinkagePage.jsx @@ -18,6 +18,14 @@ function getMatchGroups(result) { return Array.isArray(result?.match_groups) ? result.match_groups : []; } +function formatTimestamp(value) { + if (!value) return null; + const date = new Date(value); + // Date + time, not time alone: a linkage pass can span more than a day, so a + // bare time would be ambiguous across midnight (STIT-740). + return Number.isNaN(date.getTime()) ? null : date.toLocaleString(); +} + function getResultDetails(result) { if (!result || typeof result !== "object" || Array.isArray(result)) { return result; @@ -78,6 +86,77 @@ function MatchGroupsSummary({ groups }) { ); } +function LinkProgressView({ progress }) { + const scanned = + typeof progress?.resources_scanned === "number" + ? progress.resources_scanned + : 0; + const total = + typeof progress?.total_resources === "number" + ? progress.total_resources + : null; + // The total is a start-of-run snapshot. Only scanned > total means it went + // stale (dataset grew); equality is normal near the end of a run, so keep it + // determinate and treat only genuine overshoot (and an unknown total) as + // indeterminate rather than showing a stuck/false "100%". + const hasReliableTotal = total !== null && total > 0 && scanned <= total; + // Floor, not round; and clamp to 99, since this bar only shows while running so + // it should never read a "done"-looking 100%. + const percent = hasReliableTotal + ? Math.min(99, Math.floor((scanned / total) * 100)) + : null; + const updatedAt = formatTimestamp(progress?.updated_at); + + return ( +
+
+

+ {hasReliableTotal + ? `Processing ${scanned.toLocaleString()} of ${total.toLocaleString()}` + : `Processing ${scanned.toLocaleString()}…`} +

+ {percent !== null && ( + + {percent}% + + )} +
+ +
+ {percent !== null ? ( +
+ ) : ( + // Total unknown: show an indeterminate bar rather than a false 0%. +
+ )} +
+ +

+ {formatCount(progress?.merge_candidates_created ?? 0, "candidate")}{" "} + created + {" · "} + {progress?.merge_candidates_skipped ?? 0} skipped + {" · "} + {progress?.resources_failed ?? 0} failed +

+ + {updatedAt && ( +

Last updated {updatedAt}

+ )} +
+ ); +} + function RunResult({ record }) { if (!record) { return ( @@ -117,9 +196,13 @@ function RunResult({ record }) { ) : record.state === "running" ? ( -

- Run in progress — status refreshes automatically. -

+ record.progress ? ( + + ) : ( +

+ Run in progress — status refreshes automatically. +

+ ) ) : record.state === "failed" ? (
{record.error || "Run failed."} @@ -142,7 +225,7 @@ export default function EntityLinkagePage() { const { getAccessTokenSilently } = useAuth0(); const baseUrl = config.entityLinkageBaseUrl; - const [applyMerges, setApplyMerges] = useState(false); + const [applyMerges, setApplyMerges] = useState(true); const [starting, setStarting] = useState(false); const [refreshing, setRefreshing] = useState(false); const [record, setRecord] = useState(null); @@ -276,17 +359,7 @@ export default function EntityLinkagePage() {
- - -
+
+ +
{error ? ( diff --git a/deployments/stitch-frontend/src/pages/EntityLinkagePage.test.jsx b/deployments/stitch-frontend/src/pages/EntityLinkagePage.test.jsx index 9f92d0dd..4e3cc512 100644 --- a/deployments/stitch-frontend/src/pages/EntityLinkagePage.test.jsx +++ b/deployments/stitch-frontend/src/pages/EntityLinkagePage.test.jsx @@ -27,6 +27,63 @@ const SUCCEEDED_RECORD = { }, }; +const RUNNING_WITH_PROGRESS_RECORD = { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + finished_at: null, + error: null, + result: null, + progress: { + resources_scanned: 120, + total_resources: 1000, + merge_candidates_created: 4, + merge_candidates_skipped: 1, + resources_failed: 0, + updated_at: "2026-06-11T10:02:00Z", + }, +}; + +const RUNNING_NEAR_COMPLETE_RECORD = { + ...RUNNING_WITH_PROGRESS_RECORD, + progress: { + ...RUNNING_WITH_PROGRESS_RECORD.progress, + // 9951/10000 = 99.51%: rounds to 100 but must floor to 99 while running. + resources_scanned: 9951, + total_resources: 10000, + }, +}; + +const RUNNING_AT_TOTAL_RECORD = { + ...RUNNING_WITH_PROGRESS_RECORD, + progress: { + ...RUNNING_WITH_PROGRESS_RECORD.progress, + // Final 100-boundary of a run whose size is a multiple of 100: scanned == total + // while still running. Stays determinate, clamped to 99% (not a false 100%). + resources_scanned: 3000, + total_resources: 3000, + }, +}; + +const RUNNING_OVER_TOTAL_RECORD = { + ...RUNNING_WITH_PROGRESS_RECORD, + progress: { + ...RUNNING_WITH_PROGRESS_RECORD.progress, + // Start-of-run total became stale as the dataset grew: scanned passed it. + resources_scanned: 3050, + total_resources: 3000, + }, +}; + +const RUNNING_WITHOUT_TOTAL_RECORD = { + ...RUNNING_WITH_PROGRESS_RECORD, + progress: { + ...RUNNING_WITH_PROGRESS_RECORD.progress, + resources_scanned: 120, + total_resources: null, + }, +}; + function jsonResponse(status, body) { return { ok: status >= 200 && status < 300, @@ -78,6 +135,35 @@ describe("EntityLinkagePage", () => { ); }); + it("defaults 'Initiate merges' on and posts apply_merges true", async () => { + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }), + ); + + renderWithQueryClient(); + + expect( + screen.getByRole("checkbox", { name: "Initiate merges" }), + ).toBeChecked(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + + await waitFor(() => { + expect(fetchMock).toHaveBeenCalledWith( + START_URL, + expect.objectContaining({ + method: "POST", + body: JSON.stringify({ apply_merges: true }), + }), + ); + }); + }); + it("renders match groups from the polled job result", async () => { const fetchMock = vi .spyOn(globalThis, "fetch") @@ -124,6 +210,171 @@ describe("EntityLinkagePage", () => { ); }); + it("renders a progress bar with counts while a run reports progress", async () => { + vi.spyOn(globalThis, "fetch").mockImplementation(async (url, options) => { + if (String(url) === START_URL && options?.method === "POST") { + return jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }); + } + return jsonResponse(200, RUNNING_WITH_PROGRESS_RECORD); + }); + + renderWithQueryClient(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + await userEvent.click( + screen.getByRole("button", { name: "Refresh status" }), + ); + + await waitFor(() => { + expect(screen.getByText(/Processing/)).toBeInTheDocument(); + }); + + // 120 of 1000 -> 12%. + const bar = screen.getByRole("progressbar", { + name: "Linkage run progress", + }); + expect(bar).toHaveAttribute("aria-valuenow", "12"); + expect(screen.getByText("12%")).toBeInTheDocument(); + expect(screen.getByText(/4 candidates created/)).toBeInTheDocument(); + expect(screen.getByText(/Last updated/)).toBeInTheDocument(); + }); + + it("does not reach 100% until the run is complete", async () => { + vi.spyOn(globalThis, "fetch").mockImplementation(async (url, options) => { + if (String(url) === START_URL && options?.method === "POST") { + return jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }); + } + return jsonResponse(200, RUNNING_NEAR_COMPLETE_RECORD); + }); + + renderWithQueryClient(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + await userEvent.click( + screen.getByRole("button", { name: "Refresh status" }), + ); + + await waitFor(() => { + expect(screen.getByText(/Processing/)).toBeInTheDocument(); + }); + + // 9951/10000 floors to 99, not 100, while still running. + const bar = screen.getByRole("progressbar", { + name: "Linkage run progress", + }); + expect(bar).toHaveAttribute("aria-valuenow", "99"); + expect(screen.getByText("99%")).toBeInTheDocument(); + expect(screen.queryByText("100%")).not.toBeInTheDocument(); + }); + + it("stays determinate at 99% when scanned equals total (still running)", async () => { + vi.spyOn(globalThis, "fetch").mockImplementation(async (url, options) => { + if (String(url) === START_URL && options?.method === "POST") { + return jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }); + } + return jsonResponse(200, RUNNING_AT_TOTAL_RECORD); + }); + + renderWithQueryClient(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + await userEvent.click( + screen.getByRole("button", { name: "Refresh status" }), + ); + + await waitFor(() => { + expect(screen.getByText(/Processing 3,000 of 3,000/)).toBeInTheDocument(); + }); + + // scanned == total while running: determinate, clamped to 99%, never 100%. + const bar = screen.getByRole("progressbar", { + name: "Linkage run progress", + }); + expect(bar).toHaveAttribute("aria-valuenow", "99"); + expect(screen.getByText("99%")).toBeInTheDocument(); + expect(screen.queryByText("100%")).not.toBeInTheDocument(); + }); + + it("stays indeterminate when scanned passes a stale total (no stuck 100%)", async () => { + vi.spyOn(globalThis, "fetch").mockImplementation(async (url, options) => { + if (String(url) === START_URL && options?.method === "POST") { + return jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }); + } + return jsonResponse(200, RUNNING_OVER_TOTAL_RECORD); + }); + + renderWithQueryClient(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + await userEvent.click( + screen.getByRole("button", { name: "Refresh status" }), + ); + + await waitFor(() => { + expect(screen.getByText(/Processing 3,050/)).toBeInTheDocument(); + }); + + // scanned > total: no percent, no aria-valuenow, and no misleading "of". + const bar = screen.getByRole("progressbar", { + name: "Linkage run progress", + }); + expect(bar).not.toHaveAttribute("aria-valuenow"); + expect(screen.queryByText(/%/)).not.toBeInTheDocument(); + expect(screen.queryByText(/of 3,000/)).not.toBeInTheDocument(); + }); + + it("renders an indeterminate bar when the total is unknown", async () => { + vi.spyOn(globalThis, "fetch").mockImplementation(async (url, options) => { + if (String(url) === START_URL && options?.method === "POST") { + return jsonResponse(202, { + job_id: "job-1", + state: "running", + started_at: "2026-06-11T10:00:00Z", + initiated_by: "Test User", + }); + } + return jsonResponse(200, RUNNING_WITHOUT_TOTAL_RECORD); + }); + + renderWithQueryClient(); + + await userEvent.click(screen.getByRole("button", { name: "Start run" })); + await userEvent.click( + screen.getByRole("button", { name: "Refresh status" }), + ); + + await waitFor(() => { + expect(screen.getByText(/Processing 120/)).toBeInTheDocument(); + }); + + // Unknown total -> no percent and no aria-valuenow (indeterminate bar). + const bar = screen.getByRole("progressbar", { + name: "Linkage run progress", + }); + expect(bar).not.toHaveAttribute("aria-valuenow"); + expect(screen.queryByText(/%/)).not.toBeInTheDocument(); + }); + it("surfaces a friendly message when a run is already in progress (409)", async () => { vi.spyOn(globalThis, "fetch").mockResolvedValue( jsonResponse(409, { detail: "A job is already running: job-1" }),