fix(fanout): stop, force and rogue-kill must account for the daemon

Three places where the panel and the fanout daemon disagreed about who
owns a stream's encoder. The first is a showstopper; the second is a
self-inflicted outage; the third is silent.

* stopStream could not stop a supervised stream. It killed the encoder
  pid -- which is precisely the event the daemon's supervisor exists to
  react to, so the daemon started a replacement and the stream came
  straight back. It now releases supervision FIRST, so the daemon lets go
  and kills the encoder itself. (The existing FanoutClient::unregister
  call drops the ingest registration, which is a different thing.)

* The rogue-ffmpeg sweep in cron:streams kills every ffmpeg writing an
  .m3u8 that is not on its active-pid list. That list is built from the
  database pid, read at the top of the pass -- so an encoder the daemon
  restarted a moment later was not on it, and the sweep would shoot a
  perfectly healthy supervised stream and log it as a rogue. The daemon's
  own reported pid is now added to the list.

* force_stream wrote a `<id>.force` signal file, which only
  MonitorCommand ever read -- and it stands down for supervised streams.
  Forcing a source therefore did nothing at all. It now calls the daemon
  and falls back to the file when the daemon does not hold the stream or
  cannot be reached.

All three degrade to exactly the previous behaviour when the daemon is
unreachable.

Verified: php -l clean on the changed files and across all of src/.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/Cli/CronJobs/StreamsCronJob.php

@@ -53,12 +53,12 @@ class StreamsCronJob implements CommandInterface {
      *
      * @param object $db      Database handle.
      * @param array  $rStream The `streams_servers` row being reconciled.
-     * @return void
+     * @return int The encoder pid the daemon reports, or 0.
      */
-    private function reconcileSupervisedStream($db, array $rStream): void {
+    private function reconcileSupervisedStream($db, array $rStream): int {
         $rState = FanoutClient::monitorState((int) $rStream['stream_id']);
         if (!is_array($rState)) {
-            return; // daemon unreachable, or it no longer holds this stream
+            return 0; // daemon unreachable, or it no longer holds this stream
         }
 
         // stream_status: 0 = running, 1 = failed. The daemon knows which, and it
@@ -99,6 +99,8 @@ class StreamsCronJob implements CommandInterface {
         $db->query('UPDATE `streams_servers` SET ' . implode(', ', $rSets) . ' WHERE `server_stream_id` = ?', ...$rArgs);
 
         echo 'Supervised by daemon (pid ' . $rPID . ', ' . ($rStatus === 0 ? 'running' : 'failed') . ")\n";
+
+        return $rPID;
     }
 
     public function execute(array $rArgs): int {
@@ -162,7 +164,17 @@ class StreamsCronJob implements CommandInterface {
                 // stream; ask who before concluding nobody is.
                 $rDaemonMonitored = isset($rSupervisedSet[(string) $rStream['stream_id']]);
                 if ($rDaemonMonitored) {
-                    $this->reconcileSupervisedStream($db, $rStream);
+                    // Record the daemon's own idea of the live pid before anything
+                    // else runs. The rogue-ffmpeg sweep at the end of this pass kills
+                    // every ffmpeg writing an .m3u8 that is not on the active list,
+                    // and an encoder the daemon restarted a moment ago would not
+                    // otherwise be on it — the database pid was read before that
+                    // restart. Shooting it would be a self-inflicted outage, logged
+                    // as having killed a "rogue" process.
+                    $rDaemonPID = $this->reconcileSupervisedStream($db, $rStream);
+                    if ($rDaemonPID > 0) {
+                        $rActivePIDs[] = $rDaemonPID;
+                    }
                 }
 
                 if ($rDaemonMonitored || ProcessManager::isMonitorAlive($rStream['monitor_pid'], $rStream['stream_id']) || $rStream['on_demand']) {

src/Domain/Stream/StreamProcess.php

@@ -1063,6 +1063,13 @@ class StreamProcess {
 	 * @return mixed Stop result.
 	 */
 	public static function stopStream($rStreamID, $rStop = false) {
+		// Stop the SUPERVISOR before killing anything. If the fanout daemon is
+		// watching this stream's encoder, killing the process is exactly the event
+		// it exists to react to — it would start a replacement and the stream
+		// would refuse to stop. Releasing first makes the daemon let go and kill
+		// the encoder itself; a no-op when it is not supervising or not reachable.
+		FanoutClient::releaseSupervision(intval($rStreamID));
+
 		$rMonitor = self::pidFromFileOrColumn($rStreamID, 'monitor_pid', '_.monitor');
 
 		if (0 < $rMonitor && \XcVm\Streaming\Health\ProcessChecker::checkPID($rMonitor, array('XC_VM[' . $rStreamID . ']')) && is_numeric($rMonitor)) {

src/Public/Controllers/Api/InternalApiController.php

@@ -227,7 +227,14 @@ class InternalApiController {
 				$rForceID = intval($rRequest['force_id']);
 
 				if ($rStreamID > 0) {
-					file_put_contents(SIGNALS_TMP_PATH . $rStreamID . '.force', $rForceID);
+					// A supervised stream's watchdog is the fanout daemon, and it does
+					// not read this signal file — MonitorCommand, which did, has stood
+					// down for that stream. Without the control call, forcing a source
+					// would silently do nothing at all. Falls back to the file whenever
+					// the daemon does not hold the stream, or cannot be reached.
+					if (!FanoutClient::forceSource($rStreamID, $rForceID)) {
+						file_put_contents(SIGNALS_TMP_PATH . $rStreamID . '.force', $rForceID);
+					}
 				}
 
 				exit(json_encode(array('result' => true)));