fix(observability): gate global logs to admins, scope to managed containers, harden SSE (#1254)

* fix(observability): gate global logs to admins, scope to managed containers, harden SSE

Make the Logs feed an administrator view enforced on both sides (requireAdmin on
the /api/logs/global poll and SSE routes; the Logs nav item plus a redirect guard
on the frontend), and scope the feed to Sencho-managed containers only via a
shared isManagedByComposeDir helper that /stats now reuses.

Harden the SSE stream: a stateful frame demuxer that survives chunk boundaries so
a Docker frame split across reads is reassembled instead of dropped or garbled; a
per-stream error listener so one broken follow stream cannot crash the event loop
(it posts a single degraded notice and keeps the others alive); a cap on
concurrent follow streams with a truncation notice; a bounded initial tail; and
backpressure that pauses the source streams when the client is slow and resumes on
drain. Bound the polling snapshot's per-container fan-out with a concurrency limit.

Add process-local, in-memory log-stream counters exposed at the admin-only
/api/system/log-stream-metrics endpoint (active connections, lines streamed,
attach and frame errors). Collapse the view to the local hub and remove the dead
remote-node handling.

* fix(observability): close remote-proxy bypass of the global-logs admin gate

The logs feed's requireAdmin lives in the local route handler, which the remote
proxy skips when forwarding a request whose nodeId targets a remote node. A hub
user could therefore request /api/logs/global*, /api/logs/global/stream, or
/api/system/log-stream-metrics with x-node-id (or ?nodeId= for the SSE transport)
pointing at a remote node and have it served as the node-proxy admin on the far
side, sidestepping the gate entirely.

Add these paths to HUB_ONLY_PREFIXES so hubOnlyGuard rejects a remote nodeId with
403 before the proxy runs, matching the existing protection on audit-log,
scheduled-tasks, and notification-routes. Add regression tests covering the
collection path, the SSE sub-path (both the x-node-id header and the ?nodeId=
query transport), and the stream-metrics endpoint.

backend/src/__tests__/global-logs-metrics.test.ts

@@ -0,0 +1,71 @@
+import { describe, it, expect, beforeEach } from 'vitest';
+import { GlobalLogsMetrics } from '../services/GlobalLogsMetrics';
+import { mapWithConcurrency } from '../routes/metrics';
+
+describe('GlobalLogsMetrics', () => {
+  beforeEach(() => GlobalLogsMetrics.resetForTests());
+
+  it('starts at all-zero', () => {
+    const s = GlobalLogsMetrics.snapshot();
+    expect(s.active_sse_connections).toBe(0);
+    expect(s.sse_connections_total).toBe(0);
+    expect(s.lines_streamed_total).toBe(0);
+  });
+
+  it('openConnection bumps both the gauge and the total; closeConnection drops only the gauge', () => {
+    GlobalLogsMetrics.openConnection();
+    GlobalLogsMetrics.openConnection();
+    expect(GlobalLogsMetrics.snapshot().active_sse_connections).toBe(2);
+    expect(GlobalLogsMetrics.snapshot().sse_connections_total).toBe(2);
+
+    GlobalLogsMetrics.closeConnection();
+    expect(GlobalLogsMetrics.snapshot().active_sse_connections).toBe(1);
+    // The monotonic total is untouched by a close.
+    expect(GlobalLogsMetrics.snapshot().sse_connections_total).toBe(2);
+  });
+
+  it('clamps the active gauge at zero on an unbalanced close', () => {
+    GlobalLogsMetrics.closeConnection();
+    GlobalLogsMetrics.closeConnection();
+    expect(GlobalLogsMetrics.snapshot().active_sse_connections).toBe(0);
+  });
+
+  it('increments a monotonic counter by an explicit amount', () => {
+    GlobalLogsMetrics.increment('lines_streamed_total', 5);
+    GlobalLogsMetrics.increment('lines_streamed_total');
+    expect(GlobalLogsMetrics.snapshot().lines_streamed_total).toBe(6);
+  });
+
+  it('snapshot returns a copy, not a live reference', () => {
+    const s = GlobalLogsMetrics.snapshot();
+    s.lines_streamed_total = 999;
+    expect(GlobalLogsMetrics.snapshot().lines_streamed_total).toBe(0);
+  });
+});
+
+describe('mapWithConcurrency', () => {
+  it('runs every item exactly once', async () => {
+    const seen: number[] = [];
+    await mapWithConcurrency([1, 2, 3, 4, 5], 2, async (n) => { seen.push(n); });
+    expect(seen.sort((a, b) => a - b)).toEqual([1, 2, 3, 4, 5]);
+  });
+
+  it('never exceeds the concurrency limit in flight', async () => {
+    let inFlight = 0;
+    let peak = 0;
+    await mapWithConcurrency([...Array(20).keys()], 4, async () => {
+      inFlight += 1;
+      peak = Math.max(peak, inFlight);
+      await new Promise(r => setTimeout(r, 2));
+      inFlight -= 1;
+    });
+    expect(peak).toBeLessThanOrEqual(4);
+    expect(peak).toBeGreaterThan(1); // proves it actually parallelizes
+  });
+
+  it('is a no-op on an empty array', async () => {
+    let calls = 0;
+    await mapWithConcurrency([], 4, async () => { calls += 1; });
+    expect(calls).toBe(0);
+  });
+});

backend/src/__tests__/hub-only-guard.test.ts

@@ -120,6 +120,51 @@ describe('hubOnlyGuard', () => {
     expect(res.body?.code).toBe('HUB_ONLY_ENDPOINT');
   });
 
+  // Regression for the Global Observability admin gate: the logs feed's
+  // `requireAdmin` lives in the local route handler, which the proxy skips when
+  // forwarding a remote nodeId. Without these prefixes the guard would let the
+  // request through to the proxy and a hub user could read a remote node's logs
+  // as the node-proxy admin. Cover the collection, the SSE sub-path, and the
+  // stream-metrics endpoint, with and without a trailing slash.
+  it('rejects /api/logs/global with 403 when nodeId targets a remote node', async () => {
+    const res = await request(app)
+      .get('/api/logs/global')
+      .set('Authorization', authHeader)
+      .set('x-node-id', String(remoteNodeId));
+
+    expect(res.status).toBe(403);
+    expect(res.body?.code).toBe('HUB_ONLY_ENDPOINT');
+  });
+
+  it('rejects /api/logs/global/stream with 403 when nodeId targets a remote node', async () => {
+    const res = await request(app)
+      .get('/api/logs/global/stream')
+      .set('Authorization', authHeader)
+      .set('x-node-id', String(remoteNodeId));
+
+    expect(res.status).toBe(403);
+    expect(res.body?.code).toBe('HUB_ONLY_ENDPOINT');
+  });
+
+  it('rejects /api/logs/global/stream via the ?nodeId= query param (SSE transport)', async () => {
+    const res = await request(app)
+      .get(`/api/logs/global/stream?nodeId=${remoteNodeId}`)
+      .set('Authorization', authHeader);
+
+    expect(res.status).toBe(403);
+    expect(res.body?.code).toBe('HUB_ONLY_ENDPOINT');
+  });
+
+  it('rejects /api/system/log-stream-metrics with 403 when nodeId targets a remote node', async () => {
+    const res = await request(app)
+      .get('/api/system/log-stream-metrics')
+      .set('Authorization', authHeader)
+      .set('x-node-id', String(remoteNodeId));
+
+    expect(res.status).toBe(403);
+    expect(res.body?.code).toBe('HUB_ONLY_ENDPOINT');
+  });
+
   it('does not interfere with non-hub paths even when nodeId targets a remote node', async () => {
     // /api/stacks is not hub-only and should be forwarded by the proxy.
     // The exact upstream-error status is not the contract here; what

backend/src/__tests__/log-parsing.test.ts

@@ -5,8 +5,18 @@ import {
   detectLogLevel,
   stripControlChars,
   demuxDockerLog,
+  createFrameDemuxer,
 } from '../utils/log-parsing';
 
+/** Build one Docker multiplexed frame: [streamType, 0,0,0, len(BE)] + payload. */
+function frame(streamType: 1 | 2, payload: string): Buffer {
+  const body = Buffer.from(payload);
+  const header = Buffer.alloc(8);
+  header[0] = streamType;
+  header.writeUInt32BE(body.length, 4);
+  return Buffer.concat([header, body]);
+}
+
 // ── normalizeContainerName ──────────────────────────────────────────────────
 
 describe('normalizeContainerName', () => {
@@ -192,27 +202,18 @@ describe('demuxDockerLog', () => {
     const buf = Buffer.from('line1\nline2\n');
     const lines: Array<{ line: string; source: string }> = [];
     demuxDockerLog(buf, true, (line, source) => lines.push({ line, source }));
+    // A trailing newline does not produce a spurious empty final line.
     expect(lines).toEqual([
       { line: 'line1', source: 'STDOUT' },
       { line: 'line2', source: 'STDOUT' },
-      { line: '', source: 'STDOUT' },
     ]);
   });
 
   it('parses multiplexed STDOUT frame', () => {
-    // Build a frame: [1, 0, 0, 0, <length BE>, ...payload]
-    const payload = Buffer.from('hello stdout\n');
-    const header = Buffer.alloc(8);
-    header[0] = 1; // STDOUT
-    header.writeUInt32BE(payload.length, 4);
-    const buf = Buffer.concat([header, payload]);
-
+    const buf = frame(1, 'hello stdout\n');
     const lines: Array<{ line: string; source: string }> = [];
     demuxDockerLog(buf, false, (line, source) => lines.push({ line, source }));
-    expect(lines).toEqual([
-      { line: 'hello stdout', source: 'STDOUT' },
-      { line: '', source: 'STDOUT' },
-    ]);
+    expect(lines).toEqual([{ line: 'hello stdout', source: 'STDOUT' }]);
   });
 
   it('parses multiplexed STDERR frame', () => {
@@ -254,3 +255,70 @@ describe('demuxDockerLog', () => {
     expect(lines[0]).toEqual({ line: 'cleantext', source: 'STDOUT' });
   });
 });
+
+// ── createFrameDemuxer (stateful, survives chunk boundaries) ─────────────────
+
+describe('createFrameDemuxer', () => {
+  function collect() {
+    const lines: Array<{ line: string; source: string }> = [];
+    let errors = 0;
+    const d = createFrameDemuxer(false, (line, source) => lines.push({ line, source }), () => { errors += 1; });
+    return { lines, push: d.push, flush: d.flush, errors: () => errors };
+  }
+
+  it('reassembles a frame whose 8-byte header is split across two chunks', () => {
+    const full = frame(1, 'split header\n');
+    const c = collect();
+    c.push(full.subarray(0, 3)); // first 3 bytes of the header
+    c.push(full.subarray(3));    // remainder
+    c.flush();
+    expect(c.lines).toEqual([{ line: 'split header', source: 'STDOUT' }]);
+  });
+
+  it('reassembles a frame whose payload is split across two chunks', () => {
+    const full = frame(2, 'partial payload\n');
+    const c = collect();
+    c.push(full.subarray(0, 12)); // header + start of payload
+    c.push(full.subarray(12));
+    c.flush();
+    expect(c.lines).toEqual([{ line: 'partial payload', source: 'STDERR' }]);
+  });
+
+  it('joins a single log line split across two frames', () => {
+    const c = collect();
+    c.push(frame(1, 'hello ')); // no newline yet
+    c.push(frame(1, 'world\n'));
+    c.flush();
+    expect(c.lines).toEqual([{ line: 'hello world', source: 'STDOUT' }]);
+  });
+
+  it('keeps interleaved STDOUT/STDERR partial lines separate', () => {
+    const c = collect();
+    c.push(frame(1, 'out-part '));
+    c.push(frame(2, 'err-part '));
+    c.push(frame(1, 'out-end\n'));
+    c.push(frame(2, 'err-end\n'));
+    c.flush();
+    expect(c.lines).toEqual([
+      { line: 'out-part out-end', source: 'STDOUT' },
+      { line: 'err-part err-end', source: 'STDERR' },
+    ]);
+  });
+
+  it('flushes a buffered partial line with no trailing newline on stream end', () => {
+    const c = collect();
+    c.push(frame(1, 'no newline'));
+    expect(c.lines).toEqual([]); // held until flush
+    c.flush();
+    expect(c.lines).toEqual([{ line: 'no newline', source: 'STDOUT' }]);
+  });
+
+  it('counts a malformed frame header and resyncs instead of stalling', () => {
+    const c = collect();
+    // A stray byte > 2 where a stream type is expected, then a valid frame.
+    c.push(Buffer.concat([Buffer.from([0x07]), frame(1, 'recovered\n')]));
+    c.flush();
+    expect(c.errors()).toBeGreaterThanOrEqual(1);
+    expect(c.lines).toContainEqual({ line: 'recovered', source: 'STDOUT' });
+  });
+});

backend/src/__tests__/managed-containers.test.ts

@@ -0,0 +1,35 @@
+import { describe, it, expect } from 'vitest';
+import path from 'path';
+import { isManagedByComposeDir } from '../utils/managed-containers';
+
+const COMPOSE_DIR = path.resolve('/srv/compose');
+
+function withWorkingDir(workingDir?: string): { Labels?: Record<string, string> } {
+  return workingDir === undefined
+    ? {}
+    : { Labels: { 'com.docker.compose.project.working_dir': workingDir } };
+}
+
+describe('isManagedByComposeDir', () => {
+  it('treats a container in a subdirectory of COMPOSE_DIR as managed', () => {
+    expect(isManagedByComposeDir(withWorkingDir(path.join(COMPOSE_DIR, 'web')), COMPOSE_DIR)).toBe(true);
+  });
+
+  it('treats a container launched from the COMPOSE_DIR root as managed', () => {
+    expect(isManagedByComposeDir(withWorkingDir(COMPOSE_DIR), COMPOSE_DIR)).toBe(true);
+  });
+
+  it('treats a container outside COMPOSE_DIR as unmanaged', () => {
+    expect(isManagedByComposeDir(withWorkingDir('/opt/other/stack'), COMPOSE_DIR)).toBe(false);
+  });
+
+  it('does not match a sibling directory that shares the COMPOSE_DIR prefix', () => {
+    // /srv/compose-extra must not be considered inside /srv/compose.
+    expect(isManagedByComposeDir(withWorkingDir(`${COMPOSE_DIR}-extra/web`), COMPOSE_DIR)).toBe(false);
+  });
+
+  it('treats a container with no compose working-dir label as unmanaged', () => {
+    expect(isManagedByComposeDir(withWorkingDir(undefined), COMPOSE_DIR)).toBe(false);
+    expect(isManagedByComposeDir({ Labels: {} }, COMPOSE_DIR)).toBe(false);
+  });
+});

backend/src/__tests__/metrics-routes.test.ts

@@ -109,12 +109,59 @@ describe('GET /api/system/cache-stats', () => {
   });
 });
 
+describe('GET /api/logs/global (poll snapshot)', () => {
+  it('rejects unauthenticated requests with 401', async () => {
+    const res = await request(app).get('/api/logs/global');
+    expect(res.status).toBe(401);
+  });
+
+  it('rejects non-admin users with 403', async () => {
+    const res = await request(app).get('/api/logs/global').set('Cookie', viewerCookie);
+    expect(res.status).toBe(403);
+  });
+
+  it('admin gets a non-4xx response', async () => {
+    // Reaches the Docker daemon; 500 is acceptable in CI without Docker. We
+    // only prove the admin gate + routing, not the daemon read.
+    const res = await request(app).get('/api/logs/global').set('Cookie', adminCookie);
+    expect([200, 500]).toContain(res.status);
+    if (res.status === 200) expect(Array.isArray(res.body)).toBe(true);
+  });
+});
+
+describe('GET /api/system/log-stream-metrics', () => {
+  it('rejects unauthenticated requests with 401', async () => {
+    const res = await request(app).get('/api/system/log-stream-metrics');
+    expect(res.status).toBe(401);
+  });
+
+  it('rejects non-admin users with 403', async () => {
+    const res = await request(app).get('/api/system/log-stream-metrics').set('Cookie', viewerCookie);
+    expect(res.status).toBe(403);
+  });
+
+  it('returns the counter snapshot for admin', async () => {
+    const res = await request(app).get('/api/system/log-stream-metrics').set('Cookie', adminCookie);
+    expect(res.status).toBe(200);
+    expect(res.body).toHaveProperty('active_sse_connections');
+    expect(res.body).toHaveProperty('lines_streamed_total');
+    expect(res.body).toHaveProperty('stream_attach_errors_total');
+  });
+});
+
 describe('GET /api/logs/global/stream', () => {
   it('rejects unauthenticated requests with 401', async () => {
     const res = await request(app).get('/api/logs/global/stream');
     expect(res.status).toBe(401);
   });
 
+  it('rejects non-admin users with 403 before opening the stream', async () => {
+    // requireAdmin runs before flushHeaders, so a viewer gets a clean JSON 403
+    // rather than a half-open event-stream.
+    const res = await request(app).get('/api/logs/global/stream').set('Cookie', viewerCookie);
+    expect(res.status).toBe(403);
+  });
+
   it('sets SSE response headers for authenticated users', async () => {
     // The SSE handler writes headers immediately then keeps the connection
     // open. supertest .end() after we see the headers lets Express flush

backend/src/helpers/proxyExemptPaths.ts

@@ -23,11 +23,14 @@ export function isProxyExemptPath(path: string): boolean {
   return false;
 }
 
-// Path prefixes that are hub-only: they manage state owned by the local hub
-// (centralized audit, fleet schedules, notification routing rules). Routed
-// to the local hub when nodeId resolves to local, but rejected with 409 when
-// nodeId resolves to a remote node so a script/curl call cannot trick the
-// proxy into forwarding hub-only authority across a node boundary.
+// Path prefixes that are hub-only: they manage or expose state owned by the
+// local hub (centralized audit, fleet schedules, notification routing rules,
+// the admin-only aggregated logs feed and its stream counters). Routed to the
+// local hub when nodeId resolves to local, but rejected when nodeId resolves
+// to a remote node so a script/curl call cannot trick the proxy into
+// forwarding hub-only authority across a node boundary. This matters for the
+// logs feed in particular: its admin gate lives in the local route handler,
+// which the proxy would skip entirely when forwarding a remote nodeId.
 //
 // Entries are stored with a trailing slash; the matcher accepts the exact
 // collection path (without the trailing slash) AND any sub-path under it,
@@ -44,6 +47,8 @@ export const HUB_ONLY_PREFIXES: readonly string[] = [
   '/api/scheduled-tasks/',
   '/api/audit-log/',
   '/api/notification-routes/',
+  '/api/logs/global/',
+  '/api/system/log-stream-metrics/',
 ];
 
 /** Returns true when the path is hub-only and must not be proxied to a remote node. */

backend/src/routes/metrics.ts

@@ -11,16 +11,71 @@ import { requireAdmin } from '../middleware/tierGates';
 import { STATS_CACHE_TTL_MS, SYSTEM_STATS_CACHE_TTL_MS } from '../helpers/constants';
 import { isDebugEnabled } from '../utils/debug';
 import { getErrorMessage } from '../utils/errors';
+import { isManagedByComposeDir } from '../utils/managed-containers';
+import { GlobalLogsMetrics } from '../services/GlobalLogsMetrics';
 import {
   type GlobalLogEntry,
   normalizeContainerName,
   parseLogTimestamp,
   detectLogLevel,
   demuxDockerLog,
+  createFrameDemuxer,
 } from '../utils/log-parsing';
 
 export const metricsRouter = Router();
 
+// Lines of history each container replays when a feed opens. Bounds the
+// open-time burst (was 500 per container, multiplied across every container).
+const STREAM_INITIAL_TAIL = 200;
+const POLL_TAIL = 100;
+// Hard cap on simultaneous `docker logs --follow` streams behind one SSE
+// connection. Beyond this the feed is truncated and the operator is told.
+const MAX_FOLLOW_STREAMS = 60;
+// Concurrency limit for the polling snapshot's per-container Docker calls so a
+// large managed set does not fan out N simultaneous requests at the daemon.
+const POLL_CONCURRENCY = 8;
+
+interface ContainerSummary {
+  Id: string;
+  Names?: string[];
+  Labels?: Record<string, string>;
+}
+
+/** Read a managed-only set of running containers for the node. */
+async function getManagedRunningContainers(
+  nodeId: number,
+): Promise<{ containers: ContainerSummary[]; total: number }> {
+  const composeDir = path.resolve(NodeRegistry.getInstance().getComposeDir(nodeId));
+  const all = (await DockerController.getInstance(nodeId).getRunningContainers()) as ContainerSummary[];
+  const containers = all.filter(c => isManagedByComposeDir(c, composeDir));
+  return { containers, total: all.length };
+}
+
+/** Map a Docker container summary to its display stack + container name. */
+function describeContainer(c: ContainerSummary): { stackName: string; containerName: string } {
+  const stackName = c.Labels?.['com.docker.compose.project'] || 'system';
+  const rawName = c.Names?.[0]?.replace(/^\//, '') || c.Id.substring(0, 12);
+  return { stackName, containerName: normalizeContainerName(rawName, stackName) };
+}
+
+/**
+ * Run an async mapper over items with a bounded number in flight. The caller's
+ * `fn` must handle its own errors: a rejection propagates and abandons the
+ * remaining work (the log endpoints wrap their body in try/catch for this).
+ * Exported for unit testing.
+ */
+export async function mapWithConcurrency<T>(items: T[], limit: number, fn: (item: T) => Promise<void>): Promise<void> {
+  let cursor = 0;
+  const worker = async (): Promise<void> => {
+    while (cursor < items.length) {
+      const item = items[cursor++];
+      await fn(item);
+    }
+  };
+  const workerCount = Math.min(limit, items.length);
+  await Promise.all(Array.from({ length: workerCount }, () => worker()));
+}
+
 /**
  * Container stats aggregated for the dashboard. Cached per-node for 2s to
  * collapse multi-tab polling pressure. Write-path endpoints (deploy, down,
@@ -35,24 +90,13 @@ metricsRouter.get('/stats', authMiddleware, async (req: Request, res: Response):
       async () => {
         const allContainers = await DockerController.getInstance(req.nodeId).getAllContainers();
 
-        // "Managed" means Docker started the container from within COMPOSE_DIR.
-        // We key on `com.docker.compose.project.working_dir` rather than the
-        // project name so stacks launched from the COMPOSE_DIR root (not a
-        // subdirectory) aren't all mis-classified as external.
-        const isManagedByComposeDir = (c: { Labels?: Record<string, string> }): boolean => {
-          const workingDir: string | undefined = c.Labels?.['com.docker.compose.project.working_dir'];
-          if (!workingDir) return false;
-          const resolved = path.resolve(workingDir);
-          return resolved === composeDir || resolved.startsWith(composeDir + path.sep);
-        };
-
         type ContainerInfo = { State?: string; Labels?: Record<string, string> };
         const cs = allContainers as ContainerInfo[];
         const active = cs.filter(c => c.State === 'running').length;
         const exited = cs.filter(c => c.State === 'exited').length;
         const total = cs.length;
-        const managed = cs.filter(c => c.State === 'running' && isManagedByComposeDir(c)).length;
-        const unmanaged = cs.filter(c => c.State === 'running' && !isManagedByComposeDir(c)).length;
+        const managed = cs.filter(c => c.State === 'running' && isManagedByComposeDir(c, composeDir)).length;
+        const unmanaged = cs.filter(c => c.State === 'running' && !isManagedByComposeDir(c, composeDir)).length;
 
         return { active, managed, unmanaged, exited, total };
       },
@@ -73,23 +117,22 @@ metricsRouter.get('/metrics/historical', authMiddleware, async (_req: Request, r
 });
 
 metricsRouter.get('/logs/global', authMiddleware, async (req: Request, res: Response): Promise<void> => {
+  if (!requireAdmin(req, res)) return;
   try {
+    GlobalLogsMetrics.increment('poll_requests_total');
     const debug = isDebugEnabled();
     const dockerController = DockerController.getInstance(req.nodeId);
-    const containers = await dockerController.getRunningContainers();
+    const { containers, total } = await getManagedRunningContainers(req.nodeId);
     const allLogs: GlobalLogEntry[] = [];
-    if (debug) console.debug('[GlobalLogs:debug] Polling snapshot starting', { containerCount: containers.length, nodeId: req.nodeId });
-
-    await Promise.all(containers.map(async (c) => {
-      const stackName = c.Labels?.['com.docker.compose.project'] || 'system';
-      const rawName = c.Names?.[0]?.replace(/^\//, '') || c.Id.substring(0, 12);
-      const containerName = normalizeContainerName(rawName, stackName);
+    if (debug) console.debug('[GlobalLogs:debug] Polling snapshot starting', { managed: containers.length, total, nodeId: req.nodeId });
 
+    await mapWithConcurrency(containers, POLL_CONCURRENCY, async (c) => {
+      const { stackName, containerName } = describeContainer(c);
       try {
         const container = dockerController.getDocker().getContainer(c.Id);
         const inspect = await container.inspect();
         const isTty = inspect.Config.Tty;
-        const logsBuffer = await container.logs({ stdout: true, stderr: true, tail: 100, timestamps: true }) as Buffer;
+        const logsBuffer = await container.logs({ stdout: true, stderr: true, tail: POLL_TAIL, timestamps: true }) as Buffer;
 
         demuxDockerLog(logsBuffer, isTty, (line, source) => {
           if (!line.trim()) return;
@@ -98,15 +141,20 @@ metricsRouter.get('/logs/global', authMiddleware, async (req: Request, res: Resp
           allLogs.push({ stackName, containerName, source, level, message: cleanMessage, timestampMs });
         });
       } catch (err) {
+        // Mirror the SSE path so per-container read failures are visible on the
+        // same counter rather than only in the log.
+        GlobalLogsMetrics.increment('stream_attach_errors_total');
         console.warn(`[GlobalLogs] Failed to fetch/parse logs for container ${containerName} (${c.Id.substring(0, 12)}):`, getErrorMessage(err, 'unknown'));
       }
-    }));
+    });
 
     // Sort ascending by timestamp (newest bottom). Limit to 500 lines; the
     // client only renders ~300 at a time.
     allLogs.sort((a, b) => a.timestampMs - b.timestampMs);
-    if (debug) console.debug('[GlobalLogs:debug] Polling snapshot complete', { totalLines: allLogs.length });
-    res.json(allLogs.slice(-500));
+    const snapshot = allLogs.slice(-500);
+    GlobalLogsMetrics.increment('lines_streamed_total', snapshot.length);
+    if (debug) console.debug('[GlobalLogs:debug] Polling snapshot complete', { totalLines: allLogs.length, returned: snapshot.length });
+    res.json(snapshot);
   } catch (error) {
     console.error('[GlobalLogs] Snapshot fetch failed:', getErrorMessage(error, 'unknown'));
     res.status(500).json({ error: 'Failed to fetch global logs' });
@@ -114,6 +162,8 @@ metricsRouter.get('/logs/global', authMiddleware, async (req: Request, res: Resp
 });
 
 metricsRouter.get('/logs/global/stream', authMiddleware, async (req: Request, res: Response): Promise<void> => {
+  if (!requireAdmin(req, res)) return;
+
   res.setHeader('Content-Type', 'text/event-stream');
   res.setHeader('Cache-Control', 'no-cache');
   res.setHeader('Connection', 'keep-alive');
@@ -125,57 +175,106 @@ metricsRouter.get('/logs/global/stream', authMiddleware, async (req: Request, re
   const dockerController = DockerController.getInstance(req.nodeId);
   const streams: NodeJS.ReadableStream[] = [];
 
+  GlobalLogsMetrics.openConnection();
+  let closed = false;
+  let paused = false;
+
+  const destroyStream = (s: NodeJS.ReadableStream): void => {
+    try { (s as NodeJS.ReadableStream & { destroy(): void }).destroy(); } catch { /* already ended */ }
+  };
+  // Back off every source follow-stream when the socket buffer fills, then
+  // resume on 'drain', so a slow client cannot drive unbounded Node-side
+  // buffering across N concurrent streams.
+  const pauseAll = (): void => { if (paused) return; paused = true; streams.forEach(s => { try { s.pause(); } catch { /* ended */ } }); };
+  const resumeAll = (): void => { if (!paused) return; paused = false; streams.forEach(s => { try { s.resume(); } catch { /* ended */ } }); };
+  res.on('drain', resumeAll);
+
+  const writeEvent = (entry: GlobalLogEntry): void => {
+    if (res.writableEnded) return;
+    const ok = res.write(`data: ${JSON.stringify(entry)}\n\n`);
+    GlobalLogsMetrics.increment('lines_streamed_total');
+    if (!ok) pauseAll();
+  };
+
   // SSE heartbeat (: prefix is a comment, silently dropped by EventSource)
-  // every 30s keeps reverse proxies from closing idle connections.
+  // every 30s keeps reverse proxies from closing idle connections. Honor
+  // backpressure here too so a heartbeat that fills the socket buffer still
+  // pauses the source streams.
   const heartbeat = setInterval(() => {
-    if (!res.writableEnded) res.write(':heartbeat\n\n');
+    if (!res.writableEnded && !res.write(':heartbeat\n\n')) pauseAll();
   }, 30_000);
 
+  const cleanup = (): void => {
+    if (closed) return;
+    closed = true;
+    clearInterval(heartbeat);
+    res.removeListener('drain', resumeAll);
+    if (debug) console.debug('[GlobalLogs:debug] SSE stream closed, cleaning up', { streamCount: streams.length });
+    streams.forEach(destroyStream);
+    GlobalLogsMetrics.closeConnection();
+  };
+  req.on('close', cleanup);
+
   try {
-    const containers = await dockerController.getRunningContainers();
-    if (debug) console.debug('[GlobalLogs:debug] SSE stream opened', { containerCount: containers.length, nodeId: req.nodeId });
-
-    await Promise.all(containers.map(async (c) => {
-      const stackName = c.Labels?.['com.docker.compose.project'] || 'system';
-      const rawName = c.Names?.[0]?.replace(/^\//, '') || c.Id.substring(0, 12);
-      const containerName = normalizeContainerName(rawName, stackName);
+    const { containers, total } = await getManagedRunningContainers(req.nodeId);
+    const followSet = containers.slice(0, MAX_FOLLOW_STREAMS);
+    const truncated = containers.length - followSet.length;
+    if (debug) console.debug('[GlobalLogs:debug] SSE stream opened', { managed: containers.length, total, following: followSet.length, nodeId: req.nodeId });
 
+    await Promise.all(followSet.map(async (c) => {
+      const { stackName, containerName } = describeContainer(c);
       try {
         const container = dockerController.getDocker().getContainer(c.Id);
         const inspect = await container.inspect();
         const isTty = inspect.Config.Tty;
 
-        const stream = await container.logs({ follow: true, stdout: true, stderr: true, tail: 500, timestamps: true });
+        const stream = await container.logs({ follow: true, stdout: true, stderr: true, tail: STREAM_INITIAL_TAIL, timestamps: true });
+        // The connection may have closed while we awaited inspect/logs.
+        if (closed) { destroyStream(stream); return; }
         streams.push(stream);
+        if (paused) { try { stream.pause(); } catch { /* ended */ } }
 
-        stream.on('data', (chunk: Buffer) => {
-          demuxDockerLog(chunk, isTty, (line, source) => {
+        const demuxer = createFrameDemuxer(
+          isTty,
+          (line, source) => {
             if (!line.trim()) return;
             const { timestampMs, cleanMessage } = parseLogTimestamp(line);
             const level = detectLogLevel(cleanMessage, source);
-            if (!res.writableEnded) {
-              res.write(`data: ${JSON.stringify({ stackName, containerName, source, level, message: cleanMessage, timestampMs })}\n\n`);
-            }
-          });
+            writeEvent({ stackName, containerName, source, level, message: cleanMessage, timestampMs });
+          },
+          () => GlobalLogsMetrics.increment('demux_frame_errors_total'),
+        );
+
+        stream.on('data', (chunk: Buffer) => demuxer.push(chunk));
+        // Drain the demuxer's buffered trailing line (one with no newline, the
+        // common shape of a crash/exit line) when the follow stream ends or
+        // breaks, so the last thing a container said is not silently lost.
+        stream.on('end', () => demuxer.flush());
+        stream.on('error', (err) => {
+          GlobalLogsMetrics.increment('stream_attach_errors_total');
+          console.warn(`[GlobalLogs] Follow stream error for ${containerName} (${c.Id.substring(0, 12)}):`, getErrorMessage(err, 'unknown'));
+          demuxer.flush();
+          // One degraded notice per drop (not per failed read) so the operator
+          // sees the gap in the feed and the WARNINGS tile without log spam.
+          writeEvent({ stackName, containerName, source: 'STDERR', level: 'WARN', message: `[Sencho] Log stream for ${containerName} ended unexpectedly; reopen the tab to resume.`, timestampMs: Date.now() });
+          destroyStream(stream);
         });
       } catch (err) {
+        GlobalLogsMetrics.increment('stream_attach_errors_total');
         console.warn(`[GlobalLogs] Failed to attach stream for container ${containerName} (${c.Id.substring(0, 12)}):`, getErrorMessage(err, 'unknown'));
       }
     }));
 
-    req.on('close', () => {
-      clearInterval(heartbeat);
-      if (debug) console.debug('[GlobalLogs:debug] SSE stream closed, cleaning up', { streamCount: streams.length });
-      streams.forEach(s => {
-        try { (s as NodeJS.ReadableStream & { destroy(): void }).destroy(); } catch { /* stream already ended */ }
-      });
-    });
-
+    if (truncated > 0) {
+      writeEvent({ stackName: 'system', containerName: 'sencho', source: 'STDOUT', level: 'WARN', message: `[Sencho] Following ${followSet.length} of ${containers.length} managed containers; ${truncated} not shown. Use the per-container log viewer for the rest.`, timestampMs: Date.now() });
+    }
   } catch (error) {
-    clearInterval(heartbeat);
     console.error('[GlobalLogs] SSE stream attachment failed:', getErrorMessage(error, 'unknown'));
-    res.write(`data: ${JSON.stringify({ level: 'ERROR', message: '[Sencho] Failed to attach global log stream.', timestampMs: Date.now(), stackName: 'system', containerName: 'backend', source: 'STDERR' })}\n\n`);
-    res.end();
+    if (!res.writableEnded) {
+      res.write(`data: ${JSON.stringify({ level: 'ERROR', message: '[Sencho] Failed to attach global log stream.', timestampMs: Date.now(), stackName: 'system', containerName: 'backend', source: 'STDERR' })}\n\n`);
+    }
+    cleanup();
+    if (!res.writableEnded) res.end();
   }
 });
 
@@ -267,3 +366,20 @@ metricsRouter.get('/system/pilot-tunnels', authMiddleware, async (req: Request,
     res.status(500).json({ error: 'Failed to fetch pilot tunnel metrics' });
   }
 });
+
+/**
+ * Admin-only Global Observability log-stream observability. Process-local,
+ * in-memory counters (reset on restart by design; see GlobalLogsMetrics). The
+ * `active_sse_connections` gauge is the load-bearing field: it should drain to
+ * zero when no Logs tab is open, and a rising `stream_attach_errors_total` or
+ * `demux_frame_errors_total` points at a daemon or stream-corruption problem.
+ */
+metricsRouter.get('/system/log-stream-metrics', authMiddleware, async (req: Request, res: Response): Promise<void> => {
+  if (!requireAdmin(req, res)) return;
+  try {
+    res.json(GlobalLogsMetrics.snapshot());
+  } catch (error) {
+    console.error('Failed to fetch log-stream metrics:', error);
+    res.status(500).json({ error: 'Failed to fetch log-stream metrics' });
+  }
+});

backend/src/services/GlobalLogsMetrics.ts

@@ -0,0 +1,73 @@
+/**
+ * GlobalLogsMetrics: process-local counters for the Global Observability log
+ * path (the SSE stream and the polling snapshot). Strictly process-local and
+ * in-memory: Sencho does not export metrics to any external sink (privacy
+ * posture, process-local by design), and unlike PilotMetrics these counters
+ * are NOT persisted to SQLite because they describe ephemeral live-stream
+ * activity that is meaningless across a restart. They reset on process start.
+ *
+ * Surfaced via GET /api/system/log-stream-metrics (admin only). Purpose is
+ * operator support: spot a connection gauge that never drains (a leak) or a
+ * rising attach/frame-error count (a daemon or demux problem).
+ *
+ * No general in-process metrics facility exists in the backend today; this is
+ * a per-feature pattern mirroring PilotMetrics. When a shared facility lands,
+ * this module should be replaced by an instance of it rather than grown.
+ */
+
+export interface LogStreamCounters {
+  /** Live gauge: SSE connections currently open. Should drain to 0 when no tab watches. */
+  active_sse_connections: number;
+  /** Monotonic: SSE connections opened since process start. */
+  sse_connections_total: number;
+  /** Monotonic: polling-fallback snapshot requests served. */
+  poll_requests_total: number;
+  /** Monotonic: log lines pushed to clients (SSE + polling). */
+  lines_streamed_total: number;
+  /** Monotonic: follow-stream attach failures and mid-stream stream errors. */
+  stream_attach_errors_total: number;
+  /** Monotonic: malformed demux frame headers (corrupt stream, resynced). */
+  demux_frame_errors_total: number;
+}
+
+type MonotonicCounter = Exclude<keyof LogStreamCounters, 'active_sse_connections'>;
+
+const ZERO: LogStreamCounters = {
+  active_sse_connections: 0,
+  sse_connections_total: 0,
+  poll_requests_total: 0,
+  lines_streamed_total: 0,
+  stream_attach_errors_total: 0,
+  demux_frame_errors_total: 0,
+};
+
+class GlobalLogsMetricsImpl {
+  private counters: LogStreamCounters = { ...ZERO };
+
+  /** Increment a monotonic counter by `by` (default 1). */
+  public increment(name: MonotonicCounter, by = 1): void {
+    this.counters[name] += by;
+  }
+
+  /** A new SSE connection opened: bump the gauge and the total. */
+  public openConnection(): void {
+    this.counters.active_sse_connections += 1;
+    this.counters.sse_connections_total += 1;
+  }
+
+  /** An SSE connection closed: drop the gauge, clamped at zero. */
+  public closeConnection(): void {
+    this.counters.active_sse_connections = Math.max(0, this.counters.active_sse_connections - 1);
+  }
+
+  public snapshot(): LogStreamCounters {
+    return { ...this.counters };
+  }
+
+  /** Reset all in-memory state. Tests only. */
+  public resetForTests(): void {
+    this.counters = { ...ZERO };
+  }
+}
+
+export const GlobalLogsMetrics = new GlobalLogsMetricsImpl();

backend/src/utils/log-parsing.ts

@@ -7,10 +7,13 @@
  * them here eliminates duplication and provides a single place to test.
  */
 
+/** Which Docker stream a log line came from. */
+export type LogStreamSource = 'STDOUT' | 'STDERR';
+
 export interface GlobalLogEntry {
   stackName: string;
   containerName: string;
-  source: 'STDOUT' | 'STDERR';
+  source: LogStreamSource;
   level: 'INFO' | 'WARN' | 'ERROR';
   message: string;
   timestampMs: number;
@@ -87,7 +90,7 @@ export function parseLogTimestamp(line: string): { timestampMs: number; cleanMes
  *   3. ERROR/FATAL/CRIT/PANIC keywords or `Exception:` pattern
  *   4. Fallback: STDERR -> ERROR, STDOUT -> INFO
  */
-export function detectLogLevel(message: string, source: 'STDOUT' | 'STDERR'): 'INFO' | 'WARN' | 'ERROR' {
+export function detectLogLevel(message: string, source: LogStreamSource): 'INFO' | 'WARN' | 'ERROR' {
   // Tier 1: INFO/DEBUG/TRACE (overrides STDERR default)
   if (INFO_STRUCTURED_RE.test(message) || INFO_BRACKET_RE.test(message) || INFO_KEYWORD_RE.test(message)) {
     return 'INFO';
@@ -109,32 +112,106 @@ export function stripControlChars(text: string): string {
   return text.replace(CONTROL_CHARS_RE, '');
 }
 
+/** A stateful demuxer that survives chunk boundaries. See `createFrameDemuxer`. */
+export interface FrameDemuxer {
+  /** Feed the next chunk of stream bytes; emits every complete line found. */
+  push(chunk: Buffer): void;
+  /** Emit any buffered partial line that has no trailing newline. Call once on stream end. */
+  flush(): void;
+}
+
 /**
- * Parse Docker's multiplexed log stream format and call `onLine` for each
- * line. TTY containers produce raw text (no headers); non-TTY containers
- * prepend an 8-byte header per frame:
+ * Create a stateful demuxer for Docker's multiplexed log stream.
+ *
+ * Unlike a one-shot parse, this retains state across `push` calls so a frame
+ * header or payload split across two chunk boundaries is reassembled instead
+ * of dropped, and a log line split across two frames is joined instead of
+ * emitted as two partials. Line buffers are kept per source so interleaved
+ * STDOUT/STDERR frames don't bleed into one another.
+ *
+ * TTY containers produce raw text (no headers, single STDOUT stream); non-TTY
+ * containers prepend an 8-byte header per frame:
  *   [streamType(1), reserved(3), payloadLength(4 BE)]
+ *
+ * @param onFrameError called once per malformed frame header (invalid stream
+ *   type). The demuxer advances one byte to attempt resync rather than
+ *   stalling; the callback lets the caller count corruption events.
+ */
+export function createFrameDemuxer(
+  isTty: boolean,
+  onLine: (line: string, source: LogStreamSource) => void,
+  onFrameError?: () => void,
+): FrameDemuxer {
+  if (isTty) {
+    let lineBuf = '';
+    return {
+      push(chunk: Buffer): void {
+        lineBuf += stripControlChars(chunk.toString('utf-8'));
+        const parts = lineBuf.split('\n');
+        lineBuf = parts.pop() ?? '';
+        for (const line of parts) onLine(line, 'STDOUT');
+      },
+      flush(): void {
+        if (lineBuf) {
+          onLine(lineBuf, 'STDOUT');
+          lineBuf = '';
+        }
+      },
+    };
+  }
+
+  let leftover: Buffer = Buffer.alloc(0);
+  const lineBufs: Record<LogStreamSource, string> = { STDOUT: '', STDERR: '' };
+
+  const emitPayload = (payload: string, source: LogStreamSource): void => {
+    const parts = (lineBufs[source] + payload).split('\n');
+    lineBufs[source] = parts.pop() ?? '';
+    for (const line of parts) onLine(line, source);
+  };
+
+  return {
+    push(chunk: Buffer): void {
+      const buf = leftover.length ? Buffer.concat([leftover, chunk]) : chunk;
+      let offset = 0;
+      while (offset + 8 <= buf.length) {
+        const streamType = buf[offset];
+        // Valid Docker stream types: 0 (stdin, unused here), 1 (stdout), 2 (stderr).
+        if (streamType > 2) {
+          onFrameError?.();
+          offset += 1; // attempt resync rather than stalling on a corrupt header
+          continue;
+        }
+        const length = buf.readUInt32BE(offset + 4);
+        if (offset + 8 + length > buf.length) break; // payload not fully arrived yet
+        const payload = buf.slice(offset + 8, offset + 8 + length).toString('utf-8');
+        offset += 8 + length;
+        emitPayload(payload, streamType === 2 ? 'STDERR' : 'STDOUT');
+      }
+      leftover = offset < buf.length ? buf.subarray(offset) : Buffer.alloc(0);
+    },
+    flush(): void {
+      for (const source of ['STDOUT', 'STDERR'] as const) {
+        if (lineBufs[source]) {
+          onLine(lineBufs[source], source);
+          lineBufs[source] = '';
+        }
+      }
+    },
+  };
+}
+
+/**
+ * One-shot demux of a complete Docker log buffer (the polling snapshot path,
+ * where the whole response is in hand). Implemented on top of
+ * `createFrameDemuxer` so frame- and line-splitting are handled identically to
+ * the streaming path.
  */
 export function demuxDockerLog(
   buf: Buffer,
   isTty: boolean,
-  onLine: (line: string, source: 'STDOUT' | 'STDERR') => void,
+  onLine: (line: string, source: LogStreamSource) => void,
 ): void {
-  if (isTty) {
-    stripControlChars(buf.toString('utf-8'))
-      .split('\n')
-      .forEach(line => onLine(line, 'STDOUT'));
-    return;
-  }
-  let offset = 0;
-  while (offset < buf.length) {
-    if (offset + 8 > buf.length) break;
-    const streamType = buf[offset];
-    const length = buf.readUInt32BE(offset + 4);
-    offset += 8;
-    if (offset + length > buf.length) break;
-    const payload = buf.slice(offset, offset + length).toString('utf-8');
-    offset += length;
-    payload.split('\n').forEach(line => onLine(line, streamType === 2 ? 'STDERR' : 'STDOUT'));
-  }
+  const demuxer = createFrameDemuxer(isTty, onLine);
+  demuxer.push(buf);
+  demuxer.flush();
 }

backend/src/utils/managed-containers.ts

@@ -0,0 +1,25 @@
+import path from 'path';
+
+/** Minimal container shape needed to decide managed-vs-external. */
+export interface LabeledContainer {
+  Labels?: Record<string, string>;
+}
+
+/**
+ * "Managed" means Docker started the container from within COMPOSE_DIR.
+ *
+ * We key on `com.docker.compose.project.working_dir` rather than the project
+ * name so stacks launched from the COMPOSE_DIR root (not a subdirectory)
+ * aren't mis-classified as external. Containers without the label (plain
+ * `docker run`, or another tool's compose project outside COMPOSE_DIR) are
+ * treated as unmanaged.
+ *
+ * @param container any object carrying Docker labels
+ * @param composeDir an already `path.resolve`d COMPOSE_DIR for the node
+ */
+export function isManagedByComposeDir(container: LabeledContainer, composeDir: string): boolean {
+  const workingDir = container.Labels?.['com.docker.compose.project.working_dir'];
+  if (!workingDir) return false;
+  const resolved = path.resolve(workingDir);
+  return resolved === composeDir || resolved.startsWith(composeDir + path.sep);
+}