Before: main 25bf84ab07
```
# Before: main 25bf84ab07, archive runner 0.1.8-nightly.20261107.4314, dash as sh, throwaway HOME
# two launches on one state key at once, then a third connect
--- outputs
{"remotePort":3774,"serverKind":"managed"}
{"remotePort":3774,"serverKind":"managed"}
a exit 0
b exit 0
--- record
managed=managed port=3774 pid=777699
record pid 777699 NOT running
--- server-runtime.json
pid 777659 port 3774
--- server.log errors
1
Listening on http://127.0.0.1:3774
--- next connect
{"remotePort":3774,"serverKind":"external"}
--- record after next
managed=external port=3774 pid=
# Before, variant: runtime pre-installed
--- outputs
{"remotePort":3774,"serverKind":"managed"}
{"remotePort":3774,"serverKind":"managed"}
a exit 0
b exit 0
--- record
managed=managed port=3774 pid=781350
record pid 781350 NOT running
--- server-runtime.json
pid 781344 port 3774
--- server.log errors
1
Listening on http://127.0.0.1:3774
--- next connect
{"remotePort":3774,"serverKind":"external"}
--- record after next
managed=external port=3774 pid=
# Before, variant: two state keys sharing one home, main 25bf84ab07
{"remotePort":3774,"serverKind":"managed"} (both)
k1: managed=managed port=3774 pid=783769 running
k2: managed=managed port=3774 pid=783787 NOT-running
server-runtime.json pid 783769 port 3774
k2/server.log: ServerAlreadyRunningError
```
After: this branch
```
# After: fix/issue-32549 6016bfa6a4, archive runner 0.1.8-nightly.20261107.4314, dash as sh, throwaway HOME
# two launches on one state key at once, then a third connect
--- outputs
{"remotePort":3774,"serverKind":"managed"}
{"remotePort":3774,"serverKind":"managed"}
a exit 0
b exit 0
--- record
managed=managed port=3774 pid=1387604
record pid 1387604 running
--- server-runtime.json
pid 1387604 port 3774
--- server.log errors
0
Listening on http://127.0.0.1:3774
--- next connect
{"remotePort":3774,"serverKind":"managed"}
--- record after next
managed=managed port=3774 pid=1387604
# After, variant: two state keys (two aliases) sharing one home
--- outputs
k1: {"remotePort":3774,"serverKind":"external"}
k2: {"remotePort":3774,"serverKind":"managed"}
k1 exit 0
k2 exit 0
k1 record: managed=external port=3774 pid=
k2 record: managed=managed port=3774 pid=1391813
server-runtime.json pid 1391813 port 3774
ServerAlreadyRunningError count: 0
lock file left: no
# Lock recovery: three launches after a killed one (B pauses inside recovery, A takes over, C arrives)
B judged stale; paused before recovery
ENTER role=A pid=1380003 lock=1380003
B holds recovery lock; paused before recheck
OK: while A (1380003) holds the lock, B and C wait; lock=1380003
EXIT role=A
ENTER role=C pid=1380015 lock=1380015
EXIT role=C
ENTER role=B pid=1379999 lock=1379999
EXIT role=B
```
Driver: concurrent launch repro (run with a rendered launch script)
```sh
#!/bin/sh
# usage: repro.sh [prewarm-home]
L="$1"; R="$2"; mkdir -p "$R"
export HOME="$R/home"; mkdir -p "$HOME"
if [ -n "${3:-}" ]; then mkdir -p "$HOME/.t3/runtime"; cp -a "$3/.t3/runtime/versions" "$HOME/.t3/runtime/" ; fi
date +%T.%N > "$R/start"
sh -l -s -- k < "$L" > "$R/a.out" 2> "$R/a.err" & A=$!
sh -l -s -- k < "$L" > "$R/b.out" 2> "$R/b.err" & B=$!
wait $A; echo "a exit $?" >> "$R/exits"; wait $B; echo "b exit $?" >> "$R/exits"
{ echo "--- outputs"; cat "$R/a.out" "$R/b.out"; cat "$R/exits";
echo "--- record"; for f in managed port pid; do printf '%s=%s ' $f "$(cat $HOME/.t3/ssh-launch/k/$f 2>/dev/null)"; done; echo
P=$(cat $HOME/.t3/ssh-launch/k/pid 2>/dev/null); kill -0 "$P" 2>/dev/null && echo "record pid $P running" || echo "record pid $P NOT running"
echo "--- server-runtime.json"; python3 -c 'import json,sys; d=json.load(open(sys.argv[1])); print("pid",d["pid"],"port",d["port"])' "$HOME/.t3/userdata/server-runtime.json"
echo "--- server.log errors"; grep -c "ServerAlreadyRunningError" "$HOME/.t3/ssh-launch/k/server.log"; grep -o "Listening on [^ ]*" "$HOME/.t3/ssh-launch/k/server.log"
echo "--- next connect"; sh -l -s -- k < "$L" 2>&1 | tail -1
echo "--- record after next"; for f in managed port pid; do printf '%s=%s ' $f "$(cat $HOME/.t3/ssh-launch/k/$f 2>/dev/null)"; done; echo
} > "$R/summary.txt" 2>&1
cat "$R/summary.txt"
```
Driver: render REMOTE_LAUNCH_SCRIPT via launchOrReuseRemoteServer
```js
import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
import * as NodeServices from "@effect/platform-node/NodeServices";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Stream from "effect/Stream";
import * as Sink from "effect/Sink";
import { ChildProcessSpawner } from "effect/process";
import * as fs from "node:fs";
const tunnelPath = process.argv[2], out = process.argv[3], version = process.argv[4];
const SshTunnel = await import(tunnelPath);
let script = "";
const spawner = ChildProcessSpawner.make((command) => Effect.gen(function* () {
const s = command.options?.stdin?.stream ?? command.options?.stdin;
if (s && typeof s === "object") {
const chunks = yield* Stream.runCollect(s);
script = Array.from(chunks).map((c) => typeof c === "string" ? c : new TextDecoder().decode(c)).join("");
}
const so = Stream.make(new TextEncoder().encode('{"remotePort":1}\n'));
return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(1), stdout: so, stderr: Stream.empty, all: so,
exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)), isRunning: Effect.succeed(false), kill: () => Effect.void,
stdin: Sink.drain, getInputFd: () => Sink.drain, getOutputFd: () => Stream.empty, unref: Effect.succeed(Effect.void) });
}));
const layer = Layer.merge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner));
await Effect.runPromise(SshTunnel.launchOrReuseRemoteServer({ alias: "devbox", hostname: "devbox", username: "u", port: 22 }, undefined, { archiveVersion: version }).pipe(Effect.provide(layer)));
fs.writeFileSync(out, script);
console.log("wrote", out, script.length);
```
Driver: lock-recovery interleaving harness
```python
import datetime, os, pathlib, select, subprocess, time, sys
root = pathlib.Path('/tmp/issue-lead-32549-r1/race') / datetime.datetime.now(datetime.timezone.utc).strftime('run-%Y%m%dT%H%M%S%fZ')
root.mkdir()
source = pathlib.Path(sys.argv[1]).read_text()
helper = source[source.index('discard_stale_launch_lock() {'):source.index('# Concurrent launches share')]
helper = helper.replace('discard_stale_launch_lock() {\n', 'discard_stale_launch_lock() {\n pause_b "B judged stale; paused before recovery" b-before\n', 1)
recheck = ' if [ "$(cat "$LAUNCH_LOCK" 2>/dev/null || true)" = "$1" ]; then'
assert recheck in helper
helper = helper.replace(recheck, ' pause_b "B holds recovery lock; paused before recheck" b-after\n' + recheck, 1).replace('\\\\n', '\\n')
loop = source[source.index('LAUNCH_LOCK_WAITED=0'):source.index('RUNNER_CHANGED=0', source.index('LAUNCH_LOCK_WAITED=0'))].replace('@@T3_LAUNCH_LOCK_WAIT_SECONDS@@', '82').replace('\\\\n', '\\n')
script = '''set -eu
LAUNCH_LOCK="$CONTROL/lock"
LAUNCH_BREAK_LOCK="$LAUNCH_LOCK.break"
cleanup() {
for OWNED_LOCK in "$LAUNCH_LOCK" "$LAUNCH_BREAK_LOCK"; do
if [ "$(cat "$OWNED_LOCK" 2>/dev/null || true)" = "$$" ]; then rm -f "$OWNED_LOCK"; fi
done
}
trap cleanup EXIT
trap 'exit 1' HUP INT TERM
B_PAUSES=0
pause_b() {
[ "$ROLE" = B ] || return 0
[ -e "$CONTROL/$2.done" ] && return 0
printf '%s\\n' "$1"; read -r _ < "$CONTROL/$2"; : > "$CONTROL/$2.done"
}
''' + helper + loop + '''printf 'ENTER role=%s pid=%s lock=%s\\n' "$ROLE" "$$" "$(cat "$LAUNCH_LOCK")"
read -r _ < "$CONTROL/release.$ROLE"
printf 'EXIT role=%s\\n' "$ROLE"
'''
(root/'lock.sh').write_text(script)
(root/'lock').write_text('999999999\n')
for n in ('b-before','b-after','release.A','release.B','release.C'): os.mkfifo(root/n)
procs=[]; transcript=[]
def start(role):
p=subprocess.Popen(['sh',str(root/'lock.sh')],env={**os.environ,'CONTROL':str(root),'ROLE':role},stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True); procs.append(p); return p
def line(p, t=8):
if not select.select([p.stdout],[],[],t)[0]: return None
r=p.stdout.readline().strip(); transcript.append(r); print(r,flush=True); return r
def release(n):
fd=os.open(root/n,os.O_WRONLY|os.O_NONBLOCK); os.write(fd,b'go\n'); os.close(fd)
try:
b=start('B'); assert 'judged stale' in line(b)
a=start('A'); assert 'ENTER role=A' in line(a)
release('b-before'); assert 'paused before recheck' in line(b)
c=start('C')
release('b-after')
late=[line(p,3) for p in (b,c)]
assert late==[None,None], late
owner=(root/'lock').read_text().strip()
r=f'OK: while A ({a.pid}) holds the lock, B and C wait; lock={owner}'; assert owner==str(a.pid); transcript.append(r); print(r,flush=True)
release('release.A'); line(a); a.wait(timeout=5)
first=None
for _ in range(2):
ready=select.select([b.stdout,c.stdout],[],[],5)[0]; assert ready
p = b if b.stdout in ready else c; r=line(p); assert 'ENTER' in r
other = c if p is b else b
assert line(other,2) is None, 'second entrant while first holds lock'
release('release.'+('B' if p is b else 'C')); line(p); p.wait(timeout=5)
b, c = (other, other) if False else (b, c)
if first is None: first=p
else: break
# wait for the other one now
r=line(other,5); assert r and 'ENTER' in r
release('release.'+('B' if other is procs[0] else 'C')); line(other); other.wait(timeout=5); break
print('DONE serialized', flush=True)
finally:
for p in procs:
if p.poll() is None: p.terminate()
for p in procs:
try: p.wait(timeout=5)
except subprocess.TimeoutExpired: p.kill(); p.wait()
(root/'transcript.txt').write_text('\n'.join(transcript)+'\n'); print('artifact='+str(root))
```