feat(bin): add concurrent bounded remote transport lanes

bin/fm-backlog-handoff.sh

@@ -549,7 +549,7 @@ remote_deliver_outbox() { # <secondmate-id> <outbox-path>
   mv -f -- "$counter_tmp" "$counter" \
     || { rm -f -- "$snapshot" "$counter_tmp"; return 1; }
   remote_rel="state/handoff/$id.outbox.md"
-  if ! "$SCRIPT_DIR/fm-on.sh" "$id" fm-remote-file.sh put "$remote_rel" 1048576 \
+  if ! "$SCRIPT_DIR/fm-on.sh" --stdin "$id" fm-remote-file.sh put "$remote_rel" 1048576 \
     "$bytes" "$hash" "$generation" < "$snapshot"; then
     rm -f -- "$snapshot"
     echo "error: handoff transfer to $id was unavailable or completion is unknown; outbox preserved at $outbox" >&2

bin/fm-on.sh

@@ -2,7 +2,7 @@
 # Execute one tracked Firstmate command in a configured remote secondmate home.
 #
 # Usage:
-#   fm-on.sh <secondmate-id|unambiguous-ssh-alias> <fm-command> [args...]
+#   fm-on.sh [--stdin] <secondmate-id|unambiguous-ssh-alias> <fm-command> [args...]
 #
 # Routes come only from remote records in data/secondmates.md. A record names an
 # SSH config alias, remote Firstmate code root, and remote FM_HOME. A host alias
@@ -11,11 +11,14 @@
 # bin/fm-*.sh namespace. No per-command table exists.
 #
 # argv is encoded as one NUL-delimited stream and passed through the fixed
-# fm-remote-entrypoint.sh. stdin remains the caller's stdin, stdout and stderr
-# remain separate, and ssh's exit status is returned unchanged. OpenSSH never
-# receives an auto-retry instruction here. Exit 255 therefore means unavailable
-# transport or unknown remote completion and must be reconciled by the semantic
-# caller, never blindly repeated by this layer.
+# fm-remote-entrypoint.sh. The remote command's stdin is /dev/null by default,
+# because remote staging captures stdin to EOF and an open caller stream would
+# block staging indefinitely; a payload caller passes --stdin to forward its
+# own stream as the job's bounded input. stdout and stderr remain separate, and
+# ssh's exit status is returned unchanged. OpenSSH never receives an auto-retry
+# instruction here. Exit 255 therefore means unavailable transport or unknown
+# remote completion and must be reconciled by the semantic caller, never
+# blindly repeated by this layer.
 #
 # The SSH alias keeps normal public-key and strict host-key policy in ~/.ssh.
 # This command explicitly disables agent forwarding, forwarding setup, and
@@ -42,12 +45,17 @@ PROTOCOL=1
 . "$SCRIPT_DIR/fm-secondmate-registry-lib.sh"
 
 die() { printf 'error: %s\n' "$1" >&2; exit 1; }
-usage() { sed -n '2,23p' "$0" | sed 's/^# \{0,1\}//'; exit 2; }
+usage() { sed -n '2,25p' "$0" | sed 's/^# \{0,1\}//'; exit 2; }
 
 encode_base64() {
   base64 | tr -d '\n'
 }
 
+STDIN_MODE=closed
+if [ "${1:-}" = --stdin ]; then
+  STDIN_MODE=caller
+  shift
+fi
 [ "$#" -ge 2 ] || usage
 ROUTE=$1
 COMMAND=$2
@@ -103,10 +111,15 @@ case "$ALIVE_COUNT_MAX" in ''|*[!0-9]*) die "FM_SSH_ALIVE_COUNT_MAX must be a po
 [ "$ALIVE_INTERVAL" -gt 0 ] || die "FM_SSH_ALIVE_INTERVAL must be a positive integer: $ALIVE_INTERVAL"
 [ "$ALIVE_COUNT_MAX" -gt 0 ] || die "FM_SSH_ALIVE_COUNT_MAX must be a positive integer: $ALIVE_COUNT_MAX"
 
-"$SSH_BIN" \
-  -o ForwardAgent=no \
-  -o ClearAllForwardings=yes \
-  -o 'SendEnv=-*' \
-  -o "ServerAliveInterval=$ALIVE_INTERVAL" \
-  -o "ServerAliveCountMax=$ALIVE_COUNT_MAX" \
+SSH_ARGS=(
+  -o ForwardAgent=no
+  -o ClearAllForwardings=yes
+  -o 'SendEnv=-*'
+  -o "ServerAliveInterval=$ALIVE_INTERVAL"
+  -o "ServerAliveCountMax=$ALIVE_COUNT_MAX"
   -- "$HOST" fm-remote-entrypoint.sh "$PROTOCOL" "$ROOT_B64" "$HOME_B64" "$ARGV_B64"
+)
+if [ "$STDIN_MODE" = caller ]; then
+  exec "$SSH_BIN" "${SSH_ARGS[@]}"
+fi
+exec "$SSH_BIN" "${SSH_ARGS[@]}" < /dev/null

bin/fm-remote-entrypoint.sh

@@ -19,6 +19,15 @@
 # disconnect remains unknown completion to fm-on.sh, which preserves OpenSSH's
 # exit 255 behavior. The shared library header owns job fields, bounds, PATH,
 # LaunchAgent contract, and worker environment.
+#
+# A staged job whose caller goes away is cancelled rather than abandoned: any
+# exit after staging and before the published result marks the job cancelled
+# (signal traps cover a delivered HUP/TERM/PIPE/INT, and the exit trap covers a
+# failed bounded wait), and while waiting this process probes its parent about
+# once per second, so an ssh channel that dies without delivering any signal -
+# sshd exiting and reparenting this process - also cancels the job. The worker
+# then skips or stops the cancelled job instead of running it to completion for
+# nobody.
 set -eu
 
 PROTOCOL=1
@@ -74,7 +83,36 @@ sha256_file() { # <path>
 [ "$#" -eq 4 ] || die "remote entrypoint expects protocol, root, home, and argv"
 [ "$1" = "$PROTOCOL" ] || die "incompatible remote protocol: local=$1 remote=$PROTOCOL"
 TMP=$(mktemp -d "${TMPDIR:-/tmp}/fm-remote-entrypoint.XXXXXX") || die "cannot create protocol staging directory" 70
-trap 'rm -rf -- "$TMP"' EXIT
+
+JOB_ID=
+JOB_COMPLETED=0
+ACCOUNT_HOME=
+ENTRYPOINT_PPID=$(ps -o ppid= -p $$ 2>/dev/null | tr -d ' ' || true)
+
+# The recorded parent is the ssh session process; when it disappears this
+# process is reparented and the caller is provably gone. An unreadable probe
+# never cancels: only an observed parent change does.
+# shellcheck disable=SC2329 # Invoked by fm_remote_job_wait through FM_REMOTE_JOB_DISCONNECT_PROBE.
+entrypoint_caller_connected() {
+  local current
+  case "$ENTRYPOINT_PPID" in ''|*[!0-9]*) return 0 ;; esac
+  current=$(ps -o ppid= -p $$ 2>/dev/null | tr -d ' ' || true)
+  case "$current" in ''|*[!0-9]*) return 0 ;; esac
+  [ "$current" = "$ENTRYPOINT_PPID" ]
+}
+
+# shellcheck disable=SC2329 # Invoked through the EXIT trap below.
+entrypoint_cleanup() {
+  rm -rf -- "$TMP"
+  if [ -n "$JOB_ID" ] && [ "$JOB_COMPLETED" -eq 0 ] && [ -n "$ACCOUNT_HOME" ]; then
+    fm_remote_job_cancel "$ACCOUNT_HOME" "$JOB_ID" 2>/dev/null || true
+  fi
+}
+trap entrypoint_cleanup EXIT
+trap 'exit 129' HUP
+trap 'exit 130' INT
+trap 'exit 141' PIPE
+trap 'exit 143' TERM
 
 decode_text "remote root" "$2" "$TMP/root"
 decode_text "remote home" "$3" "$TMP/home"
@@ -138,11 +176,14 @@ if ! fm_remote_job_ensure_worker "$ROOT" "$ACCOUNT_HOME"; then
   die "${FM_REMOTE_JOB_ERROR:-remote job worker is unavailable; run fm-on.sh <route> fm-remote-doctor.sh --fix}"
 fi
 if ! JOB_ID=$(fm_remote_job_stage "$ACCOUNT_HOME" "$ROOT" "$HOME_PATH" "$COMMAND" "${ARGV[@]:1}"); then
+  JOB_ID=
   die "${FM_REMOTE_JOB_ERROR:-cannot stage remote job}" 70
 fi
+FM_REMOTE_JOB_DISCONNECT_PROBE=entrypoint_caller_connected
 if ! fm_remote_job_wait "$ACCOUNT_HOME" "$JOB_ID"; then
   die "${FM_REMOTE_JOB_ERROR:-remote job did not complete}" 70
 fi
+JOB_COMPLETED=1
 cat "$FM_REMOTE_JOB_STDOUT"
 cat "$FM_REMOTE_JOB_STDERR" >&2
 RESULT=$FM_REMOTE_JOB_EXIT

bin/fm-remote-home-seed.sh

@@ -242,7 +242,7 @@ if [ "$PREFLIGHT_RC" -ne 0 ]; then
 fi
 
 set +e
-PROVISION_OUT=$("$SCRIPT_DIR/fm-on.sh" "$ID" fm-remote-home-provision.sh < "$TMP/manifest" 2>&1)
+PROVISION_OUT=$("$SCRIPT_DIR/fm-on.sh" --stdin "$ID" fm-remote-home-provision.sh < "$TMP/manifest" 2>&1)
 PROVISION_RC=$?
 set -e
 if [ "$PROVISION_RC" -ne 0 ]; then

bin/fm-remote-inherit-push.sh

@@ -80,7 +80,7 @@ while IFS= read -r rel; do
     [ -f "$snapshot" ] && [ ! -L "$snapshot" ] || die "inherited source snapshot is unsafe: $source"
     bytes=$(LC_ALL=C wc -c < "$snapshot" | tr -d ' ')
     hash=$(sha256_file "$snapshot") || die "cannot hash inherited source: $source"
-    "$SCRIPT_DIR/fm-on.sh" "$ID" fm-remote-inherit.sh put "$rel" "$bytes" "$hash" "$GENERATION" < "$snapshot"
+    "$SCRIPT_DIR/fm-on.sh" --stdin "$ID" fm-remote-inherit.sh put "$rel" "$bytes" "$hash" "$GENERATION" < "$snapshot"
   else
     # This loop's heredoc is its control stream, not remote command input.
     "$SCRIPT_DIR/fm-on.sh" "$ID" fm-remote-inherit.sh absent "$rel" 0 "$EMPTY_HASH" "$GENERATION" < /dev/null