fix(bridge): unmanaged-instance lifecycle bugs in start.sh + pid-file race in health cleanup

bridge/start.sh

@@ -237,6 +237,18 @@ EOF
             ;;
         *)
             # First non-option argument as project path
+            # Reject unknown flags loudly. Silently swallowing them turns a
+            # typo (or a plausible-but-nonexistent flag like `--start`) into a
+            # foreground run whose lifecycle is tied to the invoking shell —
+            # observed on daegyo where `--start` was assumed to mean a managed
+            # start.
+            case "$1" in
+                -*)
+                    echo "❌ Error: Unknown option: $1"
+                    echo "Use --help to list supported options."
+                    exit 1
+                    ;;
+            esac
             if [ -z "$PROJECT_ROOT" ]; then
                 export PROJECT_ROOT="$1"
             fi
@@ -304,6 +316,15 @@ cleanup_pid() {
     rm -f "$PID_FILE" 2>/dev/null || true
 }
 
+# PIDs of `python -m telegram_bot --path $PROJECT_ROOT` processes for THIS
+# project root, regardless of pid-file state. Covers unmanaged instances whose
+# pid file was lost (pid-file race between concurrent instances) or never
+# written — the same fallback the fleet watchdogs already use, so --status /
+# --stop and the watchdogs agree on what "running" means.
+find_project_bot_pids() {
+    pgrep -f -- "-m telegram_bot --path ${PROJECT_ROOT}( |\$)" 2>/dev/null || true
+}
+
 cleanup_supervisor_pid() {
     rm -f "$SUPERVISOR_PID_FILE" 2>/dev/null || true
 }
@@ -415,33 +436,47 @@ PY
 
 # ── Action handlers ──
 
+# Shared fallback for do_status when the pid file is missing or stale: a bot
+# process for this project may still be alive (unmanaged — e.g. its pid file
+# was deleted by a dying concurrent instance, or it was started foreground in
+# an ssh session). Report it instead of declaring the bot dead.
+report_unmanaged_or_dead() {
+    local reason="$1" live_pids
+    live_pids="$(find_project_bot_pids | tr '\n' ' ' | sed 's/ $//')"
+    if [ -n "$live_pids" ]; then
+        echo "🟡 Bot status: degraded"
+        print_component_status "Process" "alive" "unmanaged PID(s): $live_pids ($reason)"
+        print_component_status "Service" "degraded" "running without pid file; not recoverable by --status/--stop bookkeeping"
+        echo "💡 Recover: $0 --path \"$PROJECT_ROOT\" --stop && $0 --path \"$PROJECT_ROOT\" --daemon"
+        return 0
+    fi
+    echo "🔴 Bot status: unavailable"
+    print_component_status "Process" "dead" "$reason"
+    print_component_status "Service" "unavailable" "process not running"
+    print_component_status "Telegram" "unavailable" "process not running"
+    print_component_status "Claude" "unavailable" "process not running"
+}
+
 do_status() {
     local pid
     pid="$(read_pid)"
     if [ -z "$pid" ]; then
-        echo "🔴 Bot status: unavailable"
-        print_component_status "Process" "dead" "no PID file"
-        print_component_status "Service" "unavailable" "process not running"
-        print_component_status "Telegram" "unavailable" "process not running"
-        print_component_status "Claude" "unavailable" "process not running"
+        report_unmanaged_or_dead "no PID file"
         exit 0
     fi
     if kill -0 "$pid" 2>/dev/null; then
         render_status_from_health "$HEALTH_FILE" "$pid" "$HEALTH_STALE_SECONDS"
     else
-        echo "🔴 Bot status: unavailable"
-        print_component_status "Process" "dead" "stale PID: $pid"
-        print_component_status "Service" "unavailable" "process not running"
-        print_component_status "Telegram" "unavailable" "process not running"
-        print_component_status "Claude" "unavailable" "process not running"
         cleanup_pid
+        report_unmanaged_or_dead "stale PID: $pid"
     fi
     exit 0
 }
 
 do_stop() {
-    local pid supervisor_pid
+    local pid supervisor_pid upid
     local stopped_service=0
+    local unmanaged_stopped=0
     supervisor_pid="$(read_supervisor_pid)"
     pid="$(read_pid)"
 
@@ -480,10 +515,30 @@ do_stop() {
         fi
     fi
 
+    # Also stop unmanaged instances for this project root (pid file lost or
+    # never written) — otherwise --stop reports "not running" while a live
+    # bot keeps holding the Telegram token.
+    for upid in $(find_project_bot_pids); do
+        [ -n "$pid" ] && [ "$upid" = "$pid" ] && continue
+        [ -n "$supervisor_pid" ] && [ "$upid" = "$supervisor_pid" ] && continue
+        echo "🛑 Stopping unmanaged bot process (PID: $upid)..."
+        kill "$upid" 2>/dev/null || true
+        for i in $(seq 1 10); do
+            kill -0 "$upid" 2>/dev/null || break
+            sleep 1
+        done
+        if kill -0 "$upid" 2>/dev/null; then
+            echo "⚠️  Unmanaged bot process not responding to SIGTERM, sending SIGKILL..."
+            kill -9 "$upid" 2>/dev/null
+            sleep 0.5
+        fi
+        unmanaged_stopped=1
+    done
+
     cleanup_pid
     cleanup_supervisor_pid
     cleanup_token_lock_if_safe "$supervisor_pid" "$pid"
-    if [ "$stopped_service" -eq 1 ] || [ -n "$supervisor_pid" ] || [ -n "$pid" ]; then
+    if [ "$stopped_service" -eq 1 ] || [ -n "$supervisor_pid" ] || [ -n "$pid" ] || [ "$unmanaged_stopped" -eq 1 ]; then
         echo "✅ Bot stopped"
     else
         echo "⚪ Bot is not running"
@@ -1207,6 +1262,18 @@ if [ "$RUN_AS_DAEMON_SUPERVISOR" -eq 1 ]; then
     exit $?
 fi
 
+# Double-start guard for the plain foreground path (the daemon path above has
+# its own). Managed instances are caught via pid files; unmanaged ones (pid
+# file lost/never written) via the project-scoped process match — starting a
+# second instance anyway ends in a Telegram getUpdates conflict where the
+# loser's cleanup can delete the survivor's pid file.
+if [ "$INTERNAL_RUN" -eq 0 ]; then
+    if is_supervisor_running || is_running || [ -n "$(find_project_bot_pids)" ]; then
+        echo "⚠️  Bot is already running. Use --stop first."
+        exit 1
+    fi
+fi
+
 if is_token_locked; then
     echo "⚠️  Another instance is already using the same Bot Token (PID: $(cat "$TOKEN_LOCK_FILE")). Stop it first."
     exit 1

bridge/tests/test_health.py

@@ -95,6 +95,31 @@ def test_health_transitions_and_cleanup_preserve_health_file(self):
             self.assertFalse(reporter.pid_file.exists())
             self.assertFalse(lock_file.exists())
 
+    def test_cleanup_preserves_pid_file_owned_by_another_process(self):
+        """A dying instance must not delete the pid file of a concurrent
+        surviving instance (pid-file race — observed on daegyo 2026-07-08)."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            project_root = Path(tmpdir)
+            module = self._load_health_module(project_root)
+            reporter = module.RuntimeHealthReporter(project_root / ".telegram_bot")
+
+            with patch.dict(
+                os.environ,
+                {"BOT_PROCESS_MODE": "foreground", "BOT_OWNS_TOKEN_LOCK": "0"},
+                clear=False,
+            ):
+                reporter.initialize_process()
+                # Another (surviving) instance overwrites the shared pid file.
+                other_pid = os.getpid() + 1
+                reporter.pid_file.write_text(f"{other_pid}\n", encoding="utf-8")
+                reporter.cleanup_runtime_files()
+
+            self.assertTrue(reporter.pid_file.exists())
+            self.assertEqual(
+                reporter.pid_file.read_text(encoding="utf-8").strip(),
+                str(other_pid),
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

bridge/tests/test_start_status.py

@@ -207,6 +207,55 @@ def _run_interactive_start(
             "",
         )
 
+    def _spawn_unmanaged_decoy(self, project_root: Path) -> subprocess.Popen:
+        """Spawn a harmless process whose cmdline matches the project-scoped
+        bot pattern (`-m telegram_bot --path <root>`) without running the bot.
+
+        `bash -c 'sleep 30' <extra args>` exposes the extra args as $0/$@ in
+        /proc/<pid>/cmdline, which is what `pgrep -f` matches against.
+        """
+        canonical_root = project_root.resolve()
+        decoy = subprocess.Popen(
+            [
+                "bash",
+                "-c",
+                # Compound command so bash does NOT exec-replace itself with
+                # sleep (which would rewrite the visible cmdline).
+                "sleep 30; true",
+                "python",
+                "-m",
+                "telegram_bot",
+                "--path",
+                str(canonical_root),
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        self.addCleanup(decoy.kill)
+        time.sleep(0.2)
+        return decoy
+
+    def test_unknown_flag_is_rejected(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            project_root = self._prepare_project(tmpdir)
+            result = subprocess.run(
+                [
+                    "bash",
+                    str(self.start_script),
+                    "--path",
+                    str(project_root),
+                    "--start",
+                ],
+                cwd=self.repo_root,
+                text=True,
+                capture_output=True,
+                check=False,
+                env=self._hermetic_env(None),
+            )
+
+            self.assertEqual(result.returncode, 1)
+            self.assertIn("Unknown option: --start", result.stdout)
+
     def test_status_no_pid_file(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             project_root = self._prepare_project(tmpdir)
@@ -219,6 +268,35 @@ def test_status_no_pid_file(self):
             self.assertIn("Telegram: unavailable (process not running)", result.stdout)
             self.assertIn("Claude: unavailable (process not running)", result.stdout)
 
+    def test_status_reports_unmanaged_running_process(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            project_root = self._prepare_project(tmpdir)
+            decoy = self._spawn_unmanaged_decoy(project_root)
+
+            result = self._run_status(project_root)
+
+            self.assertEqual(result.returncode, 0)
+            self.assertIn("Bot status: degraded", result.stdout)
+            self.assertIn(f"unmanaged PID(s): {decoy.pid}", result.stdout)
+            self.assertIn("no PID file", result.stdout)
+            # The decoy must survive a status probe.
+            self.assertIsNone(decoy.poll())
+
+    def test_stop_stops_unmanaged_process(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            project_root = self._prepare_project(tmpdir)
+            decoy = self._spawn_unmanaged_decoy(project_root)
+
+            result = self._run_stop(project_root)
+
+            self.assertEqual(result.returncode, 0)
+            self.assertIn(
+                f"Stopping unmanaged bot process (PID: {decoy.pid})", result.stdout
+            )
+            self.assertIn("Bot stopped", result.stdout)
+            decoy.wait(timeout=15)
+            self.assertIsNotNone(decoy.poll())
+
     def test_status_stale_pid_cleans_pid_file(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             project_root = self._prepare_project(tmpdir)

bridge/utils/health.py

@@ -202,10 +202,21 @@ def snapshot(self) -> dict[str, Any]:
     def cleanup_runtime_files(self) -> None:
         with self._lock:
             self._refresh_runtime_context_locked()
+            # Only remove the pid file if it still records THIS process.
+            # Concurrent instances share one pid file: the newer instance
+            # overwrites it in initialize_process(), and when either of them
+            # exits (e.g. Telegram getUpdates conflict kills the loser), an
+            # unconditional unlink here would delete the survivor's pid file —
+            # leaving a live bot that `start.sh --status` reports as dead.
             try:
-                self._pid_file.unlink()
-            except FileNotFoundError:
-                pass
+                recorded = self._pid_file.read_text(encoding="utf-8").strip()
+            except (FileNotFoundError, OSError):
+                recorded = ""
+            if recorded == str(os.getpid()):
+                try:
+                    self._pid_file.unlink()
+                except FileNotFoundError:
+                    pass
             if self._owns_token_lock and self._token_lock_file:
                 try:
                     Path(self._token_lock_file).unlink()