diff --git a/docs/developer/adr/0002-a-value-lives-while-something-keeps-it.md b/docs/developer/adr/0002-a-value-lives-while-something-keeps-it.md
index 256305eb..042c4d93 100644
--- a/docs/developer/adr/0002-a-value-lives-while-something-keeps-it.md
+++ b/docs/developer/adr/0002-a-value-lives-while-something-keeps-it.md
@@ -64,8 +64,9 @@ History and the store survive.
 The values the workspace kept in memory do not.
 Each pending record is
 
-- cancelled, unless a persist request names it, or a record named by a persist request reads it, directly or through other pending records;
-- run, if every value it reads is a file, is written, or comes from a pending record that runs;
+- failed, if it had started: the workspace ended while it ran, perhaps because of it, as an out-of-memory kill does, so running it again could end the process again;
+- otherwise cancelled, unless a persist request names it, or a record named by a persist request reads it, directly or through other pending records;
+- otherwise run, if every value it reads is a file, is written, or comes from a pending record that runs;
 - failed otherwise, as the record of a freeze is, since its driver is gone.
 
 ## Alternatives considered
@@ -106,4 +107,4 @@ To be designed with the service (scipp/essapps#27):
 - What a finish records besides the status, such as the software environment.
 - When values are dropped from the store before history (system story H1), and by whom.
 - The file format of each output type, and the layout of the store in a proposal's area.
-- An upgrade while persisted records run. If the old instance finishes them, the new version waits, for hours after an overnight batch, since only one workspace writes a proposal's history. If the new instance runs them again, the time they had run is lost. Either way, the upgrade closes every workspace of the old instance.
+- An upgrade while persisted records run. An orderly stop of the old instance waits for its running records, so the new version waits, for hours after an overnight batch, since only one workspace writes a proposal's history. If the old instance is killed instead, its running records fail at the new instance's start, and a person or the application submits them again, so the time they had run is lost. Either way, the upgrade closes every workspace of the old instance.
diff --git a/docs/developer/prior-art/aiida.md b/docs/developer/prior-art/aiida.md
index bb8f24c3..cc83f707 100644
--- a/docs/developer/prior-art/aiida.md
+++ b/docs/developer/prior-art/aiida.md
@@ -72,8 +72,8 @@ Whether the workspace retries before it fails a record, or marks the failure so
 **Checkpointed workflow state does not survive upgrades.** A WorkChain resumes after a daemon restart because its state is serialised between steps.
 The unreleased version says such checkpoints "cannot be continued after upgrading. Finish or terminate all active processes before upgrading" ([CHANGELOG L11-12][cl-head]).
 The FAQ says to restart the daemon after every code change ([faq.rst L29][faq]).
-Our workspace pickles nothing: after a restart it reruns persisted pending records from history ([ADR 0002](../adr/0002-a-value-lives-while-something-keeps-it.md)).
-On an upgrade with work in flight, AiiDA's answer is to drain first, the "old instance finishes them" option of ADR 0002, Open, and system story H2.
+Our workspace pickles nothing: after a restart it runs again, from history, the persisted pending records that had not started ([ADR 0002](../adr/0002-a-value-lives-while-something-keeps-it.md)).
+On an upgrade with work in flight, AiiDA's answer is to drain first, the orderly stop of ADR 0002, Open, and system story H2.
 Warm workers (scipp/essapps#27) hold imported code as AiiDA's daemon workers do, so a package upgrade must restart them.
 
 **A broker queue drifted from the database.** v1.0 made RabbitMQ mandatory for process control.
diff --git a/docs/developer/system-stories.md b/docs/developer/system-stories.md
index e29b0744..aad374f3 100644
--- a/docs/developer/system-stories.md
+++ b/docs/developer/system-stories.md
@@ -110,7 +110,7 @@ Property: stored outputs can be dropped by proposal, label, or age. Every record
 ### H2. Upgrade with runs in flight
 
 Actor: DMSC. Goal: deploy a new version of the service while requests run and others wait on them.
-Property: every record pending at the upgrade with a persist request finishes without anyone submitting it again, and so do the records it reads. Whether the old instance finishes them before it stops, or the new one runs them again, is open ([ADR 0002](adr/0002-a-value-lives-while-something-keeps-it.md), Open); a record that reads an accumulator's driver finishes only if the old instance finishes it. The upgrade closes every workspace of the old instance, and pending work that nothing persists is cancelled. Records and persisted values written before the upgrade stay readable, also when the new version stores records in another schema. A workspace started on history that another workspace still writes is refused, so the old and the new version never write the same history.
+Property: every record pending at the upgrade with a persist request finishes without anyone submitting it again, and so do the records it reads. Closing a workspace waits for them, so the old instance finishes them before it stops; how long an upgrade may wait for that is open ([ADR 0002](adr/0002-a-value-lives-while-something-keeps-it.md), Open). A process that is to stop in this orderly way on SIGTERM has to close its workspaces when it receives the signal, and no code in the repository handles the signal. An instance killed instead leaves them pending: at the next start, a record that was running fails, since running it again could end the process again, and a record that had not started runs if what it reads is there. A record that reads an accumulator's driver finishes only if the old instance finishes it. The upgrade closes every workspace of the old instance, and pending work that nothing persists and that had not started is cancelled. Records and persisted values written before the upgrade stay readable, also when the new version stores records in another schema. A workspace started on history that another workspace still writes is refused, so the old and the new version never write the same history.
 
 ### H3. History ends when the work does
 
diff --git a/packages/essapps/docs/batch.md b/packages/essapps/docs/batch.md
index 86d6ab31..acd219c6 100644
--- a/packages/essapps/docs/batch.md
+++ b/packages/essapps/docs/batch.md
@@ -263,7 +263,7 @@ $B reduce iofq.yaml runs.csv --label water-series
 
 `{'@latest': label, output: name}` is read when the command starts, as the output of the newest record under the label, of `member` if given, whoever submitted it. It is for a template, not for a cell of a table. The record of each row holds the ID that it was read as. A reference by ID is `{'@ref': output, record: ..., output: ...}`.
 
-One command holds the folder at a time. If a command was killed, the next one first finishes the rows that were running, so reduce again with `--remaining`. A workspace reads a persisted output by the type that its spec declares, so only one that runs the spec reads it. Nothing removes the files of the store.
+One command holds the folder at a time. If a command was killed, the next one first finishes the rows that had not started. A row that was running fails, since its workflow may have been what ended the process. Reduce again with `--remaining` to submit every row that has not completed, the failed ones included. A workspace reads a persisted output by the type that its spec declares, so only one that runs the spec reads it. Nothing removes the files of the store.
 
 ## The UI
 
diff --git a/packages/essapps/src/ess/apps/log.py b/packages/essapps/src/ess/apps/log.py
index fe7d5e6c..3d7030d0 100644
--- a/packages/essapps/src/ess/apps/log.py
+++ b/packages/essapps/src/ess/apps/log.py
@@ -4,22 +4,25 @@
 The workspace's history: an append-only log of what ran, with which params,
 and what came of it, for one proposal, which each submission and opening names.
 
-There are four events: a submission, a record that finished, an accumulator
-that opened, and a push into one. A persist request is part of a submission or
-a freeze, and its write is the record's: the record finishes once the outputs
-are written. A freeze is a submission of one record of the plain request of the
-state frozen, as the accumulator and its number of pushes; the views list its
-request when it is read. The workspace's views, such as the records by ID, are
-built by applying the events in order, when they are appended and again when a
-workspace starts from an existing log.
-
-An event names only records and accumulators appended before it: a finish
-names its record, a push its accumulator, a record the records it reads, and
-the record of a freeze its accumulator. Applying the events in order relies on
-this, and so does a restart, which walks the pending records backwards.
+There are five events: a submission, a record that started, a record that
+finished, an accumulator that opened, and a push into one. A record starts when
+a worker begins its work, once what it reads is there, so a record that started
+and did not finish was running when its workspace ended. A persist request is
+part of a submission or a freeze, and its write is the record's: the record
+finishes once the outputs are written. A freeze is a submission of one record of
+the plain request of the state frozen, as the accumulator and its number of
+pushes; the views list its request when it is read. The workspace's views,
+such as the records by ID, are built by applying the events in order, when they
+are appended and again when a workspace starts from an existing log.
+
+An event names only records and accumulators appended before it: a start and
+a finish name their record, a push its accumulator, a record the records it
+reads, and the record of a freeze its accumulator. Applying the events in order
+relies on this, and so does a restart, which walks the pending records
+backwards.
 
 A log file starts with a header line that names its format, such as
-``{"format": 1}``, and holds one event per line after it. A log reads only
+``{"format": 2}``, and holds one event per line after it. A log reads only
 files of its own format, :data:`FORMAT`, and leaves a file it refuses as it was.
 
 Stages and accumulators are not history, and neither are output values or
@@ -83,6 +86,13 @@ class Submitted(BaseModel, frozen=True, extra='forbid'):
     records: tuple[NewRecord | NewStateRecord, ...]
 
 
+class Started(BaseModel, frozen=True, extra='forbid'):
+    """A record whose work began on a worker: what it reads was there."""
+
+    kind: Literal['started'] = 'started'
+    record: str
+
+
 class Finished(BaseModel, frozen=True, extra='forbid'):
     """
     A record that finished: ``failure`` says why it failed, or why the
@@ -129,7 +139,7 @@ class Pushed(BaseModel, frozen=True, extra='forbid'):
 
 
 Event = Annotated[
-    Submitted | Finished | Opened | Pushed,
+    Submitted | Started | Finished | Opened | Pushed,
     Field(discriminator='kind'),
 ]
 _event: TypeAdapter[Event] = TypeAdapter(Event)
@@ -137,7 +147,7 @@ class Pushed(BaseModel, frozen=True, extra='forbid'):
 # The format of a log file, named by its first line. Bumped on any change to
 # the events or to the files of the store (ess.apps.hdf5), since stored values
 # are found through the log.
-FORMAT = 1
+FORMAT = 2
 _HEADER = json.dumps({'format': FORMAT}).encode() + b'\n'
 
 
diff --git a/packages/essapps/src/ess/apps/protocol.py b/packages/essapps/src/ess/apps/protocol.py
index ecfc7aab..5ab9f591 100644
--- a/packages/essapps/src/ess/apps/protocol.py
+++ b/packages/essapps/src/ess/apps/protocol.py
@@ -470,11 +470,7 @@ class Store(Protocol):
     """
 
     def write(self, record: str, output: str, value: Any) -> None:
-        """
-        Write the value of an output; raises if it cannot. A write replaces
-        a value written before under the same names, as a record run again
-        after a restart writes its outputs again.
-        """
+        """Write the value of an output; raises if it cannot."""
         ...
 
     def read(self, record: str, output: str, tp: Any) -> Any:
diff --git a/packages/essapps/src/ess/apps/views.py b/packages/essapps/src/ess/apps/views.py
index 64588fe6..a2e18f87 100644
--- a/packages/essapps/src/ess/apps/views.py
+++ b/packages/essapps/src/ess/apps/views.py
@@ -25,6 +25,7 @@
     NewStateRecord,
     Opened,
     Pushed,
+    Started,
     Submitted,
 )
 from .records import Changes, Record, Request, SpecId, Status
@@ -86,21 +87,23 @@ class Views:
     The records, labels, accumulators, and pushes that the events describe.
 
     ``finished`` holds the event that finished a record; a record without one
-    is pending. ``states`` holds the accumulator and number of pushes of each
-    record of a freeze. ``accumulators`` holds the event that opened each
-    accumulator, and ``pushes`` its pushes in order, each rows of its tables.
+    is pending. ``started`` names the records whose work began. ``states`` holds
+    the accumulator and number of pushes of each record of a freeze.
+    ``accumulators`` holds the event that opened each accumulator, and
+    ``pushes`` its pushes in order, each rows of its tables.
 
     ``persisted`` names the outputs each record persists; their write is the
     record's: pending while it is, done once it completes.
 
     ``feed`` holds the ID of a record each time an event submits or finishes
     it, in the order of the events. A position in it is a
-    cursor of :meth:`changes`. Opening an accumulator and pushing to it are not in
-    the feed.
+    cursor of :meth:`changes`. Starting a record, opening an accumulator, and
+    pushing to it are not in the feed.
     """
 
     def __init__(self) -> None:
         self.records = _Records(self)
+        self.started: set[str] = set()
         self.finished: dict[str, Finished] = {}
         self.labels: dict[str, list[str]] = {}
         self.states: dict[str, tuple[str, int]] = {}
@@ -115,6 +118,8 @@ def apply(self, event: Event) -> None:
                 for new in event.records:
                     self._add(new, event)
                     self.feed.append(new.id)
+            case Started():
+                self.started.add(event.record)
             case Finished():
                 self.feed.append(event.record)
                 self.finished[event.record] = event
diff --git a/packages/essapps/src/ess/apps/workspace.py b/packages/essapps/src/ess/apps/workspace.py
index d4c40952..e77b7828 100644
--- a/packages/essapps/src/ess/apps/workspace.py
+++ b/packages/essapps/src/ess/apps/workspace.py
@@ -10,16 +10,17 @@
 its pieces fit.
 
 What the workspace knows of history is its log (see ``log.py``): every submission,
-finished record, opened accumulator, and push is appended as one event and then
-applied to its views (see ``views.py``). A change is checked before its event is
-appended, so the log holds only events that apply. A workspace given a log that
-already has events applies them first; a log holds the history of one proposal,
-so a log with an event of another proposal is refused. Nothing the earlier
-workspace kept survives, so only persisted work goes on (see
-:meth:`Workspace._restart`). The log is the workspace's only record of history, so
-if an append fails, for example on a full disk, the workspace no longer knows its
-own state and stops: every call waiting in it, and every later call, raises
-``RuntimeError`` from the append's error.
+started and finished record, opened accumulator, and push is appended as one
+event and then applied to its views (see ``views.py``). A change is checked
+before its event is appended, so the log holds only events that apply. A
+workspace given a log that already has events applies them first; a log holds
+the history of one proposal, so a log with an event of another proposal is
+refused. Nothing the earlier workspace kept survives, so only persisted work
+that had not started goes on (see :meth:`Workspace._restart`). The log is the
+workspace's only record of history, so if an append fails, for example on a
+full disk, the workspace no longer knows its own state and stops: every call
+waiting in it, and every later call, raises ``RuntimeError`` from the append's
+error.
 
 The workspace keeps the outputs of records, and its stages and accumulators. A
 client is a connection to the workspace: it keeps nothing, and every client reads
@@ -185,6 +186,7 @@
     NewStateRecord,
     Opened,
     Pushed,
+    Started,
     Submitted,
 )
 from .protocol import (
@@ -637,17 +639,21 @@ def _restart(self) -> None:
         lock held. Nothing the earlier workspace kept survived, so only persisted
         work goes on:
 
-        1. A pending record is cancelled unless it persists outputs, or a
-           pending record that persists outputs reads it, directly or through
-           other pending records.
-        2. Otherwise it runs if each value it reads is a file, is written,
+        1. A pending record fails if it started: the workspace ended while it
+           ran, perhaps because of it, as an out-of-memory kill does, so
+           running it again could end the process again. A person submits its
+           request again to run it.
+        2. Otherwise it is cancelled unless it persists outputs, or a pending
+           record that persists outputs reads it, directly or through other
+           pending records.
+        3. Otherwise it runs if each value it reads is a file, is written,
            or is an output of a pending record that runs, and fails if not.
            The record of a freeze fails, since the driver it reads is gone.
            A call through a stage runs as its plain request, since the stage
            is gone, so it reads the values of the stage's template too.
 
-        A record that had written its persisted outputs, but
-        not finished, runs again, and its writes replace what the store holds.
+        So a record that runs at a restart had not started, and has written
+        nothing.
         """
         views = self._views
         pending = [i for i in views.records if i not in views.finished]
@@ -658,7 +664,11 @@ def _restart(self) -> None:
                 needed.update(ref.record for ref in upstream)
         runs: set[str] = set()
         for record_id in pending:
-            if record_id not in needed:
+            if record_id in views.started:
+                self._finish(
+                    record_id, Status.FAILED, 'the workspace ended while it ran'
+                )
+            elif record_id not in needed:
                 self._finish(record_id, Status.CANCELLED, 'the workspace restarted')
             elif (reason := self._unreadable(record_id, runs)) is not None:
                 self._finish(
@@ -997,6 +1007,8 @@ def _run(self, record_id: str) -> None:
         """
         Run a record's workflow, holding the values it reads until it returns,
         and write the outputs persisted at its submission before it completes.
+        The record's start is logged before anything is read, so that a
+        restart fails a record that was running when the workspace ended.
 
         It holds them even once the record is cancelled, since the workflow
         still reads them, and lets go of them before the write. The record of
@@ -1006,6 +1018,7 @@ def _run(self, record_id: str) -> None:
         with self._changed:
             if record_id in self._views.finished:
                 return
+            self._append(Started(record=record_id))
             record = self._views.records[record_id]
             stage = self._stage_of.get(record_id)
             state = self._state_of.get(record_id)
diff --git a/packages/essapps/tests/batch/frames_test.py b/packages/essapps/tests/batch/frames_test.py
index 8bf89687..f3dbc67c 100644
--- a/packages/essapps/tests/batch/frames_test.py
+++ b/packages/essapps/tests/batch/frames_test.py
@@ -22,7 +22,7 @@
     write_table,
 )
 from ess.apps.folder import open_folder
-from ess.apps.log import Finished, Log
+from ess.apps.log import Finished, Log, Started
 from ess.apps.testing import StagedFunction
 from ess.spec import Binding, WorkflowSpec, function_binding
 
@@ -329,12 +329,14 @@ def test_rows_pending_after_a_restart_finish_and_are_not_submitted_again(
     header, *lines = log.read_bytes().splitlines(keepends=True)
     events = Log.read(log)
     assert len(lines) == len(events)
-    unfinished = [
-        ln for ln, e in zip(lines, events, strict=True) if type(e) is not Finished
+    unstarted = [
+        ln
+        for ln, e in zip(lines, events, strict=True)
+        if not isinstance(e, Started | Finished)
     ]
-    assert len(unfinished) < len(lines)
-    # as if the process had been killed
-    log.write_bytes(b''.join([header, *unfinished]))
+    assert len(unstarted) < len(lines)
+    # as if the process had been killed before the records started
+    log.write_bytes(b''.join([header, *unstarted]))
 
     with open_folder(folder, folder_bindings) as second:
         result = reduce_table(
diff --git a/packages/essapps/tests/log_test.py b/packages/essapps/tests/log_test.py
index ab6be301..fbf7c67e 100644
--- a/packages/essapps/tests/log_test.py
+++ b/packages/essapps/tests/log_test.py
@@ -35,6 +35,7 @@
     NewStateRecord,
     Opened,
     Pushed,
+    Started,
     Submitted,
 )
 from ess.apps.testing import FakeFiles, StagedFunction, kept_rows, read_file
@@ -148,9 +149,18 @@ def windows() -> list[Any]:
     return []
 
 
+@pytest.fixture
+def entered() -> threading.Event:
+    """Set each time LOAD's workflow is entered."""
+    return threading.Event()
+
+
 @pytest.fixture
 def start(
-    files: FakeFiles, loading: threading.Event, windows: list[Any]
+    files: FakeFiles,
+    loading: threading.Event,
+    windows: list[Any],
+    entered: threading.Event,
 ) -> Iterator[Callable[[Path], Client]]:
     """A client of a new workspace over the log at a path."""
     workspaces: list[Workspace] = []
@@ -158,6 +168,7 @@ def start(
     def load(params: LoadParams) -> ValueOutputs:
         window = params.window
         windows.append(window)
+        entered.set()
         loading.wait(timeout=5)
         value = read_file(params.sample) * (window.stop - window.start)
         return ValueOutputs(value=Amount(value=value))
@@ -277,7 +288,7 @@ def test_an_accumulator_is_logged_as_its_template_and_pushes_and_a_freeze_as_a_r
     frozen = client.freeze(total)
 
     events = Log.read(tmp_path / 'log')
-    assert events[6:10] == [
+    assert events[9:13] == [
         Opened(
             accumulator=total.id,
             proposal='p1',
@@ -292,7 +303,7 @@ def test_an_accumulator_is_logged_as_its_template_and_pushes_and_a_freeze_as_a_r
             for load in loads
         ),
     ]
-    assert events[10].records == (
+    assert events[13].records == (
         NewStateRecord(
             id=frozen.id, accumulator=total.id, pushes=3, outputs=('value',)
         ),
@@ -362,30 +373,41 @@ def test_a_reference_is_logged_in_its_json_form_and_read_back(tmp_path: Path) ->
     assert Log.read(tmp_path / 'log') == events
 
 
-def test_records_pending_in_the_log_are_cancelled_at_a_restart(
+def test_records_pending_in_the_log_are_cancelled_or_failed_at_a_restart(
     tmp_path: Path,
     start: Callable[[Path], Client],
     restart: Callable[[Path], Client],
     loading: threading.Event,
+    entered: threading.Event,
 ) -> None:
     first = start(tmp_path / 'log')
     done = first.compute(LOAD, params=_sample(1))
     loading.clear()
-    pending = first.submit(LOAD, params=_sample(2))
+    entered.clear()
+    running = first.submit(LOAD, params=_sample(2))
     total = _total(first)
     total.push(part=[done.ref('value')])
-    total.push(part=[pending.ref('value')])  # waits for ``pending``
+    total.push(part=[running.ref('value')])  # waits for ``running``
     frozen = first.freeze(total)
     read = first.submit(TOTAL, params={'part': [frozen.ref('value')]})
+    assert entered.wait(timeout=5)  # ``running`` started
 
     again = restart(tmp_path / 'log')
     loading.set()
 
-    records = [pending, frozen, read]
-    assert again.status(records) == [Status.CANCELLED] * 3  # nothing keeps them
-    assert again.failure(records) == ['the workspace restarted'] * 3
-    with pytest.raises(SubmitError, match=f'record {pending.id} cancelled'):
-        again.submit(TOTAL, params={'part': [pending.ref('value')]})
+    records = [running, frozen, read]
+    assert again.status(records) == [
+        Status.FAILED,  # it started
+        Status.CANCELLED,  # nothing keeps them
+        Status.CANCELLED,
+    ]
+    assert again.failure(records) == [
+        'the workspace ended while it ran',
+        'the workspace restarted',
+        'the workspace restarted',
+    ]
+    with pytest.raises(SubmitError, match=f'record {running.id} failed'):
+        again.submit(TOTAL, params={'part': [running.ref('value')]})
     with pytest.raises(NotReadable, match=f'accumulator {total.id} is not open'):
         again.output(total)  # the accumulator did not survive
 
@@ -418,7 +440,7 @@ def test_a_record_whose_output_is_not_kept_is_refused_at_the_push(
         total.push(part=[load.ref('value')])
 
 
-def test_the_log_holds_submissions_finished_records_accumulators_and_pushes_only(
+def test_the_log_holds_submissions_starts_finishes_accumulators_and_pushes_only(
     tmp_path: Path, start: Callable[[Path], Client]
 ) -> None:
     client = start(tmp_path / 'log')
@@ -430,6 +452,7 @@ def test_the_log_holds_submissions_finished_records_accumulators_and_pushes_only
     assert [type(e) for e in Log.read(tmp_path / 'log')] == [
         Opened,
         Submitted,
+        Started,
         Finished,
         Pushed,
     ]
@@ -506,11 +529,12 @@ def test_a_header_cut_short_is_completed_and_opens_an_empty_log(
     assert path.read_text() == json.dumps({'format': FORMAT}) + '\n'
 
 
-def test_a_log_file_of_another_format_is_refused(tmp_path: Path) -> None:
+@pytest.mark.parametrize('found', [FORMAT - 1, FORMAT + 1])
+def test_a_log_file_of_another_format_is_refused(tmp_path: Path, found: int) -> None:
     path = tmp_path / 'log'
-    path.write_text(json.dumps({'format': FORMAT + 1}) + '\n')
+    path.write_text(json.dumps({'format': found}) + '\n')
 
-    message = f'is a log of format {FORMAT + 1}; this version reads format {FORMAT}'
+    message = f'is a log of format {found}; this version reads format {FORMAT}'
     for read in (Log.read, Log):
         with pytest.raises(ValueError, match=message):
             read(path)
diff --git a/packages/essapps/tests/persist_test.py b/packages/essapps/tests/persist_test.py
index ed033a86..94de90ed 100644
--- a/packages/essapps/tests/persist_test.py
+++ b/packages/essapps/tests/persist_test.py
@@ -186,6 +186,18 @@ def loading() -> Iterator[threading.Event]:
     event.set()
 
 
+@pytest.fixture
+def loaded() -> list[float]:
+    """The run that each call of LOAD's workflow read."""
+    return []
+
+
+@pytest.fixture
+def entered() -> threading.Event:
+    """Set each time LOAD's workflow is entered, once it has read its run."""
+    return threading.Event()
+
+
 @pytest.fixture
 def boxes() -> list[weakref.ref[Box]]:
     """A weak reference to each value that BOX returned."""
@@ -193,9 +205,16 @@ def boxes() -> list[weakref.ref[Box]]:
 
 
 @pytest.fixture
-def bind(loading: threading.Event, boxes: list[weakref.ref[Box]]) -> list[Binding]:
+def bind(
+    loading: threading.Event,
+    loaded: list[float],
+    entered: threading.Event,
+    boxes: list[weakref.ref[Box]],
+) -> list[Binding]:
     def load(params: RunParams) -> Loaded:
         run = read_file(params.sample)
+        loaded.append(run)
+        entered.set()
         loading.wait(timeout=5)
         return Loaded(value=Amount(value=run), extra=Amount(value=-run))
 
@@ -463,25 +482,24 @@ def test_references_to_one_persisted_output_read_it_from_the_store_once(
 @pytest.fixture
 def start(
     files: FakeFiles, bind: list[Binding], store: GatedStore
-) -> Iterator[Callable[[Path], Client]]:
+) -> Iterator[Callable[..., Client]]:
     """
-    A client of a new workspace over the log at a path, with the same store, and
-    with ``workers`` threads.
+    A client of a new workspace over the log at a path, with the same store and
+    ``workers`` threads; closing the client closes the workspace.
     """
     workspaces: list[Workspace] = []
 
     def start(path: Path, workers: int = 1) -> Client:
-        workspaces.append(
-            Workspace(
-                proposal='p1',
-                files=files,
-                bind=bind,
-                workers=workers,
-                log=Log(path),
-                store=store,
-            )
+        workspace = Workspace(
+            proposal='p1',
+            files=files,
+            bind=bind,
+            workers=workers,
+            log=Log(path),
+            store=store,
         )
-        return Client(workspaces[-1], submitter='anna')
+        workspaces.append(workspace)
+        return Client(workspace, submitter='anna', on_close=workspace.close)
 
     yield start
     store.gate.set()
@@ -597,42 +615,53 @@ def test_readable_gives_nothing_of_a_record_that_failed_or_was_cancelled(
     loading.set()
 
 
-def test_a_restart_runs_pending_persisted_work_and_cancels_the_rest(
+def test_a_restart_fails_started_work_runs_persisted_work_and_cancels_the_rest(
     tmp_path: Path,
-    start: Callable[[Path], Client],
+    start: Callable[..., Client],
     restart: Callable[[Path], Client],
     loading: threading.Event,
+    entered: threading.Event,
 ) -> None:
-    first = start(tmp_path / 'log')
+    first = start(tmp_path / 'log', workers=1)
     done = first.compute(LOAD, params=_sample(2), persist=('value',))
     loading.clear()
-    load = first.submit(LOAD, params=_sample(1))  # read by persisted work
+    entered.clear()
+    # read by nothing persisted; holds the only worker, so the rest wait for it
+    running = first.submit(LOAD, params=_sample(2))
     alone = first.submit(LOAD, params=_sample(2))  # read by nothing persisted
+    load = first.submit(LOAD, params=_sample(1))  # read by persisted work
     total = first.submit(
         TOTAL, params={'part': [load.ref('value'), done.ref('value')]}, persist=True
     )
+    assert entered.wait(timeout=5)  # ``running`` started
 
     again = restart(tmp_path / 'log')
     loading.set()
 
-    assert again.wait([load, alone, total]) == [
-        Status.COMPLETED,
+    records = [running, alone, load, total]
+    assert again.wait(records) == [
+        Status.FAILED,
         Status.CANCELLED,
         Status.COMPLETED,
+        Status.COMPLETED,
+    ]
+    assert again.failure(records[:2]) == [
+        'the workspace ended while it ran',
+        'the workspace restarted',
     ]
-    assert again.failure(alone) == 'the workspace restarted'
     assert again.output(total, 'value') == Amount(value=3.0)
 
 
 def test_a_restart_fails_persisted_work_that_reads_what_is_gone(
     tmp_path: Path,
-    start: Callable[[Path], Client],
+    start: Callable[..., Client],
     restart: Callable[[Path], Client],
     loading: threading.Event,
 ) -> None:
-    first = start(tmp_path / 'log')
+    first = start(tmp_path / 'log', workers=1)
     kept = first.compute(LOAD, params=_sample(1))
     loading.clear()
+    first.submit(LOAD, params=_sample(1))  # holds the only worker
     pending = first.submit(LOAD, params=_sample(2))
     total = first.accumulator(Template(TOTAL, blanks=('part',)))
     total.push(part=[pending.ref('value')])  # its state waits for ``pending``
@@ -669,21 +698,23 @@ def test_a_restart_runs_a_call_through_a_stage_as_its_plain_request(
     restart: Callable[[Path], Client],
     loading: threading.Event,
 ) -> None:
-    # a second thread runs the stage while the first waits in LOAD
-    first = start(tmp_path / 'log', workers=2)
+    first = start(tmp_path / 'log', workers=1)
     kept = first.compute(LOAD, params=_sample(1))
     stored = first.compute(LOAD, params=_sample(2), persist=('value',))
-    loading.clear()
-    offset = first.submit(LOAD, params=_sample(2))
-    calls = []
+    stages = []
     for fixed in (kept, stored):
         stage = first.stage(
             Template(OFFSET, params={'source': fixed.ref('value')}, blanks=('shift',))
         )
         first.compute(stage, params={'shift': kept.ref('value')})  # stages it
-        calls.append(
-            first.submit(stage, params={'shift': offset.ref('value')}, persist=True)
-        )
+        stages.append(stage)
+    loading.clear()
+    first.submit(LOAD, params=_sample(1))  # holds the only worker
+    offset = first.submit(LOAD, params=_sample(2))
+    calls = [
+        first.submit(stage, params={'shift': offset.ref('value')}, persist=True)
+        for stage in stages
+    ]
 
     again = restart(tmp_path / 'log')
     loading.set()
@@ -695,19 +726,30 @@ def test_a_restart_runs_a_call_through_a_stage_as_its_plain_request(
     assert again.output(calls[1], 'value') == Amount(value=4.0)
 
 
-def test_a_restart_runs_again_a_record_written_but_not_finished(
-    tmp_path: Path, start: Callable[[Path], Client], store: GatedStore
+def test_a_restart_fails_a_record_that_was_running_and_does_not_run_it_again(
+    tmp_path: Path,
+    start: Callable[[Path], Client],
+    restart: Callable[[Path], Client],
+    loading: threading.Event,
+    entered: threading.Event,
+    loaded: list[float],
 ) -> None:
     first = start(tmp_path / 'log')
-    load = first.compute(LOAD, params=_sample(1), persist=('value',))
-    lines = (tmp_path / 'log').read_bytes().splitlines(keepends=True)
-    assert isinstance(Log.read(tmp_path / 'log')[-1], Finished)
-    (tmp_path / 'crashed').write_bytes(b''.join(lines[:-1]))  # before it finished
+    loading.clear()
+    load = first.submit(LOAD, params=_sample(1), persist=True)
+    shift = first.submit(SHIFT, params={'source': load.ref('value')}, persist=True)
+    assert entered.wait(timeout=5)  # ``load`` is in its workflow, ``shift`` waits
 
-    again = start(tmp_path / 'crashed')
+    again = restart(tmp_path / 'log')
 
-    assert again.wait(load) is Status.COMPLETED
-    assert again.output(load, 'value') == Amount(value=1.0)  # written again
+    assert again.status([load, shift]) == [Status.FAILED, Status.FAILED]
+    assert again.failure([load, shift]) == [
+        'the workspace ended while it ran',
+        f'the workspace restarted: record {load.id} failed',
+    ]
+    again.close()  # once every record it runs has finished
+    assert loaded == [1.0]  # by the first workspace only
+    loading.set()
 
 
 def test_a_read_of_a_persisted_output_waits_for_its_write_and_no_other_does(