Merge remote-tracking branch 'origin/main' into fix/test-reset-stall

"Tag opens the dialog without selecting the card" read the panel while it
still said "Loading file details…", then compared it with the same card's
finished details. On a slow WebKit runner the read raced the load.

TESTING.md

@@ -50,6 +50,13 @@ stdin open and the server (`--exit-on-stdin-eof`) shuts down cleanly, removing i
 temp root, when that pipe closes, so a worker killed without teardown leaves no
 server behind. A test that fails gets the stderr its server wrote during the test
 attached to the report as `server-stderr` (an aiohttp traceback behind a 500).
+A `/__test__/.../reset` still running after 2 s writes a report to the job log,
+passing test or not: every thread's stack taken while it waits, the default
+executor's threads and queue, the open connections and WAL size, then the total
+and how long each executor call queued and ran. Its lines start with
+`pm-slow-reset| ` (`tests/browser/slow_reset_watch.py`). The report never reads a
+connection another thread is using: one that is busy inside SQLite would block
+that read with the GIL held and freeze the whole server.
 
 The coverage run can be split the way CI splits it, and each shard keeps the raw
 V8 files it writes under `coverage/js/raw`:

tests/browser/concept-inventory.mjs

@@ -78,6 +78,17 @@ export async function seedConcept(request,data={}) {
     return response.json();
 }
 
+/**
+ * Seed the concept fixture and fix the browser's clock at the fixture's own instant (`manifest.now`), so the page and
+ * the fixture read one "now": a prompt used now reads "Used today" and the caption names the fixture's day even when
+ * the run crosses midnight between seeding and rendering. Timers keep running; only `Date` stands still.
+ */
+export async function seedConceptAt(page,request,data={}) {
+    const manifest=await seedConcept(request,data);
+    await page.clock.setFixedTime(new Date(manifest.now*1000));
+    return manifest;
+}
+
 /**
  * Save `tags` as the current image tag set of the listed file at `path` (root 0) through the real observe and tag
  * routes (#249 PR C), releasing the capture afterwards; returns the receipt's saved names.

tests/browser/images-concept.spec.mjs

@@ -1,7 +1,7 @@
 import {writeFile} from 'node:fs/promises';
 import {test,expect} from './fixtures.mjs';
 import {captureCanonicalState} from './history-canonical-comparison.mjs';
-import {SIZES,THEMES,MOTIONS,seedConcept,openConcept,openProduction,inventoryOf,diffInventory,mutate,stripExact,countCopy,reflow,withoutItems} from './concept-inventory.mjs';
+import {SIZES,THEMES,MOTIONS,seedConceptAt,openConcept,openProduction,inventoryOf,diffInventory,mutate,stripExact,countCopy,reflow,withoutItems} from './concept-inventory.mjs';
 
 // Task C7 of #249: the production Images page against the immutable concept, on the concept-shaped real fixture (24
 // images of the concept's six prompts, its seeds, sizes and favorites). Heading, toolbar, filter row, caption, grid,
@@ -85,7 +85,7 @@ function listInfo(card,actionsWidth) {
 /** Recorded Images deviations only (spec §8); everything else in the concept inventory must equal production. */
 const nocase=(a,b)=>{const x=a.toLowerCase(),y=b.toLowerCase();return x<y?-1:x>y?1:a<b?-1:a>b?1:0;};
 
-export function mapImagesConcept(reference,{manifest,now=new Date()}) {
+export function mapImagesConcept(reference,{manifest,now=new Date(manifest.now*1000)}) {
     const paths=new Map(manifest.images.map(image=>[image.filename,image.path]));
     // D14: the sort appends production's two sorts. Save view is the concept's own (saved views, #249 D16 retired).
     const toolbar={...reference.toolbar};
@@ -318,7 +318,7 @@ async function behaviorsOf(page,scope) {
     return {hover,opened,viewerTitle,focusAfterViewer,selected,layouts,search,single};
 }
 
-test.beforeEach(async({request},testInfo)=>{testInfo.manifest=await seedConcept(request);});
+test.beforeEach(async({page,request},testInfo)=>{testInfo.manifest=await seedConceptAt(page,request);});
 
 for(const [width,height] of SIZES)for(const theme of THEMES)test(`images page geometry, tokens, controls and behaviors match the concept ${width}x${height} ${theme}`,async({page,request},testInfo)=>{
     test.setTimeout(240_000);
@@ -348,7 +348,7 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`images page ge
             importDiff:diffInventory(mapImportConcept(referenceImport),actualImport),referenceBehaviors,actualBehaviors};
     }
     // Load more in production: a fixture past one listing page appends one batch and shows the concept's toast.
-    testInfo.manifest=await seedConcept(request,{extra_images:LOAD_MORE_EXTRA});
+    testInfo.manifest=await seedConceptAt(page,request,{extra_images:LOAD_MORE_EXTRA});
     for(const motion of MOTIONS) {
         const total=FIXTURE_IMAGES+LOAD_MORE_EXTRA;
         await openProductionLoaded(page,theme,motion,total);
@@ -514,7 +514,7 @@ test('one appended image reads in the singular',async({page,request})=>{
     test.setTimeout(120_000);
     await page.setViewportSize({width:1600,height:1100});
     const total=PRODUCTION_BATCH+1;
-    await seedConcept(request,{extra_images:total-FIXTURE_IMAGES});
+    await seedConceptAt(page,request,{extra_images:total-FIXTURE_IMAGES});
     await openProductionLoaded(page,'dark','no-preference',total);
     await expect(page.locator('.infinite-continuation > p').first()).toHaveText(countCopy(`${PRODUCTION_BATCH} of ${total} images · more load as you scroll`));
     // A dispatched press: scrolling the button into view lets the page's own observer load the one image first.
@@ -530,7 +530,7 @@ test('one image appended by scrolling reads in the singular',async({page,request
     test.setTimeout(120_000);
     await page.setViewportSize({width:1600,height:1100});
     const total=PRODUCTION_BATCH+1;
-    await seedConcept(request,{extra_images:total-FIXTURE_IMAGES});
+    await seedConceptAt(page,request,{extra_images:total-FIXTURE_IMAGES});
     await openProductionLoaded(page,'dark','no-preference',total);
     await expect(page.locator('#result-count')).toHaveText(new RegExp(`^${PRODUCTION_BATCH} of ${total} images`));
     await page.locator('.infinite-continuation').scrollIntoViewIfNeeded();
@@ -551,7 +551,7 @@ test('one footer height for the collection (D3)',async({page,request})=>{
     const reference=await inventoryOf(page,'#main');
     expect(reference.grid.columns).toBe(2);
     expect(reference.cards.every(card=>card.info.p.height>parseFloat(card.info.p.lineHeight))).toBe(true);
-    const manifest=await seedConcept(request,{seedless_images:2,short_seedless:true});
+    const manifest=await seedConceptAt(page,request,{seedless_images:2,short_seedless:true});
     await openProductionLoaded(page,'dark','no-preference',FIXTURE_IMAGES+2);
     const actual=await inventoryOf(page,'#workspace-main');
     const box=card=>({left:card.left,top:card.top,width:card.width,height:card.height});

tests/browser/library-concept.spec.mjs

@@ -1,7 +1,7 @@
 import {writeFile} from 'node:fs/promises';
 import {test,expect} from './fixtures.mjs';
 import {captureCanonicalState} from './history-canonical-comparison.mjs';
-import {SIZES,THEMES,MOTIONS,seedConcept,openConcept,openProduction,inventoryOf,diffInventory,mutate,stripExact,countCopy,reflow} from './concept-inventory.mjs';
+import {SIZES,THEMES,MOTIONS,seedConceptAt,openConcept,openProduction,inventoryOf,diffInventory,mutate,stripExact,countCopy,reflow} from './concept-inventory.mjs';
 
 // Task E3 of #249: the production Prompt library shelf against the immutable concept (app.js renderLibrary a572–586,
 // style.css s648–708), on the concept-shaped real fixture (the concept's six prompts with their negatives, categories,
@@ -37,7 +37,8 @@ export function mapLibraryConcept(reference) {
         // D43: tag chips come in the store's alphabetical order; the row flows again from the concept's own chips.
         tags:sortedTags(card.tags),
         // D23: the stars are aria-hidden, with a visually hidden "Rated N out of 5" spoken in their place. The fixture's
-        // run_count (4) and last_used_at (now) give the concept's own "4 runs" and "Used today". The two-line clamp is
+        // run_count (4) and last_used_at (its own "now", where the page's clock stands) give the concept's own "4 runs"
+        // and "Used today". The two-line clamp is
         // applied below through `clampedCards`, measured on the concept with `CLAMP_CSS` (its samples wrap at 390px).
         footer:{...card.footer,facts:{...card.footer.facts,spoken:card.footer.facts.spoken.replace(/^(★+)/,stars=>`Rated ${stars.length} out of 5`)}}}));
     return {...reference,library:{...library,toolbar,
@@ -238,7 +239,8 @@ async function openHost(context) {
 }
 
 test.beforeEach(async({page,request},testInfo)=>{
-    testInfo.manifest=await seedConcept(request);
+    // The page's clock stands at the fixture's own "now", so its last use reads "Used today" on either side of midnight.
+    testInfo.manifest=await seedConceptAt(page,request);
     // Both documents copy through the clipboard; the stub records what they wrote.
     await page.addInitScript(()=>{window.copied=[];Object.defineProperty(navigator,'clipboard',{configurable:true,value:{writeText:async text=>{window.copied.push(text);}}});});
 });
@@ -322,6 +324,27 @@ for(const theme of THEMES)test(`library grid and cards match the concept 1800x11
     expect(compare(reference,actual,{sections:['library'],library:['grid','cards','trailing']})).toEqual([]);
 });
 
+// PR #284's CI run: the fixture recorded the last use at 23:59:5x UTC and the shelf rendered at 00:00:05, so production
+// read "Used yesterday" against the concept's "Used today". Here the browser's wall clock starts at 23:59:58 UTC, the
+// fixture records its use at that instant, and midnight passes before either document opens.
+const BEFORE_MIDNIGHT=Date.UTC(2026,9,22,23,59,58);
+const PAST_MIDNIGHT_MS=5_000;
+test('a run that crosses midnight still reads the concept fixture\'s day',async({page,request})=>{
+    await page.setViewportSize({width:1800,height:1100});
+    await page.clock.install({time:BEFORE_MIDNIGHT});
+    const manifest=await seedConceptAt(page,request,{now:BEFORE_MIDNIGHT/1000});
+    expect(manifest.now*1000,'the fixture\'s now').toBe(BEFORE_MIDNIGHT);
+    const used=(await (await request.get('/prompt_manager/library')).json()).items.map(item=>Date.parse(item.last_used_at));
+    expect(used,'every last use at the fixture\'s now').toEqual(Array(CONCEPT_PROMPTS).fill(BEFORE_MIDNIGHT));
+    await page.clock.fastForward(PAST_MIDNIGHT_MS);
+    await openConceptLibrary(page,'dark','no-preference');
+    const reference=await conceptInventory(page);
+    await openProductionLibrary(page,'dark','no-preference');
+    const actual=await inventoryOf(page,'#workspace-main');
+    expect(actual.library.cards.map(card=>card.footer.facts.text.replace(/^★+ · /,''))).toEqual(Array(CONCEPT_PROMPTS).fill('Used today'));
+    expect(compare(reference,actual,{sections:['library'],library:['grid','cards','trailing']})).toEqual([]);
+});
+
 // The comparison is only worth something if it notices a change (spec §7 mutation check): one CSS mutation (the footer
 // padding) and one DOM mutation (the negative-variants chip removed), each applied after an exact match.
 test('library comparison detects mutations',async({page})=>{

tests/browser/server.py

@@ -19,10 +19,12 @@ from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
 from tests.browser.request_hold import register_request_holds
+from tests.browser import slow_reset_watch
 from tests.browser.workspace_activity_fixture import register_activity_commands
 from tests.browser.workspace_concept_fixture import (
     MAX_EXTRA_IMAGES as CONCEPT_MAX_EXTRA_IMAGES,
     MAX_SEEDLESS_IMAGES as CONCEPT_MAX_SEEDLESS_IMAGES,
+    MAX_EPOCH_SECONDS as CONCEPT_MAX_EPOCH_SECONDS,
     seed_workspace_concept,
 )
 from tests.browser.workspace_visual_fixture import seed_workspace_visual
@@ -197,6 +199,7 @@ def create_app(root: Path) -> web.Application:
     reveal_stub = register_reveal_stub(routes, api, root)
     register_activity_commands(app, routes, api)
     request_holds = register_request_holds(app, routes)
+    slow_reset_watch.install(app, api)
 
     async def canonical_asset(request: web.Request) -> web.Response:
         filename, content_type = CANONICAL_ASSETS[request.path]
@@ -489,8 +492,17 @@ def create_app(root: Path) -> web.Application:
         if (
             request.query
             or not isinstance(data, dict)
-            or set(data) - set(limits) - {"short_seedless"}
+            or set(data) - set(limits) - {"short_seedless", "now"}
             or not isinstance(data.get("short_seedless", False), bool)
+            # A chosen "now" (epoch seconds) pins every time the fixture writes.
+            or (
+                "now" in data
+                and (
+                    isinstance(data["now"], bool)
+                    or not isinstance(data["now"], (int, float))
+                    or not 0 < data["now"] < CONCEPT_MAX_EPOCH_SECONDS
+                )
+            )
             or any(
                 isinstance(data.get(key, 0), bool)
                 or not isinstance(data.get(key, 0), int)
@@ -510,6 +522,7 @@ def create_app(root: Path) -> web.Application:
                 extra_images=data.get("extra_images", 0),
                 seedless_images=data.get("seedless_images", 0),
                 short_seedless=data.get("short_seedless", False),
+                now=data.get("now"),
             )
         )
         return web.json_response({"success": True, **manifest})

tests/browser/slow_reset_watch.py

@@ -0,0 +1,237 @@
+"""Report a slow test reset with what the server was doing while it waited.
+
+A reset normally answers within about a second. Once one runs past
+SLOW_RESET_SECONDS, this writes to stderr, while the reset is still waiting:
+
+- every thread's stack, so a reset waiting on a lock, a full executor or a
+  slow disk shows where it waits and who holds it;
+- the default executor's threads and queued work;
+- how many database connections are open, and the WAL size;
+- on Linux, each thread's scheduler state and the kernel function it sleeps
+  in (``D`` with a journal or writeback wait is a slow disk flush, a nanosleep
+  is SQLite's busy handler, a futex is a Python lock or an idle worker).
+
+When the reset ends it writes how long it took and, for each executor call it
+made, how long the call queued and how long it ran.
+
+Every report line starts with REPORT_PREFIX. The worker fixture
+(tools/vendor/start-browser-server.mjs) copies those lines to the job log,
+so a slow reset is visible even in a test that still passes.
+"""
+
+import asyncio
+import contextvars
+import faulthandler
+import os
+import sys
+import tempfile
+import threading
+import time
+from typing import Callable, TextIO
+
+from aiohttp import web
+
+SLOW_RESET_SECONDS = 2.0
+REPORT_PREFIX = "pm-slow-reset| "
+
+_current: contextvars.ContextVar["ResetWatch | None"] = contextvars.ContextVar(
+    "pm_reset_watch", default=None
+)
+
+
+def is_reset_path(path: str) -> bool:
+    return path.startswith("/__test__/") and path.endswith("/reset")
+
+
+class ResetWatch:
+    """Time one reset; dump the server's state if it passes the threshold."""
+
+    def __init__(
+        self,
+        label: str,
+        describe: Callable[[], list[str]],
+        *,
+        threshold: float = SLOW_RESET_SECONDS,
+        stream: TextIO | None = None,
+    ) -> None:
+        self.label = label
+        self.threshold = threshold
+        self._describe = describe
+        self._stream = stream
+        self._start = time.monotonic()
+        self._calls: list[str] = []
+        self._calls_lock = threading.Lock()
+        self._stalled = threading.Event()
+        self._timer = threading.Timer(threshold, self._report_stall)
+        self._timer.daemon = True
+        self._timer.name = "pm-slow-reset-watch"
+
+    def start(self) -> "ResetWatch":
+        self._timer.start()
+        return self
+
+    def elapsed(self) -> float:
+        return time.monotonic() - self._start
+
+    def record_call(self, name: str, queued: float, ran: float) -> None:
+        with self._calls_lock:
+            self._calls.append(
+                f"executor call {name}: queued {queued:.3f}s, ran {ran:.3f}s"
+            )
+
+    def finish(self, outcome: str) -> None:
+        self._timer.cancel()
+        elapsed = self.elapsed()
+        # A coarse clock (Windows) can fire the timer just short of the
+        # threshold; a reset that was reported always gets its summary.
+        if elapsed < self.threshold and not self._stalled.is_set():
+            return
+        with self._calls_lock:
+            calls = list(self._calls)
+        self._write([f"{self.label} {outcome} after {elapsed:.3f}s", *calls])
+
+    def _report_stall(self) -> None:
+        self._stalled.set()
+        lines = [f"{self.label} still running after {self.elapsed():.3f}s"]
+        try:
+            lines.extend(self._describe())
+        except Exception as error:  # a diagnostic never breaks the reset
+            lines.append(f"state unavailable: {error!r}")
+        lines.extend(thread_stacks())
+        self._write(lines)
+
+    def _write(self, lines: list[str]) -> None:
+        stream = self._stream or sys.stderr
+        text = "".join(
+            f"{REPORT_PREFIX}{part}\n" for line in lines for part in line.splitlines()
+        )
+        try:
+            stream.write(text)
+            stream.flush()
+        except (OSError, ValueError):
+            pass
+
+
+def thread_stacks() -> list[str]:
+    """Every thread's stack, from faulthandler.
+
+    Formatting sys._current_frames() showed a thread blocked in SQLite as idle
+    (Python 3.14); faulthandler reports where each thread really is.
+    """
+    # Before Python 3.14 faulthandler prints ids only; name each id here.
+    names = [
+        f"thread 0x{thread.ident:016x} is {thread.name}"
+        for thread in threading.enumerate()
+        if thread.ident is not None
+    ]
+    with tempfile.TemporaryFile() as dump:
+        faulthandler.dump_traceback(file=dump, all_threads=True)
+        dump.seek(0)
+        return [*names, *dump.read().decode("utf-8", "replace").splitlines()]
+
+
+def executor_state(loop: asyncio.AbstractEventLoop) -> list[str]:
+    executor = getattr(loop, "_default_executor", None)
+    if executor is None:
+        return ["default executor: not started"]
+    threads = getattr(executor, "_threads", ())
+    queue = getattr(executor, "_work_queue", None)
+    queued = queue.qsize() if queue is not None else "?"
+    return [
+        f"default executor: {len(threads)} of "
+        f"{getattr(executor, '_max_workers', '?')} threads, {queued} queued"
+    ]
+
+
+def database_state(model) -> list[str]:
+    """How many connections are open, and the WAL size.
+
+    Never touches another thread's connection: a connection busy in SQLite
+    holds its mutex, and reading even ``in_transaction`` from here would block
+    on it while holding the GIL, freezing every thread until that query ends.
+    """
+    with model._conn_lock:
+        count = len(model._connections)
+    wal = model.db_path + "-wal"
+    size = os.path.getsize(wal) if os.path.exists(wal) else 0
+    return [f"database: {count} connections, WAL {size} bytes"]
+
+
+def kernel_wait_state() -> list[str]:
+    """Each thread's state and kernel wait channel, where /proc has them."""
+    lines = []
+    for thread in threading.enumerate():
+        task = f"/proc/self/task/{thread.native_id}"
+        try:
+            with open(f"{task}/stat", encoding="ascii") as stat:
+                state = stat.read().rsplit(")", 1)[1].split()[0]
+            with open(f"{task}/wchan", encoding="ascii") as wchan:
+                channel = wchan.read().strip() or "0"
+        except (OSError, IndexError):
+            continue
+        # The id matches the "Thread 0x..." headers of the stacks below.
+        lines.append(
+            f"kernel: {thread.name} (0x{thread.ident:016x}) state {state}, "
+            f"waiting in {channel}"
+        )
+    return lines
+
+
+def install(app: web.Application, api) -> None:
+    """Watch every reset route, and time the executor calls each one makes."""
+    run_in_executor = api._run_in_executor
+
+    async def timed_run_in_executor(func, *args, **kwargs):
+        watch = _current.get()
+        if watch is None:
+            return await run_in_executor(func, *args, **kwargs)
+        queued_at = time.monotonic()
+        started: list[float] = []
+
+        def timed(*call_args, **call_kwargs):
+            started.append(time.monotonic())
+            return func(*call_args, **call_kwargs)
+
+        try:
+            return await run_in_executor(timed, *args, **kwargs)
+        finally:
+            end = time.monotonic()
+            begin = started[0] if started else end
+            watch.record_call(
+                getattr(func, "__qualname__", repr(func)),
+                begin - queued_at,
+                end - begin,
+            )
+
+    api._run_in_executor = timed_run_in_executor
+
+    @web.middleware
+    async def watch_resets(request: web.Request, handler):
+        if not is_reset_path(request.path):
+            return await handler(request)
+        loop = asyncio.get_running_loop()
+
+        def describe() -> list[str]:
+            return [
+                *executor_state(loop),
+                *database_state(api.db.model),
+                *kernel_wait_state(),
+            ]
+
+        watch = ResetWatch(
+            f"{request.method} {request.path}", describe, threshold=SLOW_RESET_SECONDS
+        ).start()
+        token = _current.set(watch)
+        outcome = "failed"
+        try:
+            response = await handler(request)
+            outcome = f"answered {response.status}"
+            return response
+        except web.HTTPException as error:
+            outcome = f"answered {error.status}"
+            raise
+        finally:
+            _current.reset(token)
+            watch.finish(outcome)
+
+    app.middlewares.append(watch_resets)

tests/browser/workspace-activity.spec.mjs

@@ -2,6 +2,13 @@ import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage
 import { test, expect } from './fixtures.mjs';
 import { seedConcept, openConcept, openProduction } from './concept-inventory.mjs';
 
+// The Activity store's periodic list read (activity-store.mjs POLL_MS) and the detail view's read throttle
+// (activity-job-view.mjs DETAIL_INTERVAL_MS).
+const POLL_MS = 15_000;
+const DETAIL_INTERVAL_MS = 1_000;
+// A request is reported to the test a little after the browser sends it; that delay is not a shorter wait.
+const REPORT_SLACK_MS = 100;
+
 // Activity (#249 Activity-1): the concept's Activity page (app.js renderActivity 732–738), its top-bar button
 // (index.html 51–52) and sidebar row (app.js ~341), with the recorded D52 (counts of running jobs), D53 (the progress
 // pill) and D54 (real job items). Fake jobs come from tests/browser/workspace_activity_fixture.py.
@@ -278,6 +285,8 @@ test('reload during a scan reattaches to the running job', async ({ page, reques
 });
 
 test('lost server clears the pill and marks jobs ended', async ({ page, request }) => {
+    // A poll interval after the restart, then failed re-reads backing off for 1 + 2 + 4 s.
+    test.setTimeout(90_000);
     await page.setViewportSize({ width: 1600, height: 1100 });
     const job = await startJob(request);
     await step(request, job.id, { processed: 4000 });
@@ -294,9 +303,12 @@ test('lost server clears the pill and marks jobs ended', async ({ page, request
     await expect(items(page).first().locator('p')).toHaveText(LOST);
     await expect(items(page).first().locator('progress, .row-actions')).toHaveCount(0);
     await expect(navRow(page).locator('.nav-count')).toHaveCount(0);
-    // One re-read found the restart; there is no retry storm and no finish toast for a job that only vanished.
-    await page.waitForTimeout(2500);
-    expect(reads.length).toBeLessThanOrEqual(2);
+    // One re-read found the restart; there is no retry storm and no finish toast for a job that only vanished. The
+    // read after it is the periodic one, a whole poll interval later: any extra read lands in that gap and fails it.
+    const found = reads.length;
+    expect(found, 'reads that found the restart').toBeGreaterThanOrEqual(1);
+    await expect.poll(() => reads.length, { timeout: POLL_MS + 10_000 }).toBeGreaterThan(found);
+    expect(reads[found] - reads[found - 1], 'the next read waits for the poll').toBeGreaterThanOrEqual(POLL_MS - REPORT_SLACK_MS);
     await expect(toast(page)).toHaveText('');
 
     // A server that cannot be read at all also ends the running job, then is read again with a growing backoff.
@@ -308,9 +320,12 @@ test('lost server clears the pill and marks jobs ended', async ({ page, request
     await expect(await request.post('/__test__/jobs/restart')).toBeOK();
     await expect(pill(page)).toBeHidden();
     await expect(items(page).filter({ hasText: LOST })).toHaveCount(1);
-    await page.waitForTimeout(3500);
-    // 1 s, then 2 s: at most two re-reads in 3.5 s.
-    expect(failed.length).toBeLessThanOrEqual(2);
+    // 1 s, then 2 s, then 4 s: each failed re-read waits twice as long as the one before. The gaps between the
+    // reads themselves are the evidence, so a slow runner noticing the ended job late cannot add a read to a window.
+    await expect.poll(() => failed.length, { timeout: 20_000 }).toBeGreaterThanOrEqual(3);
+    const gaps = failed.slice(1, 3).map((at, index) => at - failed[index]);
+    expect(gaps[0]).toBeGreaterThanOrEqual(2000 - REPORT_SLACK_MS);
+    expect(gaps[1]).toBeGreaterThanOrEqual(4000 - REPORT_SLACK_MS);
     await page.unroute('**/prompt_manager/jobs**');
     expect(second.id).not.toBe(job.id);
 });
@@ -607,6 +622,8 @@ test('filter narrows the selected outcome', async ({ page, request }) => {
 
 test('job view renders 200-row cap note', async ({ page, request }) => {
     await page.setViewportSize({ width: 1600, height: 1100 });
+    // The page's timers are Playwright's, running at real speed, so the M12 check below can run a read throttle out.
+    await page.clock.install();
     const job = await startJob(request);
     await step(request, job.id, { processed: 260, bulk: { group: 'added', count: 250 }, records: [
         ...Array.from({ length: 10 }, (_, index) => ({ group: 'no_metadata', path: `misc/plain_${index}.png`, detail: 'No prompt metadata in the file' })),
@@ -646,11 +663,15 @@ test('job view renders 200-row cap note', async ({ page, request }) => {
     page.on('request', item => { if (new URL(item.url()).pathname === `/prompt_manager/jobs/${job.id}`) reads.push(Date.now()); });
     await step(request, job.id, { processed: 300, bulk: { group: 'added', count: 40 } });
     await expect(tiles(page).first().locator('dd')).toHaveText('290');
-    await page.waitForTimeout(1500);
+    // Every read the counter update could have scheduled is due within one throttle interval: run it out, so any such
+    // read is sent before the log opens. Requests reach the test in the order they were sent, so once the log's read
+    // has answered, a read sent before it would already be listed.
+    await page.clock.runFor(DETAIL_INTERVAL_MS);
     expect(reads).toEqual([]);
     await scanLog(page).locator('summary').click();
     await expect.poll(() => reads.length).toBe(1);
     await expect(scanLog(page).locator('pre')).toContainText('bulk/file_00039.png');
+    expect(reads, 'only the log\'s own read').toHaveLength(1);
 });
 
 test('cancel scan ends cancelled with partial counts', async ({ page, request }) => {
@@ -705,14 +726,17 @@ test('a held legacy job shows no Cancel in the list or its view', async ({ page,
 
 test('busy start names the running job', async ({ page, request }) => {
     await page.setViewportSize({ width: 1600, height: 1100 });
+    // The page's timers are Playwright's, so the toast's lifetime below is run out rather than waited for.
+    await page.clock.install();
     const running = await startJob(request, { title: 'Database maintenance', kind: 'maintenance' });
     await startFromImport(page);
     await expect(toast(page)).toHaveText('Another job is running: Database maintenance.');
     await expect(toast(page).getByRole('link', { name: 'Database maintenance', exact: true })).toHaveAttribute('href', `#/activity/${running.id}`);
     await expect(page).toHaveURL(/#\/images$/);
     expect((await listed(request)).map(item => item.id)).toEqual([running.id]);
-    // Review M14: a toast with a link outlasts the plain 6.5 s, so the link stays reachable.
-    await page.waitForTimeout(7000);
+    // Review M14: a toast with a link outlasts the plain 6.5 s, so the link stays reachable. Running the page's timers
+    // 7 s on fires a 6.5 s clear for certain, however slowly this runner reached this line.
+    await page.clock.runFor(7000);
     await expect(toast(page).getByRole('link', { name: 'Database maintenance', exact: true })).toBeVisible();
     await toast(page).getByRole('link', { name: 'Database maintenance', exact: true }).click();
     await expect(heading(page).locator('h1')).toHaveText('Database maintenance');

tests/browser/workspace-images.spec.mjs

@@ -907,6 +907,10 @@ test('browser Back closes the open Images dialog and leaves the next page usable
 
 test('count line shows images and positive prompts', async ({ page, request }) => {
     await page.addInitScript(() => { window.IntersectionObserver = undefined; });
+    // D17's "Today" is the newest file's day: the page's clock stands at that file's mtime, so the fixture's day and
+    // the page's agree on either side of midnight.
+    const { images } = await (await request.get('/prompt_manager/images/output?mode=workspace&limit=40')).json();
+    await page.clock.setFixedTime(new Date(Math.max(...images.map(image => image.modified_time)) * 1000));
     await page.goto(workspace);
     await expect(cards(page)).toHaveCount(2);
     await expect(caption(page).locator('#result-count')).toHaveText('2 of 2 images · 2 positive prompts');

tests/browser/workspace-library.spec.mjs

@@ -610,7 +610,9 @@ test('card matches the concept and never wraps the category mid-word',async({pag
     await expect(await request.post('/prompt_manager/bulk/category',{data:{prompt_ids:[dual.id],category:''}})).toBeOK();
     // A long saved title fills the title row, so only the chip's own min-content keeps "Uncategorized" and "asian" whole.
     const wide='A remarkably long saved positive prompt title that fills the whole card title row';
-    await patchShelf(page,item=>item.id===cat.id?{run_count:4,last_used_at:new Date().toISOString(),rating:5}:item.id===dual.id||item.id===dog.id?{title:wide,run_count:1,last_used_at:null,rating:null}:{});
+    // One "now" for the patched last use and the page's clock, so it reads "Used today" on either side of midnight.
+    const usedAt=new Date();await page.clock.setFixedTime(usedAt);
+    await patchShelf(page,item=>item.id===cat.id?{run_count:4,last_used_at:usedAt.toISOString(),rating:5}:item.id===dual.id||item.id===dog.id?{title:wide,run_count:1,last_used_at:null,rating:null}:{});
     for(const [width,height] of [[1600,1100],[390,844]]) {
         await page.setViewportSize({width,height});await page.goto(route);await expect(cards(page)).toHaveCount(3);
         const card=page.locator(`article[data-prompt-id="${cat.id}"]`);

tests/browser/workspace-tagging.spec.mjs

@@ -340,6 +340,9 @@ test('Tag opens the dialog without selecting the card', async ({ page }) => {
     const panel = page.getByRole('complementary', { name: 'Selected image details', exact: true });
     await page.goto(images);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
+    // Capture the panel once it shows the first card's details, not while it is still reading them.
+    await expect(panel.getByRole('heading', { level: 2 })).toBeVisible();
+    await expect(panel).not.toContainText('Loading file details');
     const before = await panel.innerText();
     await openTag(page, SECOND, { goto: false });
     await page.keyboard.press('Escape');

tests/browser/workspace_concept_fixture.py

@@ -11,6 +11,7 @@ the same sidebar counts (24 images, 6 prompts, 15 tags). Nothing is mocked.
 from __future__ import annotations
 
 from dataclasses import dataclass
+import datetime
 import json
 import math
 import os
@@ -157,6 +158,8 @@ CONCEPT_MODEL = "illustrative-model.safetensors"
 MAX_EXTRA_IMAGES = 120
 # Unlinked, seedless images (D11 "W × H · PNG") newer than the concept's, for D3 checks.
 MAX_SEEDLESS_IMAGES = 2
+# A test's chosen "now" (epoch seconds) stays before the year 3000.
+MAX_EPOCH_SECONDS = 32_503_680_000
 SEEDLESS_SIZE = (1536, 1152)
 # The same 4:3 shape with a short "96 × 72 · PNG" subtitle that stays on one line at
 # 390px beside the card's Info and Tag (the D3 footer check).
@@ -342,6 +345,17 @@ def _bounded(name: str, value: int, maximum: int) -> None:
         raise ValueError(f"{name} is out of range")
 
 
+def _instant(now: float | None) -> float:
+    """The fixture's one "now": ``now`` itself, or the wall clock when omitted."""
+    if now is None:
+        return time.time()
+    if isinstance(now, bool) or not isinstance(now, (int, float)):
+        raise TypeError("now must be a number of epoch seconds")
+    if not 0 < now < MAX_EPOCH_SECONDS:
+        raise ValueError("now is out of range")
+    return float(now)
+
+
 def seed_workspace_concept(
     api: PromptManagerAPI,
     root: Path,
@@ -349,9 +363,16 @@ def seed_workspace_concept(
     extra_images: int = 0,
     seedless_images: int = 0,
     short_seedless: bool = False,
+    now: float | None = None,
 ) -> dict:
     """Replace this server's disposable rows and files with the concept's data.
 
+    Every relative time the fixture writes (the prompts' last use, the files'
+    mtimes, the runs' observation times) is taken from one instant, ``now``
+    (epoch seconds, the wall clock when omitted), returned as ``manifest["now"]``.
+    A browser whose clock is fixed at that instant reads the concept's "Used
+    today" and caption day however close to midnight the run starts.
+
     ``extra_images`` continues the concept's rounds (same prompts, dimensions and
     seed rule) after its 24 images, for bounded-rendering and Load more checks.
     ``seedless_images`` adds unlinked PNGs without metadata, newest of all;
@@ -361,6 +382,10 @@ def seed_workspace_concept(
         raise TypeError("short_seedless must be a boolean")
     _bounded("extra_images", extra_images, MAX_EXTRA_IMAGES)
     _bounded("seedless_images", seedless_images, MAX_SEEDLESS_IMAGES)
+    now = _instant(now)
+    used_at = datetime.datetime.fromtimestamp(now, datetime.timezone.utc).isoformat(
+        timespec="microseconds"
+    )
     _clear(api, root)
     db = api.db
     epoch = db.get_execution_epoch()
@@ -385,8 +410,15 @@ def seed_workspace_concept(
         db.update_prompt_pair_metadata(pair_ids[0], title=first_name)
         for name, text in prompt.negatives[1:]:
             pair_ids.append(db.save_prompt_pair(prompt.positive, text, title=name))
-        # Real queue-hook counting (D23): four runs, the last one now.
+        # Real queue-hook counting (D23): four runs, the last one now, at the
+        # fixture's own instant rather than the database's later wall clock.
         db.record_prompt_use(positive_id, times=CONCEPT_RUN_COUNT)
+        with db.model.get_connection() as conn:
+            conn.execute(
+                "UPDATE prompts SET last_used_at = ? WHERE id = ?",
+                (used_at, positive_id),
+            )
+            conn.commit()
         prompts.insert(
             0, {"id": prompt.id, "positive_id": positive_id, "pair_ids": pair_ids}
         )
@@ -400,7 +432,6 @@ def seed_workspace_concept(
     favorite_root = _favorite_root_key(list(api._get_all_output_dirs())[0])
     total = len(CONCEPT_PROMPTS) * CONCEPT_IMAGE_ROUNDS + extra_images
     rounds = math.ceil(total / len(CONCEPT_PROMPTS))
-    now = time.time()
     images, runs = [], []
     slots = [
         (round_index, prompt_index, prompt)
@@ -484,6 +515,8 @@ def seed_workspace_concept(
     api.invalidate_gallery_cache()
     return {
         "epoch": epoch,
+        # The fixture's "now" (epoch seconds), for the browser's fixed clock.
+        "now": now,
         # The newest file's mtime (epoch seconds), for the D17 caption date.
         "newest_modified": now - 10 if seedless_images else now - 60,
         "seedless": seedless,

tests/test_browser_slow_reset_watch.py

@@ -0,0 +1,234 @@
+"""A slow page-workflow reset reports where the server waited.
+
+CI run 39904288262 had one ``/__test__/reset`` wait 16 s with nothing in the
+server's stderr. tests/browser/slow_reset_watch.py reports any reset past
+SLOW_RESET_SECONDS with every thread's stack, taken while it still waits.
+These tests hold a real SQLite write lock behind a real test server's reset.
+"""
+
+import glob
+import io
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import urllib.request
+from pathlib import Path
+
+from tests.browser import slow_reset_watch
+from tests.browser.slow_reset_watch import REPORT_PREFIX, ResetWatch, is_reset_path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SERVER_TIMEOUT_S = 60
+HOLD_SECONDS = slow_reset_watch.SLOW_RESET_SECONDS + 4
+# The reset may take a little longer than the hold, never the 30 s busy timeout.
+RESET_MARGIN_SECONDS = 10
+# While the reset waits, the server still answers other requests this fast.
+# Reading the waiting connection from the report's thread froze it (the GIL is
+# held while SQLite's connection mutex is awaited) until the lock was released.
+RESPONSIVE_SECONDS = 2.0
+PROBE_INTERVAL_SECONDS = 0.1
+
+
+def blocked_here(gate: threading.Event) -> None:
+    gate.wait()
+
+
+class TestResetWatch(unittest.TestCase):
+    def test_reset_paths_are_the_test_reset_routes(self):
+        self.assertTrue(is_reset_path("/__test__/reset"))
+        self.assertTrue(is_reset_path("/__test__/tagging/reset"))
+        self.assertFalse(is_reset_path("/prompt_manager/reset"))
+        self.assertFalse(is_reset_path("/__test__/reset/extra"))
+
+    def test_a_stalled_reset_reports_state_and_stacks_while_it_waits(self):
+        stream = io.StringIO()
+        reported = threading.Event()
+
+        class Stream(io.StringIO):
+            def flush(self):
+                reported.set()
+
+        stream = Stream()
+        gate = threading.Event()
+        worker = threading.Thread(target=blocked_here, args=(gate,), name="held-up")
+        worker.start()
+        self.addCleanup(worker.join)
+        self.addCleanup(gate.set)
+        watch = ResetWatch(
+            "POST /__test__/reset",
+            lambda: ["database: 1 connections"],
+            threshold=0.05,
+            stream=stream,
+        ).start()
+        self.assertTrue(reported.wait(SERVER_TIMEOUT_S))
+        report = stream.getvalue().splitlines()
+        self.assertTrue(all(line.startswith(REPORT_PREFIX) for line in report))
+        text = "\n".join(report)
+        self.assertIn("POST /__test__/reset still running after", text)
+        self.assertIn("database: 1 connections", text)
+        self.assertIn(f"thread 0x{worker.ident:016x} is held-up", text)
+        self.assertIn("in blocked_here", text)
+
+        watch.record_call("seed", 0.25, 1.5)
+        watch.finish("answered 200")
+        self.assertIn("POST /__test__/reset answered 200 after", stream.getvalue())
+        self.assertIn(
+            "executor call seed: queued 0.250s, ran 1.500s", stream.getvalue()
+        )
+
+    def test_a_fast_reset_writes_nothing_and_stops_its_timer(self):
+        stream = io.StringIO()
+        watch = ResetWatch("reset", lambda: [], threshold=60, stream=stream).start()
+        watch.finish("answered 200")
+        self.assertTrue(watch._timer.finished.is_set())
+        self.assertEqual(stream.getvalue(), "")
+
+    def test_a_failing_state_probe_still_reports_the_stacks(self):
+        def broken():
+            raise RuntimeError("no state")
+
+        reported = threading.Event()
+
+        class Stream(io.StringIO):
+            def flush(self):
+                reported.set()
+
+        stream = Stream()
+        ResetWatch("reset", broken, threshold=0.01, stream=stream).start()
+        self.assertTrue(reported.wait(SERVER_TIMEOUT_S))
+        self.assertIn("state unavailable: RuntimeError('no state')", stream.getvalue())
+        self.assertIn("most recent call first", stream.getvalue())
+
+
+class TestSlowResetOnTheServer(unittest.TestCase):
+    """A reset behind another process's write lock, on tests/browser/server.py."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        env = {**os.environ, "TMPDIR": self.tmp.name}
+        env.update(TEMP=self.tmp.name, TMP=self.tmp.name)
+        self.stderr = tempfile.TemporaryFile()
+        self.addCleanup(self.stderr.close)
+        process = subprocess.Popen(
+            [sys.executable, "-m", "tests.browser.server", "--port", "0"],
+            cwd=REPO_ROOT,
+            env=env,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=self.stderr,
+        )
+
+        def stop():
+            if process.poll() is None:
+                process.kill()
+            process.wait(timeout=SERVER_TIMEOUT_S)
+            for stream in (process.stdin, process.stdout):
+                stream.close()
+
+        self.addCleanup(stop)
+        line = process.stdout.readline().decode()
+        self.assertTrue(line.startswith("PM_TEST_SERVER_PORT="), line)
+        self.base = f"http://127.0.0.1:{int(line.split('=', 1)[1])}"
+        deadline = time.monotonic() + SERVER_TIMEOUT_S
+        while True:
+            try:
+                with urllib.request.urlopen(  # nosec B310
+                    self.base + "/prompt_manager/test", timeout=1
+                ):
+                    break
+            except OSError:
+                self.assertIsNone(process.poll(), "the server exited early")
+                self.assertLess(time.monotonic(), deadline, "never ready")
+                time.sleep(0.1)
+        [self.db_path] = glob.glob(
+            os.path.join(self.tmp.name, "pm-browser-*", "browser.db")
+        )
+
+    def reset(self) -> tuple[int, float]:
+        request = urllib.request.Request(
+            self.base + "/__test__/reset",
+            data=b"{}",
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        start = time.monotonic()
+        with urllib.request.urlopen(  # nosec B310
+            request, timeout=SERVER_TIMEOUT_S
+        ) as response:
+            return response.status, time.monotonic() - start
+
+    def server_stderr(self) -> str:
+        self.stderr.seek(0)
+        return self.stderr.read().decode("utf-8", "replace")
+
+    def test_a_reset_held_behind_a_write_lock_names_the_wait_and_ends_with_it(self):
+        self.assertEqual(self.reset()[0], 200, self.server_stderr())
+        self.assertNotIn(REPORT_PREFIX, self.server_stderr())
+
+        held = threading.Event()
+
+        def hold_write_lock():
+            holder = sqlite3.connect(self.db_path, timeout=SERVER_TIMEOUT_S)
+            try:
+                holder.execute("BEGIN IMMEDIATE")
+                held.set()
+                time.sleep(HOLD_SECONDS)
+                holder.rollback()
+            finally:
+                holder.close()
+
+        lock = threading.Thread(target=hold_write_lock)
+        lock.start()
+        self.addCleanup(lock.join)
+        self.assertTrue(held.wait(SERVER_TIMEOUT_S))
+        outcome = []
+        resetting = threading.Thread(target=lambda: outcome.append(self.reset()))
+        resetting.start()
+        self.addCleanup(resetting.join)
+        slowest = 0.0
+        while resetting.is_alive():
+            start = time.monotonic()
+            with urllib.request.urlopen(  # nosec B310
+                self.base + "/prompt_manager/test", timeout=SERVER_TIMEOUT_S
+            ):
+                slowest = max(slowest, time.monotonic() - start)
+            time.sleep(PROBE_INTERVAL_SECONDS)
+        resetting.join()
+        lock.join()
+        self.assertEqual(len(outcome), 1, self.server_stderr())
+        [(status, elapsed)] = outcome
+
+        self.assertLess(slowest, RESPONSIVE_SECONDS)
+        self.assertEqual(status, 200)
+        self.assertGreaterEqual(elapsed, HOLD_SECONDS - 0.5)
+        self.assertLess(elapsed, HOLD_SECONDS + RESET_MARGIN_SECONDS)
+        report = [
+            line[len(REPORT_PREFIX) :]
+            for line in self.server_stderr().splitlines()
+            if line.startswith(REPORT_PREFIX)
+        ]
+        text = "\n".join(report)
+        self.assertIn("POST /__test__/reset still running after", text)
+        self.assertIn("default executor:", text)
+        if sys.platform.startswith("linux"):
+            # SQLite's busy handler sleeps between its retries.
+            self.assertRegex(
+                text, r"kernel: asyncio_\d+ \(0x[0-9a-f]{16}\) state \w, waiting in "
+            )
+        # The plain reset's seed() waits on its first DELETE.
+        self.assertRegex(text, r"line \d+ in seed\n")
+        self.assertIn("POST /__test__/reset answered 200 after", text)
+        self.assertRegex(
+            text,
+            r"executor call create_app.<locals>.seed: queued \d+\.\d+s, ran \d+\.\d+s",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_execution_output_writer.py

@@ -299,45 +299,126 @@ class OutputWriterTests(fixtures.unittest.TestCase):
         )
         self.assertEqual(writer.context_budget.charged_bytes, 0)
 
+    def isolated_writer(self) -> HistoryWriter:
+        """A ready writer whose budget no other writer or test can charge."""
+        writer = HistoryWriter(
+            lambda: self.db, clock=self.clock, context_budget=ContextBudget()
+        )
+        self.writers.append(writer)
+        writer.start()
+        self.wait_for(lambda: writer.status().accepting)
+        return writer
+
+    def rotate_between_transactions(self, writer, start, after, restore, hold=None):
+        """Wrap ``after`` so a restore or invalidation lands right after it.
+
+        Returns the patch and an event set once that rotation has returned on
+        the worker. A restore is a full verify/migrate/backup/replace cycle, so
+        the handoff waits on this event with the generous ``BLOCK_SECONDS``
+        ceiling instead of the short worker-behavior deadline (TESTING.md,
+        Windows portability preflight).
+        """
+        backup = os.path.join(self.directory.name, f"restore-{start.run_id}.db")
+        self.assertTrue(self.db.model.backup_database(backup))
+        original = getattr(self.db, after)
+        rotated = threading.Event()
+
+        def write(*args, **kwargs):
+            result = original(*args, **kwargs)
+            try:
+                if hold is not None:
+                    entered, release = hold
+                    entered.set()
+                    release.wait(BLOCK_SECONDS)
+                if restore:
+                    self.db.model.restore_from_file(backup)
+                else:
+                    writer.invalidate()
+            finally:
+                rotated.set()
+            return result
+
+        return patch.object(self.db, after, write), rotated
+
     def test_restore_or_generation_rotation_between_transactions_discards_evidence(
         self,
     ):
         for after in ("write_execution_finish", "write_execution_contexts_for_finish"):
             for restore in (False, True):
                 with self.subTest(after=after, restore=restore):
-                    writer = self.writer()
-                    start = self.start_command(len(self.writers))
-                    self.db.write_execution_start(start)
-                    backup = os.path.join(self.directory.name, "restore.db")
-                    self.assertTrue(self.db.model.backup_database(backup))
-                    original = getattr(self.db, after)
+                    # Each subtest owns its writer and budget and retires them
+                    # before the next starts, so one slow restore cannot leave
+                    # in-flight charges or an epoch rotation for the next.
+                    writer = self.isolated_writer()
+                    try:
+                        start = self.start_command(len(self.writers))
+                        self.db.write_execution_start(start)
+                        rotation, rotated = self.rotate_between_transactions(
+                            writer, start, after, restore
+                        )
+                        with rotation:
+                            self.finalization(writer, start)
+                            self.assertTrue(rotated.wait(BLOCK_SECONDS))
+                            self.wait_for(lambda: writer.status().pending == 0)
+                        self.assertEqual(
+                            self.db.get_execution_outputs(start.run_id)["state"],
+                            "not_captured",
+                        )
+                        self.assertEqual(
+                            self.db.get_execution_contexts(start.run_id)["state"],
+                            (
+                                "complete"
+                                if after == "write_execution_contexts_for_finish"
+                                and not restore
+                                else "not_captured"
+                            ),
+                        )
+                        self.assertEqual(writer.context_budget.charged_bytes, 0)
+                    finally:
+                        self.assertTrue(writer.close(BLOCK_SECONDS))
 
-                    def write(*args, **kwargs):
-                        result = original(*args, **kwargs)
-                        if restore:
-                            self.db.model.restore_from_file(backup)
-                        else:
-                            writer.invalidate()
-                        return result
-
-                    with patch.object(self.db, after, write):
-                        self.finalization(writer, start)
-                        self.wait_for(lambda: writer.status().pending == 0)
-                    self.assertEqual(
-                        self.db.get_execution_outputs(start.run_id)["state"],
-                        "not_captured",
-                    )
-                    self.assertEqual(
-                        self.db.get_execution_contexts(start.run_id)["state"],
-                        (
-                            "complete"
-                            if after == "write_execution_contexts_for_finish"
-                            and not restore
-                            else "not_captured"
-                        ),
-                    )
-                    self.assertEqual(writer.context_budget.charged_bytes, 0)
-                    self.assertTrue(writer.close())
+    def test_slow_restore_between_transactions_holds_charge_only_while_in_flight(
+        self,
+    ):
+        writer = self.isolated_writer()
+        start = self.start_command()
+        self.db.write_execution_start(start)
+        entered, release = self.block()
+        rotation, rotated = self.rotate_between_transactions(
+            writer, start, "write_execution_finish", True, (entered, release)
+        )
+        with rotation:
+            context = self.sealed(writer)
+            outputs = self.outputs(writer, context.lease.byte_count)
+            charge = writer.context_budget.charged_bytes
+            self.assertGreater(charge, 0)
+            self.assertTrue(
+                writer.submit_finalization(self.finish_command(start), context, outputs)
+            )
+            self.assertTrue(entered.wait(BLOCK_SECONDS))
+            # The worker is between transactions: the payload is owned, not leaked.
+            self.assertEqual(writer.status().pending, 1)
+            self.assertEqual(writer.context_budget.charged_bytes, charge)
+            release.set()
+            self.assertTrue(rotated.wait(BLOCK_SECONDS))
+            self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        self.assertFalse(writer.status().accepting)
+        self.assertEqual(
+            self.db.get_execution_outputs(start.run_id)["state"], "not_captured"
+        )
+        # The restored epoch is probed again and new evidence is accepted.
+        self.clock.advance(5)
+        self.wait_for(lambda: writer.status().accepting)
+        restored = self.start_command(2)
+        self.db.write_execution_start(restored)
+        self.finalization(writer, restored)
+        self.wait_for(lambda: writer.status().pending == 0)
+        self.assertEqual(
+            self.db.get_execution_outputs(restored.run_id)["state"], "complete"
+        )
+        self.assertEqual(writer.context_budget.charged_bytes, 0)
+        self.assertTrue(writer.close(BLOCK_SECONDS))
 
     def test_invalidation_during_output_preparation_rechecks_actual_commit(self):
         import database.execution_outputs as outputs_db

tests/test_workspace_concept_fixture.py

@@ -1,7 +1,9 @@
 """The concept fixture files carry the concept inspector facts (app.js:400-414)."""
 
 import json
+import math
 import tempfile
+import time
 import unittest
 from pathlib import Path
 
@@ -15,6 +17,8 @@ except ImportError:  # discovered with tests/ as the top-level directory
 from tests.browser.workspace_concept_fixture import (
     CONCEPT_MODEL,
     CONCEPT_PROMPTS,
+    MAX_EPOCH_SECONDS,
+    _instant,
     write_concept_png,
 )
 from utils.prompt_inspector import inspect_prompt_evidence, read_inspection_file
@@ -80,5 +84,29 @@ class TestWorkspaceConceptFixture(unittest.TestCase):
             self.assertEqual((image["width"], image["height"]), size)
 
 
+class TestConceptFixtureInstant(unittest.TestCase):
+    """One "now" for every relative time the fixture writes (PR #284's midnight)."""
+
+    def test_a_chosen_instant_is_kept(self):
+        self.assertEqual(_instant(1_792_713_598), 1_792_713_598.0)
+        self.assertEqual(_instant(1_792_713_598.25), 1_792_713_598.25)
+
+    def test_the_wall_clock_when_omitted(self):
+        before = time.time()
+        now = _instant(None)
+        self.assertLessEqual(before, now)
+        self.assertLessEqual(now, time.time())
+
+    def test_a_non_number_is_refused(self):
+        for value in (True, "1792713598", [1]):
+            with self.subTest(value=value), self.assertRaises(TypeError):
+                _instant(value)
+
+    def test_an_instant_out_of_range_is_refused(self):
+        for value in (0, -1, MAX_EPOCH_SECONDS, math.inf, math.nan):
+            with self.subTest(value=value), self.assertRaises(ValueError):
+                _instant(value)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/vendor/browser-server-fixture.test.mjs

@@ -6,7 +6,7 @@ import { once } from 'node:events';
 import { readFileSync } from 'node:fs';
 import { fileURLToPath, pathToFileURL } from 'node:url';
 import { setTimeout as delay } from 'node:timers/promises';
-import { startBrowserServer, createServerSupervisor, attachServerStderr } from '../../tools/vendor/start-browser-server.mjs';
+import { startBrowserServer, createServerSupervisor, attachServerStderr, forwardPrefixedLines, SLOW_RESET_PREFIX } from '../../tools/vendor/start-browser-server.mjs';
 import { workflowWorkers } from '../../tools/vendor/test-workers.mjs';
 
 const fakeServer = fileURLToPath(new URL('./fake-browser-server.mjs', import.meta.url));
@@ -193,6 +193,29 @@ test('a failed test gets the server stderr written during it attached; a passing
     assert.equal(String(attached[0].body), 'Traceback (most recent call last): boom\n');
 });
 
+test('only complete slow-reset report lines are forwarded, however the chunks split them', () => {
+    const written = [];
+    const forward = forwardPrefixedLines(SLOW_RESET_PREFIX, text => written.push(text));
+    forward(`aiohttp access line\n${SLOW_RESET_PREFIX}POST /__test__/reset still`);
+    assert.deepEqual(written, [], 'a report line waits for its newline');
+    forward(` running after 2.000s\nplain stderr\n${SLOW_RESET_PREFIX}Thread 0x1 [asyncio_0]\n${SLOW_RESET_PREFIX.slice(0, 5)}`);
+    forward(`${SLOW_RESET_PREFIX.slice(5)}  File "server.py", line 1\nnot a report ${SLOW_RESET_PREFIX}\n`);
+    assert.deepEqual(written, [
+        `${SLOW_RESET_PREFIX}POST /__test__/reset still running after 2.000s\n${SLOW_RESET_PREFIX}Thread 0x1 [asyncio_0]\n`,
+        `${SLOW_RESET_PREFIX}  File "server.py", line 1\n`,
+    ]);
+});
+
+test('a slow-reset report on server stderr reaches the worker stderr; other stderr does not', async t => {
+    const reported = [];
+    const server = await startBrowserServer({ ...fake(), report: text => reported.push(text) });
+    t.after(() => server.stop());
+    await writeStderr(server, 'Traceback: ordinary stderr\n');
+    await writeStderr(server, `${SLOW_RESET_PREFIX}POST /__test__/reset answered 200 after 16.064s\n`);
+    assert.deepEqual(reported, [`${SLOW_RESET_PREFIX}POST /__test__/reset answered 200 after 16.064s\n`]);
+    assert.ok(server.stderrTail().includes('Traceback: ordinary stderr'), 'the stderr tail still keeps everything');
+});
+
 test('a server exits with its worker even when the worker is SIGKILLed', async () => {
     const pid = await serverOfKilledWorker(`{ command: process.execPath, args: port => [${JSON.stringify(fakeServer)}, '--port', String(port), '--exit-on-stdin-eof'] }`);
     await waitFor(() => !alive(pid), `server ${pid} to exit after its worker died`);

tools/vendor/start-browser-server.mjs

@@ -17,6 +17,22 @@ const HOST = '127.0.0.1';
 // tests/browser/server.py prints this once it holds its listening socket. Only a
 // complete line counts: a chunk may end mid-number.
 const PORT_ANNOUNCEMENT = /^PM_TEST_SERVER_PORT=(\d+)\r?\n/m;
+// tests/browser/slow_reset_watch.py starts every line of a slow-reset report with
+// this. Those lines go to the worker's stderr, so the job log shows a slow reset
+// even when its test passes.
+export const SLOW_RESET_PREFIX = 'pm-slow-reset| ';
+
+// A stderr chunk handler that passes every complete line starting with `prefix`
+// to `write`. A line split across chunks is held until its newline arrives.
+export function forwardPrefixedLines(prefix, write) {
+    let pending = '';
+    return chunk => {
+        const lines = (pending + chunk).split('\n');
+        pending = lines.pop().slice(-STDERR_TAIL_BYTES);
+        const matched = lines.filter(line => line.startsWith(prefix));
+        if (matched.length) write(matched.map(line => `${line}\n`).join(''));
+    };
+}
 
 // The server exits when its stdin closes, so a worker that dies without teardown
 // (SIGKILL) leaves no server behind; the launcher holds the pipe's other end.
@@ -40,6 +56,7 @@ export async function startBrowserServer({
     cwd = repositoryRoot,
     timeoutMs = STARTUP_TIMEOUT_MS,
     stderrLogChars = STDERR_LOG_CHARS,
+    report = chunk => process.stderr.write(chunk),
 } = {}) {
     if (!Number.isInteger(port) || port < 0 || port > 65535) {
         throw new Error(`startBrowserServer needs a port from 0 to 65535, got ${port}`);
@@ -50,7 +67,9 @@ export async function startBrowserServer({
     let stderr = '';
     let stderrTotal = 0;
     child.stderr.setEncoding('utf8');
+    const forwardReport = forwardPrefixedLines(SLOW_RESET_PREFIX, report);
     child.stderr.on('data', chunk => {
+        forwardReport(chunk);
         stderr = (stderr + chunk).slice(-Math.max(stderrLogChars, STDERR_TAIL_BYTES));
         stderrTotal += chunk.length;
     });