diff --git a/README.md b/README.md
index 743cccf..018bc28 100644
--- a/README.md
+++ b/README.md
@@ -236,19 +236,25 @@ way Ctrl-C would, so a shell loop around it aborts.
 
 A `background` script (or `wf run -b NAME`) is detached instead: it runs
 under a supervisor of its own — the same process-group and cleanup
-handling, minus the terminal — with its output in
-`.git/workforest/logs/NAME/WORKTREE.log` of the main checkout, and `wf`
-returns as soon as it is clearly running (a command that dies right away
-is reported with the tail of its log). That way several long-running
-scripts can be started from one terminal. `wf stop NAME` stops this
-worktree's instance, foreground or background, from any terminal;
-`--all` stops every worktree's. A script runs once per worktree: starting
-one that is already running there fails (`wf stop` it first).
-
-An `exclusive` script runs at most once per project: starting it stops the
-running instance in any worktree (this one included), waits for that
-instance's cleanup to finish, and only then starts. Stopping — by `wf
-stop` or by preemption — is SIGTERM, then SIGKILL after `stop_timeout`
+handling, minus the terminal — with its output in a log file under
+`.git/workforest/logs/NAME/` of the main checkout, and `wf` returns as
+soon as it is clearly running (a command that dies right away is reported
+with the tail of its log; the path is printed when it starts). That way
+several long-running scripts can be started from one terminal. Any number
+of instances of a script may run at once, in one worktree or across
+several: each keeps a record and — in the background — a log of its own,
+named after the worktree and the pid of the `wf run` that owns it
+(`WORKTREE.PID.log`), and each runs its own `cleanup` when it ends, so a
+`cleanup` has to tolerate a sibling instance still running. `wf stop
+NAME` stops every instance in this worktree, foreground or background,
+started from any terminal; `--all` stops every worktree's. Logs of
+instances that are no longer running are removed the next time the script
+starts there. Use `exclusive` for a script that must not run twice.
+
+An `exclusive` script runs at most once per project: starting it stops
+every running instance in any worktree (this one included), waits for
+their cleanup to finish, and only then starts. Stopping — by `wf stop` or
+by preemption — is SIGTERM, then SIGKILL after `stop_timeout`
 seconds (30 by default; the top-level key changes it for every script, an
 entry's own `stop_timeout` for that one). The stopped instance's `wf run`
 reports ``killed by SIGTERM (stopped by `wf run backend` in 'feat-x')``
@@ -283,8 +289,9 @@ A group is a script like any other: `wf run -b`, `background`,
 member, waits for all their cleanups, then runs the group's own. A group
 takes no extra arguments; its `stop_timeout` defaults to the longest of
 its members'. Members stay individually visible: `wf stop MEMBER` stops
-that member out from under a running group, and a member already running
-in the worktree fails its step.
+that member out from under a running group, and a member that is already
+running elsewhere is started again — `exclusive` is what makes a member
+stop the running one instead.
 
 A pipeline's steps run like consecutive `wf run`s, each with the
 terminal; a `background` member is started and left running while the
@@ -382,7 +389,7 @@ directives for the shell function, `--porcelain`/`--json` listings, dumps).
 `list --json` describes the whole forest for programs — `main` (the main
 checkout, in the same `name`/`branch`/`path`/`dirty`/`running` shape as each
 entry of `worktrees`, `running` being the names of the scripts running
-there) and the resolved `worktrees_dir` — and is what the editor extensions
+there, each once however many instances of it run) and the resolved `worktrees_dir` — and is what the editor extensions
 read.
 
 ## JetBrains IDE plugin
diff --git a/man/workforest.1 b/man/workforest.1
index 3921d01..c278e94 100644
--- a/man/workforest.1
+++ b/man/workforest.1
@@ -204,7 +204,8 @@ carry
 (true/false),
 and
 .B running
-(the names of the scripts running there);
+(the names of the scripts running there, each once however many instances
+of it run);
 .B worktrees_dir
 is the resolved worktrees directory.
 .SS delete \fINAME\fR...
@@ -259,9 +260,9 @@ command (see
 then runs however the command ended.
 Starting an
 .B exclusive
-script first stops its running instance in any worktree of the project
-and waits for that instance's cleanup; any other script refuses to start
-while it is already running in this worktree (see
+script first stops every running instance in the project and waits for
+their cleanup; any other script simply joins the instances already
+running, in this worktree or elsewhere (see
 .BR stop ).
 The exit status is the command's, or 128+\fIN\fR when it was killed by
 signal
@@ -280,8 +281,10 @@ member's status once all have ended.
 .BR \-b ", " \-\-background
 Detach instead: the command runs under a supervisor of its own with its
 output in
-.IR .git/workforest/logs/ SCRIPT / WORKTREE .log
-of the main checkout, and
+.IR .git/workforest/logs/ SCRIPT / WORKTREE . PID .log
+of the main checkout \(em
+.I PID
+being the process that owns the instance \(em and
 .B workforest
 returns once it is clearly running (a command that dies right away is
 reported with the tail of its log).
@@ -293,7 +296,7 @@ Must precede
 after it, it belongs to
 .IR ARGS .
 .SS stop \fISCRIPT\fR [\fB\-\-all\fR]
-Stop the script's running instance in the current worktree \(em
+Stop every running instance of the script in the current worktree \(em
 foreground or background, started from any terminal \(em with SIGTERM,
 then SIGKILL after
 .B stop_timeout
@@ -304,7 +307,7 @@ under a running group.
 Fails when nothing is running.
 .TP
 .B \-\-all
-Every worktree's instance of the script, not just this one's.
+Every worktree's instances of the script, not just this one's.
 .SS tui [\fIMODE\fR]
 Interactive mode, driven by
 .BR fzf (1).
diff --git a/man/workforest.5 b/man/workforest.5
index 64254a4..d594054 100644
--- a/man/workforest.5
+++ b/man/workforest.5
@@ -191,8 +191,9 @@ Members are named, never written inline, and keep their own settings and
 records: a group may name another group (a cycle is a configuration
 error),
 .B workforest stop
-on a member stops it out from under a running group, and a member
-already running in the worktree fails its step.
+on a member stops it out from under a running group, and a member that is
+already running elsewhere is started again unless it is
+.BR exclusive .
 .TP
 .B background
 Detach by default, as
@@ -211,19 +212,23 @@ variables.
 Its failure is a warning.
 .TP
 .B exclusive
-At most one instance per project: starting the script stops the running
+At most one instance per project: starting the script stops every running
 instance in any worktree (this one included) with SIGTERM, then SIGKILL
 after
 .B stop_timeout
-seconds, waits for that instance's cleanup to finish, and only then
-starts.
+seconds, waits for their cleanup to finish, and only then starts.
+Without it any number of instances may run at once, in one worktree or
+across several, each with a record, a log, and a
+.B cleanup
+of its own.
 The stopped instance's
 .B workforest run
 reports who preempted it and exits 143.
 Running instances are recorded under
 .I .git/workforest/running/
-in the main checkout; a record is verified (pid, process group, boot)
-before anything is signalled, so a crash, a reboot, or a killed
+in the main checkout, one file per instance; a record is verified (pid,
+process group, boot) before anything is signalled, so a crash, a reboot,
+or a killed
 .B workforest
 leaves nothing to tidy by hand.
 A command still running after its
diff --git a/src/workforest/cli.py b/src/workforest/cli.py
index 126b245..fa39991 100644
--- a/src/workforest/cli.py
+++ b/src/workforest/cli.py
@@ -22,7 +22,7 @@
     "delete": "delete worktree(s)",
     "checkout": "delete a worktree and check its branch out in main",
     "run": "run a named script from the merged config",
-    "stop": "stop a running script (this worktree's instance, or --all)",
+    "stop": "stop a running script (this worktree's instances, or --all)",
     "tui": "interactive mode (requires fzf)",
     "init": "write a commented .workforest.yaml starter",
     "config": "show the merged configuration and its sources",
@@ -216,7 +216,7 @@ def opener_args(p: argparse.ArgumentParser) -> None:
     p = sub.add_parser("stop", help=SUBCOMMAND_HELP["stop"])
     p.add_argument("script", metavar="SCRIPT")
     p.add_argument(
-        "--all", action="store_true", help="every worktree's instance, not just this one's"
+        "--all", action="store_true", help="every worktree's instances, not just this one's"
     )
     p.set_defaults(func=_handle_stop)
 
diff --git a/src/workforest/config.py b/src/workforest/config.py
index 77917cd..3719855 100644
--- a/src/workforest/config.py
+++ b/src/workforest/config.py
@@ -55,7 +55,7 @@ class ScriptSpec:
     bulk: tuple[str, ...] | None = None  # members run at once; done when all are
     pipeline: tuple[str, ...] | None = None  # members run in order; stops at the first failure
     background: bool = False  # detach, with output to a log file, instead of holding the terminal
-    exclusive: bool = False  # starting it stops the running instance anywhere in the project
+    exclusive: bool = False  # starting it stops every running instance in the project
     cleanup: str | None = None  # runs after the command ends, however it ended
     stop_timeout: float | None = None  # seconds between SIGTERM and SIGKILL; None: the global one
 
diff --git a/src/workforest/examples/.workforest.yaml b/src/workforest/examples/.workforest.yaml
index 6fdfa39..9859874 100644
--- a/src/workforest/examples/.workforest.yaml
+++ b/src/workforest/examples/.workforest.yaml
@@ -67,7 +67,8 @@ setup_scripts:
 #   background: true  detached, output in a log file (`workforest run -b` does
 #                     the same once); `workforest stop NAME` ends it
 #   exclusive: true   at most one instance per project — starting it stops
-#                     the running one in any worktree (cleanup included)
+#                     the running ones in any worktree (cleanup included);
+#                     without it a script may run several times at once
 #   cleanup: CMD      runs after the command ends, however it ended
 #   stop_timeout: N   seconds between SIGTERM and SIGKILL when stopped
 #                     (default: the top-level `stop_timeout`, 30; for a
diff --git a/src/workforest/hooks.py b/src/workforest/hooks.py
index daf1b9c..4fc0a3c 100644
--- a/src/workforest/hooks.py
+++ b/src/workforest/hooks.py
@@ -257,9 +257,18 @@ class _Job:
     snippet: str
     cwd: Path
     env: dict[str, str]
-    record_path: Path
+    common_dir: Path  # where the record and the log live
     tty: bool = True
 
+    def record_path(self, pid: int) -> Path:
+        return jobs.record_path(self.common_dir, self.name, self.cwd, pid)
+
+    def log_path(self, pid: int) -> Path:
+        """The log of the instance owned by `pid`: the detached supervisor,
+        which names its own log from the inside and is named from the
+        outside by the `wf run` that forked it."""
+        return jobs.log_path(self.common_dir, self.name, self.cwd, pid)
+
 
 def _fileno(stream: int | IO[bytes]) -> int:
     return stream if isinstance(stream, int) else stream.fileno()
@@ -310,14 +319,14 @@ def _spawn(job: _Job, sink: _Sink, tty_fd: int | None) -> int:
     return pid
 
 
-def _run_command(job: _Job) -> _JobResult:
+def _run_command(job: _Job, record_path: Path) -> _JobResult:
     """Run the job in its own process group with a job record on disk for
     as long as it runs."""
     tty_fd = _controlling_tty() if job.tty else None
     with _diverted_output() as sink:
         pgid = _spawn(job, sink, tty_fd)
         jobs.write_record(
-            job.record_path,
+            record_path,
             jobs.JobRecord(
                 script=job.name,
                 worktree=str(job.cwd),
@@ -329,7 +338,7 @@ def _run_command(job: _Job) -> _JobResult:
             ),
         )
         code = _wait(pgid, tty_fd=tty_fd)
-    record = jobs.read_record(job.record_path)
+    record = jobs.read_record(record_path)
     return _JobResult(code, record.stopped_by if record else None)
 
 
@@ -343,12 +352,14 @@ def _run_cleanup(spec: ScriptSpec, name: str, *, cwd: Path, env: dict[str, str])
 
 
 def _run_to_completion(job: _Job) -> _JobResult:
-    """Command, then cleanup, then — and only then — the record goes."""
+    """Command, then cleanup, then — and only then — the record goes. We
+    own the run, so our pid names the instance."""
+    record_path = job.record_path(os.getpid())
     try:
-        result = _run_command(job)
+        result = _run_command(job, record_path)
         _run_cleanup(job.spec, job.name, cwd=job.cwd, env=job.env)
     finally:
-        job.record_path.unlink(missing_ok=True)
+        record_path.unlink(missing_ok=True)
     return result
 
 
@@ -398,7 +409,7 @@ def _exit_with(code: int) -> None:  # pragma: no cover - ends a forked child
     os._exit(code)
 
 
-def _supervise_detached(job: _Job, log_fd: int) -> None:  # pragma: no cover - the forked child
+def _supervise_detached(job: _Job) -> None:  # pragma: no cover - the forked child
     """The background supervisor: our fork, in a session of its own, with
     the log as its stdout/stderr. Runs the command exactly like the
     foreground path does and exits as `wf run` would; never returns."""
@@ -408,6 +419,7 @@ def _supervise_detached(job: _Job, log_fd: int) -> None:  # pragma: no cover - t
         devnull = os.open(os.devnull, os.O_RDONLY)
         os.dup2(devnull, 0)
         os.close(devnull)
+        log_fd = os.open(job.log_path(os.getpid()), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
         _redirect_output(log_fd, log_fd)
         os.close(log_fd)
         result = _run_to_completion(job)
@@ -421,18 +433,28 @@ def _supervise_detached(job: _Job, log_fd: int) -> None:  # pragma: no cover - t
         os._exit(code)
 
 
-def _start_background(job: _Job, log_path: Path) -> None:
+def _log_tail(log_path: Path, lines: int = 10) -> str:
+    """The last lines of a log, or "" when there is nothing to read (the
+    supervisor died before it could open one)."""
+    try:
+        text = log_path.read_text(errors="replace")
+    except OSError:
+        return ""
+    return "\n".join(text.strip().splitlines()[-lines:])
+
+
+def _start_background(job: _Job) -> None:
     """Fork a detached supervisor for the command and return once it is
     clearly running — a command that dies within the grace period is
-    reported with the tail of its log instead of failing invisibly."""
-    log_path.parent.mkdir(parents=True, exist_ok=True)
-    log_fd = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)
+    reported with the tail of its log instead of failing invisibly. The
+    supervisor writes the log; its pid is what names it."""
+    job.log_path(os.getpid()).parent.mkdir(parents=True, exist_ok=True)  # one dir per script
     sys.stdout.flush()
     sys.stderr.flush()
     pid = os.fork()
     if pid == 0:  # pragma: no cover - the forked child
-        _supervise_detached(job, log_fd)
-    os.close(log_fd)
+        _supervise_detached(job)
+    log_path = job.log_path(pid)
     deadline = time.monotonic() + _GRACE_SECONDS
     while time.monotonic() < deadline:
         reaped, status = os.waitpid(pid, os.WNOHANG)
@@ -441,10 +463,10 @@ def _start_background(job: _Job, log_path: Path) -> None:
             if code == 0:
                 output.success(f"{job.name!r} already finished (log: {log_path})")
                 return
-            tail = log_path.read_text(errors="replace").strip().splitlines()[-10:]
+            tail = _log_tail(log_path)
             message = f"script {job.name!r} exited with status {code} right after launch"
             if tail:
-                message += ":\n" + "\n".join(tail)
+                message += ":\n" + tail
             raise WorkforestError(message)
         time.sleep(0.02)
     output.success(f"started {job.name!r} in the background (pid {pid}, log: {log_path})")
@@ -478,8 +500,7 @@ def _run_step(job: _Job) -> int:
     for a death by signal N — instead of raising."""
     try:
         if job.spec.background:
-            common_dir = gitutil.git_common_dir(job.cwd)
-            _start_background(job, jobs.log_path(common_dir, job.name, job.cwd))
+            _start_background(job)
             return 0
         output.success(_start_message(job))
         result = _run_to_completion(job)
@@ -707,9 +728,9 @@ def _signal_members(
     for runner in runners:
         if runner.code is not None:
             continue
-        record = jobs.read_record(jobs.record_path(common_dir, runner.name, job.cwd))
-        if record is not None:
-            jobs.signal_group(record.pgid, signum)
+        for member in jobs.jobs_for(common_dir, runner.name):
+            if member.record.owner_pid == runner.pid:  # this runner's instance, not another
+                jobs.signal_group(member.record.pgid, signum)
 
 
 def _suspend_members(runners: list[_Runner], job: _Job) -> None:  # pragma: no cover
@@ -814,9 +835,8 @@ def _prepare(
     tty: bool = True,
 ) -> _Job:
     """Resolve a script and clear the way for it: an `exclusive` one first
-    stops its running instance anywhere in the project (that instance's
-    cleanup included); any other refuses to start while it is already
-    running in this worktree."""
+    stops every running instance in the project (their cleanup included);
+    any other simply joins the instances already running here."""
     spec = _resolve_script(config, name)
     snippet = spec.command or ""
     if extra_args:
@@ -832,18 +852,11 @@ def _prepare(
             by=f"`wf run {name}` in {cwd.name!r}",
             env=env,
         )
-    record_path = jobs.record_path(common_dir, name, cwd)
-    # One instance per script per worktree: records and logs are keyed that
-    # way, and `wf stop NAME` means "the" instance here.
-    if (existing := jobs.read_record(record_path)) is not None:
-        if jobs.classify(existing) is jobs.JobState.STALE:
-            record_path.unlink(missing_ok=True)
-        else:
-            raise WorkforestError(
-                f"{name!r} is already running in {cwd.name!r} (pid {existing.pgid}); "
-                f"`wf stop {name}` first"
-            )
-    return _Job(config, spec, name, snippet, cwd, env, record_path, tty)
+    # Any number of instances may share a worktree, each recorded and
+    # logged under the pid of the run that owns it; what dead ones left
+    # behind goes now.
+    jobs.prune(common_dir, name, cwd)
+    return _Job(config, spec, name, snippet, cwd, env, common_dir, tty)
 
 
 def run_named_script(
@@ -864,7 +877,7 @@ def run_named_script(
     """
     job = _prepare(config, name, cwd=cwd, env=env, extra_args=extra_args)
     if job.spec.background if background is None else background:
-        _start_background(job, jobs.log_path(gitutil.git_common_dir(cwd), name, cwd))
+        _start_background(job)
         return
     output.success(_start_message(job))
     _raise_for(_run_to_completion(job), name)
diff --git a/src/workforest/jobs.py b/src/workforest/jobs.py
index c071823..e65e648 100644
--- a/src/workforest/jobs.py
+++ b/src/workforest/jobs.py
@@ -2,10 +2,12 @@
 
 One JSON record per running `wf run`, under the repository's common git
 dir so every worktree of a project sees the same set:
-`<common>/workforest/running/<script>/<worktree-name>`. A record is a
-hint, never the truth: it outlives a `kill -9` or a reboot, so whoever
-reads one verifies that its processes are alive, from this boot, and
-still the group we started, and drops it otherwise.
+`<common>/workforest/running/<script>/<worktree-name>.<pid>`, the pid
+being the `wf run` that owns the instance — one worktree may hold
+several of them at once. A record is a hint, never the truth: it
+outlives a `kill -9` or a reboot, so whoever reads one verifies that its
+processes are alive, from this boot, and still the group we started, and
+drops it otherwise.
 
 Cleanup belongs to the owning `wf run`: it runs the script's `cleanup`
 once its command ends — however it ended — and only then removes its
@@ -77,14 +79,21 @@ def records_dir(common_dir: Path, script: str) -> Path:
     return common_dir / RUNNING_SUBDIR / script
 
 
-def record_path(common_dir: Path, script: str, worktree: Path) -> Path:
-    return records_dir(common_dir, script) / worktree.name
+def _instance_name(worktree: Path, pid: int) -> str:
+    """What one instance is called on disk: the worktree it runs in and the
+    pid of the `wf run` that owns it, since several instances of a script
+    may share a worktree."""
+    return f"{worktree.name}.{pid}"
 
 
-def log_path(common_dir: Path, script: str, worktree: Path) -> Path:
-    """Where a `background` run's output goes; overwritten by the next run
-    and kept afterwards, so a crash can be read up on."""
-    return common_dir / LOGS_SUBDIR / script / f"{worktree.name}.log"
+def record_path(common_dir: Path, script: str, worktree: Path, pid: int) -> Path:
+    return records_dir(common_dir, script) / _instance_name(worktree, pid)
+
+
+def log_path(common_dir: Path, script: str, worktree: Path, pid: int) -> Path:
+    """Where a `background` instance's output goes; kept after it ends, so
+    a crash can be read up on, until `prune` clears it away."""
+    return common_dir / LOGS_SUBDIR / script / f"{_instance_name(worktree, pid)}.log"
 
 
 def write_record(path: Path, record: JobRecord) -> None:
@@ -120,6 +129,30 @@ def jobs_for(common_dir: Path, script: str) -> list[Job]:
     return jobs
 
 
+def prune(common_dir: Path, script: str, worktree: Path) -> None:
+    """Forget what the script's dead instances left behind in this
+    worktree: their stale records, and the logs of every instance no
+    longer recorded. The live ones keep both."""
+    live = set()
+    for job in jobs_for(common_dir, script):
+        if Path(job.record.worktree) != worktree:
+            continue
+        if classify(job.record) is JobState.STALE:
+            job.path.unlink(missing_ok=True)
+        else:
+            live.add(job.path.name)
+    log_dir = common_dir / LOGS_SUBDIR / script
+    if not log_dir.is_dir():
+        return
+    prefix = f"{worktree.name}."
+    for path in log_dir.iterdir():
+        stem = path.stem  # "<worktree>.<pid>" for one of this worktree's instances
+        if path.suffix != ".log" or not stem.startswith(prefix):
+            continue
+        if stem[len(prefix) :].isdigit() and stem not in live:
+            path.unlink(missing_ok=True)
+
+
 def _alive(pid: int) -> bool:
     try:
         os.kill(pid, 0)
@@ -147,19 +180,20 @@ def classify(record: JobRecord) -> JobState:
 
 def running_scripts(common_dir: Path) -> dict[Path, list[str]]:
     """Which scripts are running where: worktree path → script names,
-    sorted. A record whose processes are gone (STALE) does not count; it is
-    left on disk for its owner or `wf stop` to clean up.
+    sorted, each name once however many instances of it run there. A record
+    whose processes are gone (STALE) does not count; it is left on disk for
+    its owner or `wf stop` to clean up.
     """
     directory = common_dir / RUNNING_SUBDIR
     if not directory.is_dir():
         return {}
-    found: dict[Path, list[str]] = {}
+    found: dict[Path, set[str]] = {}
     for script_dir in sorted(directory.iterdir()):
         if not script_dir.is_dir():
             continue
         for job in jobs_for(common_dir, script_dir.name):
             if classify(job.record) is not JobState.STALE:
-                found.setdefault(Path(job.record.worktree), []).append(script_dir.name)
+                found.setdefault(Path(job.record.worktree), set()).add(script_dir.name)
     return {worktree: sorted(names) for worktree, names in found.items()}
 
 
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 320a987..d9db9f1 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -289,7 +289,7 @@ def test_json_reports_running_scripts(self, repo: Repo) -> None:
         process = subprocess.Popen(["sleep", "30"], process_group=0)
         try:
             jobs.write_record(
-                jobs.record_path(common, "dev", ctx.worktrees_dir / "feat"),
+                jobs.record_path(common, "dev", ctx.worktrees_dir / "feat", os.getpid()),
                 jobs.JobRecord(
                     script="dev",
                     worktree=str(ctx.worktrees_dir / "feat"),
diff --git a/tests/test_hooks.py b/tests/test_hooks.py
index 6a32dcc..c056c5e 100644
--- a/tests/test_hooks.py
+++ b/tests/test_hooks.py
@@ -19,6 +19,13 @@
 from .conftest import Repo
 
 
+def sole_log(common: Path, script: str, worktree: Path) -> Path:
+    """The log of the only instance of `script` in `worktree`: its name
+    carries the pid of a supervisor the test never sees."""
+    [path] = sorted((common / jobs.LOGS_SUBDIR / script).glob(f"{worktree.name}.*.log"))
+    return path
+
+
 def make_worktree(repo: Repo, name: str = "feat") -> Path:
     target = repo.path.parent / "worktrees" / repo.path.name / name
     gitutil.worktree_add(repo.path, target, name)
@@ -241,7 +248,7 @@ def test_failing_cleanup_only_warns(
 
     def test_record_lives_only_while_running(self, repo: Repo) -> None:
         common = gitutil.git_common_dir(repo.path)
-        probe = f"test -f {jobs.record_path(common, 'x', repo.path)} && echo yes"
+        probe = f"test -f {jobs.record_path(common, 'x', repo.path, os.getpid())} && echo yes"
         cfg = Config(scripts={"x": ScriptSpec(probe)})
         hooks.run_named_script(cfg, "x", cwd=repo.path, env=env_for(repo, repo.path, "main"))
         assert jobs.jobs_for(common, "x") == []
@@ -375,7 +382,7 @@ def test_orphan_is_cleaned_up_by_the_preemptor(
             boot_id=jobs.boot_id(),
             started_at=time.time(),
         )
-        jobs.write_record(jobs.record_path(common, "dev", feat), record)
+        jobs.write_record(jobs.record_path(common, "dev", feat, dead.pid), record)
         monkeypatch.setattr(jobs, "process_started_before", lambda pid, when: True)
 
         hooks.run_named_script(
@@ -398,16 +405,17 @@ def test_orphan_of_a_deleted_worktree_skips_cleanup(
         sleeper = subprocess.Popen(["sleep", "30"], process_group=0)
         threading.Thread(target=sleeper.wait, daemon=True).start()
         gone = repo.path.parent / "gone"
+        owner = _reaped_pid()
         record = jobs.JobRecord(
             script="dev",
             worktree=str(gone),
             branch="gone",
             pgid=sleeper.pid,
-            owner_pid=_reaped_pid(),
+            owner_pid=owner,
             boot_id=jobs.boot_id(),
             started_at=time.time(),
         )
-        jobs.write_record(jobs.record_path(common, "dev", gone), record)
+        jobs.write_record(jobs.record_path(common, "dev", gone, owner), record)
         monkeypatch.setattr(jobs, "process_started_before", lambda pid, when: True)
 
         hooks.run_named_script(
@@ -494,12 +502,12 @@ def test_detaches_logs_and_cleans_up(
         hooks.run_named_script(cfg, "bg", cwd=repo.path, env=env_for(repo, repo.path, "main"))
         assert time.monotonic() - started < 0.5  # returned while the command still ran
 
-        log = jobs.log_path(common, "bg", repo.path)
+        [job] = jobs.jobs_for(common, "bg")
+        assert job.record.owner_pid != os.getpid()  # the supervisor owns it
+        log = jobs.log_path(common, "bg", repo.path, job.record.owner_pid)
         err = capsys.readouterr().err
         assert "started 'bg' in the background (pid " in err
         assert str(log) in err
-        [job] = jobs.jobs_for(common, "bg")
-        assert job.record.owner_pid != os.getpid()  # the supervisor owns it
         assert jobs.classify(job.record) is jobs.JobState.LIVE
 
         wait_for(lambda: not job.path.exists())
@@ -509,28 +517,6 @@ def test_detaches_logs_and_cleans_up(
         assert "err\n" in text
         assert "running cleanup for 'bg'" in text
 
-    def test_second_instance_in_the_same_worktree_is_refused(
-        self, repo: Repo, tmp_path: Path
-    ) -> None:
-        cfg = Config(scripts={"srv": ScriptSpec("sleep 30", background=True)})
-        env = env_for(repo, repo.path, "main")
-        hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
-        with pytest.raises(WorkforestError, match=r"'srv' is already running in 'api' \(pid \d+\)"):
-            hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
-        with pytest.raises(WorkforestError, match="`wf stop srv` first"):
-            hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env, background=False)
-        hooks.stop_script(cfg, "srv", cwd=repo.path, env=env)
-
-        # a stale record from a dead instance is no obstacle
-        common = gitutil.git_common_dir(repo.path)
-        jobs.write_record(
-            jobs.record_path(common, "srv", repo.path),
-            jobs.JobRecord("srv", str(repo.path), "main", _reaped_pid(), _reaped_pid(), "", 0.0),
-        )
-        hooks.run_named_script(
-            Config(scripts={"srv": ScriptSpec("true")}), "srv", cwd=repo.path, env=env
-        )
-
     def test_flag_overrides_entry(self, repo: Repo, capsys: pytest.CaptureFixture[str]) -> None:
         cfg = Config(scripts={"bg": ScriptSpec("true", background=True)})
         env = env_for(repo, repo.path, "main")
@@ -575,7 +561,80 @@ def test_stop_ends_a_background_instance(
         assert not job.path.exists()
         assert cleaned.exists()
         assert "stopping 'bg' in 'api'" in capsys.readouterr().err
-        assert "stopped by `wf stop` in 'api'" in jobs.log_path(common, "bg", repo.path).read_text()
+        log = jobs.log_path(common, "bg", repo.path, job.record.owner_pid)
+        assert "stopped by `wf stop` in 'api'" in log.read_text()
+
+
+class TestConcurrent:
+    """A script that is not `exclusive` runs as many times as it is asked
+    to, in one worktree or across several."""
+
+    def test_foreground_run_joins_a_live_instance(self, repo: Repo, tmp_path: Path) -> None:
+        cfg = Config(scripts={"srv": ScriptSpec(f"touch {tmp_path / 'ran'}")})
+        common = gitutil.git_common_dir(repo.path)
+        sleeper = subprocess.Popen(["sleep", "30"], process_group=0)
+        live = jobs.record_path(common, "srv", repo.path, os.getpid() - 1)
+        jobs.write_record(
+            live,
+            jobs.JobRecord(
+                script="srv",
+                worktree=str(repo.path),
+                branch="main",
+                pgid=sleeper.pid,
+                owner_pid=os.getpid(),
+                boot_id=jobs.boot_id(),
+                started_at=time.time(),
+            ),
+        )
+        try:
+            hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env_for(repo, repo.path, "main"))
+        finally:
+            os.killpg(sleeper.pid, signal.SIGKILL)
+            sleeper.wait()
+        assert (tmp_path / "ran").exists()
+        assert live.is_file()  # the instance already running is untouched
+
+    def test_instances_get_a_record_and_a_log_each(
+        self, repo: Repo, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        cleaned = tmp_path / "cleaned"
+        cfg = Config(
+            scripts={"srv": ScriptSpec("sleep 30", background=True, cleanup=f"echo x >> {cleaned}")}
+        )
+        env = env_for(repo, repo.path, "main")
+        common = gitutil.git_common_dir(repo.path)
+        hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
+        hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
+
+        found = jobs.jobs_for(common, "srv")
+        owners = [job.record.owner_pid for job in found]
+        assert len(set(owners)) == 2
+        assert [job.path.name for job in found] == sorted(f"api.{pid}" for pid in owners)
+        for pid in owners:
+            assert jobs.log_path(common, "srv", repo.path, pid).is_file()
+        assert capsys.readouterr().err.count("started 'srv' in the background") == 2
+
+        hooks.stop_script(cfg, "srv", cwd=repo.path, env=env)  # every instance here
+        assert jobs.jobs_for(common, "srv") == []
+        assert cleaned.read_text() == "x\nx\n"  # each instance ran its own cleanup
+
+    def test_start_clears_out_dead_instances(self, repo: Repo) -> None:
+        cfg = Config(scripts={"srv": ScriptSpec("sleep 30", background=True)})
+        env = env_for(repo, repo.path, "main")
+        common = gitutil.git_common_dir(repo.path)
+        dead = _reaped_pid()
+        jobs.write_record(
+            jobs.record_path(common, "srv", repo.path, dead),
+            jobs.JobRecord("srv", str(repo.path), "main", dead, dead, "", 0.0),
+        )
+        stale_log = jobs.log_path(common, "srv", repo.path, dead)
+        stale_log.parent.mkdir(parents=True, exist_ok=True)
+        stale_log.write_text("output of a run that is long gone")
+
+        hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
+        assert not stale_log.exists()
+        assert len(jobs.jobs_for(common, "srv")) == 1
+        hooks.stop_script(cfg, "srv", cwd=repo.path, env=env)
 
 
 class TestStop:
@@ -593,11 +652,9 @@ def test_nothing_running(self, repo: Repo) -> None:
     def test_stale_records_do_not_count(self, repo: Repo) -> None:
         cfg = Config(scripts={"x": ScriptSpec("true")})
         common = gitutil.git_common_dir(repo.path)
-        path = jobs.record_path(common, "x", repo.path)
-        jobs.write_record(
-            path,
-            jobs.JobRecord("x", str(repo.path), "main", _reaped_pid(), _reaped_pid(), "", 0.0),
-        )
+        dead = _reaped_pid()
+        path = jobs.record_path(common, "x", repo.path, dead)
+        jobs.write_record(path, jobs.JobRecord("x", str(repo.path), "main", dead, dead, "", 0.0))
         with pytest.raises(WorkforestError, match="not running"):
             hooks.stop_script(cfg, "x", cwd=repo.path, env={})
         assert not path.exists()
@@ -613,14 +670,15 @@ def test_this_worktree_only_unless_everywhere(
             process = subprocess.Popen(["sleep", "30"], process_group=0)
             threading.Thread(target=process.wait, daemon=True).start()
             sleepers.append(process)
+            owner = _reaped_pid()
             jobs.write_record(
-                jobs.record_path(common, "x", worktree),
+                jobs.record_path(common, "x", worktree, owner),
                 jobs.JobRecord(
                     "x",
                     str(worktree),
                     worktree.name,
                     process.pid,
-                    _reaped_pid(),
+                    owner,
                     jobs.boot_id(),
                     time.time(),
                 ),
@@ -654,11 +712,11 @@ def fake_stop(job: jobs.Job, *, by: str, timeout: float, orphan_cleanup: object)
         monkeypatch.setattr(jobs, "classify", lambda record: jobs.JobState.LIVE)
         env = env_for(repo, repo.path, "main")
 
-        jobs.write_record(jobs.record_path(common, "x", repo.path), record)
+        jobs.write_record(jobs.record_path(common, "x", repo.path, os.getpid()), record)
         hooks.stop_script(
             Config(scripts={"x": ScriptSpec("true")}, stop_timeout=7), "x", cwd=repo.path, env=env
         )
-        jobs.write_record(jobs.record_path(common, "x", repo.path), record)
+        jobs.write_record(jobs.record_path(common, "x", repo.path, os.getpid()), record)
         hooks.stop_script(
             Config(scripts={"x": ScriptSpec("true", stop_timeout=2)}, stop_timeout=7),
             "x",
@@ -780,7 +838,7 @@ def test_members_keep_their_own_records_while_the_group_runs(
         common = gitutil.git_common_dir(repo.path)
         out = tmp_path / "seen"
         probe = " && ".join(
-            f"test -f {jobs.record_path(common, name, repo.path)}" for name in ("m", "g")
+            f'test -n "$(ls {jobs.records_dir(common, name)})"' for name in ("m", "g")
         )
         cfg = Config(
             scripts={"m": _spec(f"{probe} && echo yes > {out}"), "g": ScriptSpec(bulk=("m",))}
@@ -789,7 +847,7 @@ def test_members_keep_their_own_records_while_the_group_runs(
         assert out.read_text() == "yes\n"
         assert jobs.jobs_for(common, "m") == [] and jobs.jobs_for(common, "g") == []
 
-    def test_member_already_running_fails_the_step(self, repo: Repo, tmp_path: Path) -> None:
+    def test_member_already_running_is_started_again(self, repo: Repo) -> None:
         cfg = Config(
             scripts={
                 "srv": _spec("sleep 30", background=True),
@@ -797,12 +855,14 @@ def test_member_already_running_fails_the_step(self, repo: Repo, tmp_path: Path)
             }
         )
         env = env_for(repo, repo.path, "main")
+        common = gitutil.git_common_dir(repo.path)
         hooks.run_named_script(cfg, "srv", cwd=repo.path, env=env)
         try:
-            with pytest.raises(WorkforestError, match="script 'chain' failed with exit code 1"):
-                hooks.run_named_script(cfg, "chain", cwd=repo.path, env=env)
+            hooks.run_named_script(cfg, "chain", cwd=repo.path, env=env)
+            assert len(jobs.jobs_for(common, "srv")) == 2
         finally:
-            hooks.stop_script(cfg, "srv", cwd=repo.path, env=env)
+            hooks.stop_script(cfg, "srv", cwd=repo.path, env=env)  # both of them
+        assert jobs.jobs_for(common, "srv") == []
 
     def test_stop_group_stops_members_and_runs_every_cleanup(
         self, repo: Repo, tmp_path: Path, capsys: pytest.CaptureFixture[str]
@@ -828,7 +888,7 @@ def test_stop_group_stops_members_and_runs_every_cleanup(
         assert cleaned.read_text().endswith("servers\n")  # the group's own cleanup comes last
         for name in ("s1", "s2", "servers"):
             assert jobs.jobs_for(common, name) == []
-        log = jobs.log_path(common, "servers", repo.path).read_text()
+        log = sole_log(common, "servers", repo.path).read_text()
         assert "s1 | Error: script 's1' was killed by SIGTERM" in log
         assert "Error: member 's2' of 'servers' was killed by SIGTERM" in log
         assert "stopped by `wf stop` in 'api'" in log
@@ -851,7 +911,7 @@ def test_stop_one_member_out_of_a_running_bulk(self, repo: Repo, tmp_path: Path)
 
         assert cleaned.read_text() == "s1\n"
         wait_for(lambda: jobs.jobs_for(common, "servers") == [])  # s2 ends on its own
-        log = jobs.log_path(common, "servers", repo.path).read_text()
+        log = sole_log(common, "servers", repo.path).read_text()
         assert (
             "s1 | Error: script 's1' was killed by SIGTERM (stopped by `wf stop` in 'api')" in log
         )
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index e1f02b7..472c941 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -72,13 +72,14 @@ def make_record(pgid: int, *, owner_pid: int | None = None, **overrides: object)
 class TestRecords:
     def test_round_trip_and_layout(self, tmp_path: Path) -> None:
         record = make_record(1234, stopped_by="other")
-        path = jobs.record_path(tmp_path, "dev", Path("/x/worktrees/api/feat"))
-        assert path == tmp_path / "workforest" / "running" / "dev" / "feat"
+        worktree = Path("/x/worktrees/api/feat")
+        path = jobs.record_path(tmp_path, "dev", worktree, 4321)
+        assert path == tmp_path / "workforest" / "running" / "dev" / "feat.4321"
         jobs.write_record(path, record)
         assert jobs.read_record(path) == record
         assert not list(path.parent.glob(".*"))  # no temp file left behind
-        assert jobs.log_path(tmp_path, "dev", Path("/x/worktrees/api/feat")) == (
-            tmp_path / "workforest" / "logs" / "dev" / "feat.log"
+        assert jobs.log_path(tmp_path, "dev", worktree, 4321) == (
+            tmp_path / "workforest" / "logs" / "dev" / "feat.4321.log"
         )
 
     def test_missing_and_corrupt(self, tmp_path: Path) -> None:
@@ -96,15 +97,47 @@ def test_jobs_for_lists_readable_records(self, tmp_path: Path) -> None:
         assert jobs.jobs_for(tmp_path, "dev") == []
         a = make_record(1)
         b = make_record(2)
-        jobs.write_record(jobs.record_path(tmp_path, "dev", Path("/w/feat-a")), a)
-        jobs.write_record(jobs.record_path(tmp_path, "dev", Path("/w/feat-b")), b)
-        (jobs.records_dir(tmp_path, "dev") / ".feat-c.tmp").write_text("partial")
-        (jobs.records_dir(tmp_path, "dev") / "feat-d").write_text("garbage")
+        jobs.write_record(jobs.record_path(tmp_path, "dev", Path("/w/feat-a"), 1), a)
+        jobs.write_record(jobs.record_path(tmp_path, "dev", Path("/w/feat-b"), 2), b)
+        (jobs.records_dir(tmp_path, "dev") / ".feat-c.1.tmp").write_text("partial")
+        (jobs.records_dir(tmp_path, "dev") / "feat-d.4").write_text("garbage")
         found = jobs.jobs_for(tmp_path, "dev")
         assert [job.record for job in found] == [a, b]
-        assert [job.path.name for job in found] == ["feat-a", "feat-b"]
+        assert [job.path.name for job in found] == ["feat-a.1", "feat-b.2"]
         assert jobs.jobs_for(tmp_path, "other") == []
 
+    def test_prune_keeps_live_instances_and_other_worktrees(
+        self, tmp_path: Path, sleeper: Callable[[str], subprocess.Popen[bytes]]
+    ) -> None:
+        worktree = Path("/w/feat")
+        live = make_record(sleeper("sleep 30").pid, worktree=str(worktree))
+        dead = make_record(_reaped_pid(), worktree=str(worktree))
+        elsewhere = make_record(_reaped_pid(), worktree="/w/other")
+        for record, name in ((live, "feat.1"), (dead, "feat.2"), (elsewhere, "other.3")):
+            jobs.write_record(jobs.records_dir(tmp_path, "dev") / name, record)
+        log_dir = tmp_path / "workforest" / "logs" / "dev"
+        log_dir.mkdir(parents=True)
+        for name in ("feat.1.log", "feat.2.log", "feat.notapid.log", "other.3.log"):
+            (log_dir / name).write_text("output")
+
+        jobs.prune(tmp_path, "dev", worktree)
+
+        assert sorted(p.name for p in jobs.records_dir(tmp_path, "dev").iterdir()) == [
+            "feat.1",
+            "other.3",
+        ]
+        # The dead instance's log goes with its record; the live one's, a
+        # file that is not an instance log, and another worktree's stay.
+        assert sorted(p.name for p in log_dir.iterdir()) == [
+            "feat.1.log",
+            "feat.notapid.log",
+            "other.3.log",
+        ]
+
+    def test_prune_without_records_or_logs(self, tmp_path: Path) -> None:
+        jobs.prune(tmp_path, "dev", Path("/w/feat"))  # nothing on disk yet
+        assert not (tmp_path / "workforest").exists()
+
     def test_boot_id_is_stable(self) -> None:
         assert jobs.boot_id() == jobs.boot_id()
 
@@ -113,15 +146,22 @@ def test_running_scripts_groups_live_records_by_worktree(
     ) -> None:
         assert jobs.running_scripts(tmp_path) == {}
         live = sleeper("sleep 30").pid
-        for script, worktree in (("dev", "/w/feat"), ("build", "/w/feat"), ("dev", "/w/other")):
+        instances = (
+            ("dev", "/w/feat", 1),
+            ("dev", "/w/feat", 2),  # a second instance adds no second name
+            ("build", "/w/feat", 3),
+            ("dev", "/w/other", 4),
+        )
+        for script, worktree, owner in instances:
             jobs.write_record(
-                jobs.record_path(tmp_path, script, Path(worktree)),
+                jobs.record_path(tmp_path, script, Path(worktree), owner),
                 make_record(live, script=script, worktree=worktree),
             )
         # A dead group: a record nobody cleaned up, which counts for nothing.
+        dead = _reaped_pid()
         jobs.write_record(
-            jobs.record_path(tmp_path, "stale", Path("/w/feat")),
-            make_record(_reaped_pid(), script="stale", worktree="/w/feat"),
+            jobs.record_path(tmp_path, "stale", Path("/w/feat"), dead),
+            make_record(dead, script="stale", worktree="/w/feat"),
         )
         (tmp_path / jobs.RUNNING_SUBDIR / "not-a-dir").write_text("")
         assert jobs.running_scripts(tmp_path) == {