fix(#1297 review): real socket probe, honest placement abort, observable broker state

rev-security-03 REQUEST_CHANGES (review note on the brain), all four
findings verified by measurement before fixing:

F1 BLOCKER — brokerSocketPresent used access(path, S_IFSOCK).
S_IFSOCK (0xC000) is a file-type constant, not an access() mode (0-7):
on node 24.18.0 the call throws ERR_OUT_OF_RANGE, so the probe could
NEVER return true and the swallow-catch made every un-seamed call report
the broker absent — on this host (roster v1) 'mosaic fleet start' would
have refused broker-absent forever. Now stat().isSocket(), matching the
bash side's [ -S ]. New spec exercises the REAL probe against a REAL
unix socket (true), a regular file (false), and an absent path (false) —
no seam, so no seam can hide this again.

F2 — placeUnitFile's single catch conflated destination-absent with
unlink-FAILED; on unlink failure it copied through the still-live
symlink (copy-through overwrite measured by the reviewer). Now: ENOENT
is the only swallowed lstat outcome; unlink failure aborts with a named
UnitPlacementError BEFORE any copy. New spec proves the residue target's
bytes survive an unlink failure and the destination link is untouched.

F3 — observeBroker defaulted unit/socket to false when seams unset, and
production injects none: plan/status/doctor reported a healthy broker as
absent. Seams still take precedence; with no seam the real stat()
probes run. The v2 start path (and apply-with-running) now re-check the
socket after enable+start with a NAMED lifecycle-precondition-failed
refusal — previously only the v1 path had that protection, and the
refusal was masked into the generic recoverable result. Acceptance
fixtures gain hermetic brokerSocketEnv paths (the old fixture asserted
the lying default).

F4 (note) — unlink race stays inside the user-owned dir; no action.

Local gates on this tree: vitest 1566/1566, typecheck, lint, prettier
3.8.1, verify-sanitized all pass. CI 2468's sanitization failure did not
reproduce locally on the identical tree; watching the fresh pipeline.

packages/mosaic/src/commands/fleet.spec.ts

@@ -1,11 +1,24 @@
-import { chmod, lstat, mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises';
+import {
+  chmod,
+  lstat,
+  mkdir,
+  mkdtemp,
+  readFile,
+  readlink,
+  rm,
+  stat,
+  symlink,
+  writeFile,
+} from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { dirname, join, resolve } from 'node:path';
+import { createServer } from 'node:net';
 import { Command } from 'commander';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import {
   acquireRestartLock,
   addAgentToRoster,
+  brokerSocketPresent,
   buildAgentSendCommand,
   buildAgentWatchAttachCommand,
   buildAgentWatchCommand,
@@ -42,6 +55,7 @@ import {
   parseSystemdShow,
   parseTmuxListPanes,
   parseTmuxListSessions,
+  placeUnitFile,
   registerFleetCommand,
   removeAgentFromRoster,
   resolveFleetPaths,
@@ -50,6 +64,7 @@ import {
   RESTART_LOCK_STALE_MS,
   RUNTIME_ACCEPTABLE_COMMANDS,
   serializeRosterToYaml,
+  UnitPlacementError,
   VERIFY_DEFAULT_TIMEOUT_MS,
   VERIFY_POLL_INTERVAL_MS,
   type AgentPsRow,
@@ -4471,3 +4486,59 @@ describe('fleet ps — heartbeat path resolution', () => {
     );
   });
 });
+
+describe('#1297 review: the real broker probe, exercised without any seam', () => {
+  it('brokerSocketPresent answers a REAL unix socket via stat().isSocket() (access(S_IFSOCK) threw ERR_OUT_OF_RANGE)', async () => {
+    const dir = await tempDir();
+    const sockPath = join(dir, 'broker.sock');
+    const server = createServer();
+    await new Promise<void>((resolve) => {
+      server.listen(sockPath, resolve);
+    });
+    try {
+      // A live unix socket answers true through the REAL probe — no seam.
+      expect(await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: sockPath })).toBe(true);
+      // Discrimination is by file type: a regular file that EXISTS is not a
+      // socket. The old implementation could not reach either verdict — it
+      // threw ERR_OUT_OF_RANGE (node >= 24) and the catch answered false.
+      const notASocket = join(dir, 'not-a-sock');
+      await writeFile(notASocket, 'x');
+      expect(await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: notASocket })).toBe(false);
+      // Absent path: false, not a throw.
+      expect(
+        await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: join(dir, 'gone.sock') }),
+      ).toBe(false);
+    } finally {
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+    }
+    await rm(dir, { recursive: true, force: true });
+  });
+
+  it('placeUnitFile aborts with UnitPlacementError when unlink fails — never copies through a live symlink', async () => {
+    const dir = await tempDir();
+    const unitDir = join(dir, 'systemd', 'user');
+    await mkdir(unitDir, { recursive: true });
+    // By-path residue: destination is a symlink pointing somewhere else.
+    const residueTarget = join(dir, 'residue-target');
+    await writeFile(residueTarget, 'RESIDUE-BYTES');
+    const destination = join(unitDir, 'x.service');
+    await symlink(residueTarget, destination);
+    const source = join(dir, 'seed.service');
+    await writeFile(source, 'UNIT-BYTES');
+    // Read-only unit dir: unlink now fails EACCES (test runs as the owner,
+    // not root, so mode bits are enforced).
+    await chmod(unitDir, 0o500);
+    try {
+      await expect(placeUnitFile(source, unitDir, 'x.service')).rejects.toThrow(UnitPlacementError);
+    } finally {
+      await chmod(unitDir, 0o700);
+    }
+    // The copy-through never happened: residue bytes intact, destination
+    // still the symlink (abort, not overwrite-through).
+    expect(await readFile(residueTarget, 'utf8')).toBe('RESIDUE-BYTES');
+    expect(await readlink(destination)).toBe(residueTarget);
+    await rm(dir, { recursive: true, force: true });
+  });
+});