Double-start guard: launching marker, probe-and-adopt, fresh-dir relaunch; skip unreachable terminals per tick

Before the launch call the planned run (run id, terminal, handle, bookkeeping) is saved as a launching
marker. If the launch call fails or agentd dies before recording the run, the next tick probes the run
directory (exit.txt / launch.pid / pid.txt): evidence -> adopt the run; nothing -> relaunch with the same
run id in <run_id>-r2 (-r3 ...); probe unreachable -> keep the marker, which holds the terminal slot.
Transports refuse to clean a run directory holding a pid file. A terminal whose relay fails is skipped
for the rest of the tick (heartbeat.down) while other terminals carry on.

agentd/executors/base.py

@@ -17,6 +17,7 @@ class Run:
     tier: Optional[str] = None
     thread_id: Optional[str] = None   # set => resume this thread
     script: Optional[str] = None      # shell executor
+    dir_name: Optional[str] = None    # run directory / task name; default run_id (`<run_id>-r2` on a relaunch)
 
 
 @dataclass
@@ -35,9 +36,15 @@ class Executor:
     def __init__(self, name, cfg, terminal, tcfg, transport):
         self.name, self.cfg, self.terminal, self.tcfg, self.transport = name, cfg, terminal, tcfg, transport
 
+    def handle(self, run):           # -> the handle `start(run)` will return, computed without touching the terminal
+        raise NotImplementedError
+
     def start(self, run):            # -> handle (dict)
         raise NotImplementedError
 
+    def probe(self, handle):         # -> names of launch evidence present on the terminal ([] = never launched)
+        raise NotImplementedError
+
     def poll(self, handle):          # -> "running" | "finished"
         raise NotImplementedError
 
@@ -84,11 +91,15 @@ class RunnerExecutor(Executor):
             program = "& " + program
         return (program + " " + rest).strip()
 
+    def handle(self, run):
+        name = run.dir_name or run.run_id
+        return {"run_dir": self.transport.join(self.run_root(), name), "task": "agentd-" + name, "thread_id": run.thread_id}
+
     def start(self, run):
         t = self.transport
-        run_dir = t.join(self.run_root(), run.run_id)
+        h = self.handle(run)
+        run_dir, task = h["run_dir"], h["task"]
         script_name, script = t.runner(run_dir, run.repo, self.command(run, run_dir), self.stdout_name)
-        task = "agentd-" + run.run_id
         files = {"prompt.md": run.prompt, script_name: script}
         if hasattr(t, "start_run"):  # remote POSIX: stage + launch in one relay round trip
             t.start_run(task, run_dir, files, script_name)
@@ -97,7 +108,13 @@ class RunnerExecutor(Executor):
             for name, text in files.items():
                 t.write_text(t.join(run_dir, name), text)
             t.launch(task, t.join(run_dir, script_name), run_dir)
-        return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id}
+        return h
+
+    PROBE_FILES = ("exit.txt", "launch.pid", "pid.txt")
+
+    def probe(self, handle):
+        """Launch evidence in the run directory (one transport call; raises TransportError when unreachable)."""
+        return self.transport.probe(handle["run_dir"], self.PROBE_FILES)
 
     def poll(self, handle):
         text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt"))