fix(logging): ALS-safe mixin, quieter errors, matrix login creds (#16)

* fix(logging): ALS-safe mixin, quieter errors, matrix login creds

## Fixed
- Clone AsyncLocalStorage context in pino mixin so merge strategy does not mutate stored poll/chat context across log lines.
- Drop duplicate failure logs from withDurationMs and multi-insight sink; engine logs delivery outcome once and skips dedupe on post failure.
- Install Matrix access token and user id after password login so requests are not sent as M_MISSING_TOKEN.

## Changed
- Emit service/env/version only on the startup log line; README notes Loki query fields accordingly.

* refactor(logging): tighten comments and multi-sink fan-out

Author: kgrubb

README.md

@@ -154,7 +154,7 @@ Agent tools include flow search, rankings, fan-in, rare-destination / port-scan
 
 Logging verbosity, redaction, and the `env` field use fixed defaults and are not configurable via environment variables.
 
-In Kubernetes with Loki or VictoriaLogs, query on stable fields such as `service="kaytoo"`, `component`, `level`, `pollId`, `eventId`, and `msg`.
+In Kubernetes with Loki or VictoriaLogs, query on stable fields such as `component`, `level`, `pollId`, `eventId`, and `msg`. Service/env/version are emitted once on the `kaytoo starting` line; rely on pod labels elsewhere.
 
 ### Chat adapters
 

src/chat/adapters/matrix.ts

@@ -50,12 +50,14 @@ export async function startMatrixAdapter(opts: {
   });
 
   if ('user' in opts.auth) {
-    await client.loginRequest({
+    const resp = await client.loginRequest({
       type: 'm.login.password',
       identifier: loginIdentifier(opts.auth.user),
       password: opts.auth.password,
       initial_device_display_name: 'kaytoo',
     });
+    client.setAccessToken(resp.access_token);
+    client.credentials = { userId: resp.user_id };
   }
 
   const onTimeline = async (

src/insights/engine.ts

@@ -200,7 +200,13 @@ export async function startInsightEngine(opts: { config: KaytooConfig; insightSi
         return formatFindingsFallback(novel);
       });
 
-    await opts.insightSink.postInsight(text);
+    try {
+      await opts.insightSink.postInsight(text);
+    } catch (e) {
+      // Notifier already logged the cause; record outcome and skip dedupe so the next poll retries.
+      log.warn({ findingCount: novel.length, output: config.output, ...logErr(e) }, 'post findings failed');
+      return;
+    }
     for (const f of novel) dedupe.mark(f.id);
     log.info({ findingCount: novel.length, output: config.output }, 'posted findings');
   }

src/logging/logger.ts

@@ -6,6 +6,14 @@ import { getLogContext } from './context.js';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
 
+export function getServiceMetadata(): { service: 'kaytoo'; env: string; version: string } {
+  return {
+    service: 'kaytoo',
+    env: process.env.NODE_ENV ?? 'development',
+    version: readPackageVersion(),
+  };
+}
+
 const defaultRedactPaths = [
   'password',
   '*.token',
@@ -40,14 +48,13 @@ export function initLogging(opts: LoggingInit): PinoLogger {
   const redact = [...defaultRedactPaths, ...opts.redactPaths];
   const baseOpts = {
     level: opts.level,
-    base: {
-      service: 'kaytoo',
-      env: process.env.NODE_ENV ?? 'development',
-      version: readPackageVersion(),
-    },
+    base: null,
     redact: { paths: redact, censor: '[Redacted]' },
     mixin() {
-      return getLogContext() ?? {};
+      // Pino merges per-call fields into the returned object in place; copy so
+      // the AsyncLocalStorage store isn't mutated across log calls.
+      const ctx = getLogContext();
+      return ctx ? { ...ctx } : {};
     },
   };
   root.current = opts.destination ? pino(baseOpts, opts.destination) : pino(baseOpts);
@@ -102,18 +109,7 @@ export function logErr(e: unknown): { err: { name: string; message: string; stac
 
 export async function withDurationMs<T>(log: PinoLogger, msg: string, fn: () => Promise<T>): Promise<T> {
   const t0 = Date.now();
-  try {
-    const out = await fn();
-    log.debug({ durationMs: Date.now() - t0 }, msg);
-    return out;
-  } catch (e) {
-    log.warn(
-      {
-        durationMs: Date.now() - t0,
-        ...(e instanceof Error ? { err: e } : logErr(e)),
-      },
-      `${msg} failed`,
-    );
-    throw e;
-  }
+  const out = await fn();
+  log.debug({ durationMs: Date.now() - t0 }, msg);
+  return out;
 }

src/main.ts

@@ -3,7 +3,7 @@ import { getConfig } from './config.js';
 import { createConsoleInsightSink } from './notify/consoleInsightSink.js';
 import { createMultiInsightSink, createPlatformInsightSink, type InsightSink } from './notify/insightSink.js';
 import { startInsightEngine } from './insights/engine.js';
-import { getLogger, initLogging } from './logging/logger.js';
+import { getLogger, getServiceMetadata, initLogging } from './logging/logger.js';
 import { startE2eHttpChatServer } from './e2eHttpChatServer.js';
 import type { ChatRouter as ChatRouterT } from './chat/router.js';
 import type { Notifier } from './notify/notifier.js';
@@ -43,7 +43,12 @@ const { level, redactPaths } = config.logging;
 initLogging({ level, redactPaths });
 const log = getLogger({ component: 'main' });
 log.info(
-  { indexPattern: config.search.indexPattern, pollSeconds: config.behavior.pollIntervalSeconds, output: config.output },
+  {
+    ...getServiceMetadata(),
+    indexPattern: config.search.indexPattern,
+    pollSeconds: config.behavior.pollIntervalSeconds,
+    output: config.output,
+  },
   'kaytoo starting',
 );
 
@@ -168,7 +173,6 @@ if (config.output === 'console') {
       { platform: 'matrix', notifier: matrixNotifier, channelId: config.matrix?.defaultRoomId },
       { platform: 'mattermost', notifier: mattermostNotifier, channelId: config.mattermost?.channelId },
     ]),
-    log: getLogger({ component: 'insights.out' }),
   });
 
   const agent = await createAgentRuntime({ config });

src/notify/insightSink.ts

@@ -1,5 +1,3 @@
-import type { Logger } from 'pino';
-import { logErr } from '../logging/logger.js';
 import type { ChatAddress } from '../chat/types.js';
 import type { Notifier } from './notifier.js';
 
@@ -13,18 +11,17 @@ export function createPlatformInsightSink(notifier: Notifier, address: ChatAddre
   };
 }
 
-export function createMultiInsightSink(opts: {
-  sinks: readonly InsightSink[];
-  log: Logger;
-}): InsightSink {
-  const { sinks, log } = opts;
+// Resolves if any sink succeeds; rejects only when every sink fails. Inner
+// notifiers own per-sink failure logging.
+export function createMultiInsightSink(opts: { sinks: readonly InsightSink[] }): InsightSink {
+  const { sinks } = opts;
   return {
     async postInsight(text: string): Promise<void> {
       if (sinks.length === 0) return;
       const results = await Promise.allSettled(sinks.map((s) => s.postInsight(text)));
-      for (const [i, r] of results.entries()) {
-        if (r.status === 'rejected') log.warn({ ...logErr(r.reason), sinkIndex: i }, 'insight sink failed');
-      }
+      const reasons = results.flatMap((r) => (r.status === 'rejected' ? [r.reason] : []));
+      if (reasons.length !== sinks.length) return;
+      throw reasons.length === 1 ? reasons[0] : new AggregateError(reasons, 'all insight sinks failed');
     },
   };
 }

test/insightSink.test.ts

@@ -1,5 +1,4 @@
 import { describe, expect, it, vi } from 'vitest';
-import type { Logger } from 'pino';
 import type { ChatAddress, ChatPost } from '../src/chat/types.js';
 import type { Notifier } from '../src/notify/notifier.js';
 import { createMultiInsightSink, createPlatformInsightSink } from '../src/notify/insightSink.js';
@@ -14,18 +13,6 @@ function recordingNotifier(): Notifier & { calls: ChatPost[] } {
   };
 }
 
-function silentLogger(): Logger & { warns: Array<{ obj: unknown; msg: string }> } {
-  const warns: Array<{ obj: unknown; msg: string }> = [];
-  const log = {
-    warns,
-    warn: (obj: unknown, msg: string) => warns.push({ obj, msg }),
-    info: () => {},
-    error: () => {},
-    debug: () => {},
-  } as unknown as Logger & { warns: Array<{ obj: unknown; msg: string }> };
-  return log;
-}
-
 describe('createPlatformInsightSink', () => {
   it('delegates to the notifier with a fixed ChatAddress', async () => {
     const notifier = recordingNotifier();
@@ -52,30 +39,39 @@ describe('createMultiInsightSink', () => {
   it('fans out to every sink', async () => {
     const a = { postInsight: vi.fn().mockResolvedValue(undefined) };
     const b = { postInsight: vi.fn().mockResolvedValue(undefined) };
-    const sink = createMultiInsightSink({ sinks: [a, b], log: silentLogger() });
+    const sink = createMultiInsightSink({ sinks: [a, b] });
 
     await sink.postInsight('msg');
 
     expect(a.postInsight).toHaveBeenCalledWith('msg');
     expect(b.postInsight).toHaveBeenCalledWith('msg');
   });
 
-  it('continues past a failing sink and logs the rejection', async () => {
+  it('resolves when at least one sink succeeds (inner notifier owns failure logging)', async () => {
     const ok = { postInsight: vi.fn().mockResolvedValue(undefined) };
     const fail = { postInsight: vi.fn().mockRejectedValue(new Error('matrix down')) };
-    const log = silentLogger();
-    const sink = createMultiInsightSink({ sinks: [fail, ok], log });
+    const sink = createMultiInsightSink({ sinks: [fail, ok] });
 
     await expect(sink.postInsight('msg')).resolves.toBeUndefined();
     expect(ok.postInsight).toHaveBeenCalled();
-    expect(log.warns).toHaveLength(1);
-    expect(log.warns[0]?.msg).toBe('insight sink failed');
+  });
+
+  it('rethrows the single reason when only one sink is configured and it fails', async () => {
+    const fail = { postInsight: vi.fn().mockRejectedValue(new Error('matrix down')) };
+    const sink = createMultiInsightSink({ sinks: [fail] });
+    await expect(sink.postInsight('msg')).rejects.toThrow('matrix down');
+  });
+
+  it('rejects with AggregateError when every sink fails', async () => {
+    const a = { postInsight: vi.fn().mockRejectedValue(new Error('a down')) };
+    const b = { postInsight: vi.fn().mockRejectedValue(new Error('b down')) };
+    const sink = createMultiInsightSink({ sinks: [a, b] });
+
+    await expect(sink.postInsight('msg')).rejects.toBeInstanceOf(AggregateError);
   });
 
   it('is a no-op with no sinks', async () => {
-    const log = silentLogger();
-    const sink = createMultiInsightSink({ sinks: [], log });
+    const sink = createMultiInsightSink({ sinks: [] });
     await expect(sink.postInsight('msg')).resolves.toBeUndefined();
-    expect(log.warns).toHaveLength(0);
   });
 });

test/logging.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from 'node:os';
 import pino from 'pino';
 import { describe, expect, it, beforeEach } from 'vitest';
 import { runWithLogContext } from '../src/logging/context.js';
-import { getLogger, initLogging, resetLogging } from '../src/logging/logger.js';
+import { getLogger, getServiceMetadata, initLogging, resetLogging } from '../src/logging/logger.js';
 
 describe('logging', () => {
   beforeEach(() => {
@@ -26,16 +26,19 @@ describe('logging', () => {
     };
   }
 
-  it('emits JSON with service and redacts password', () => {
+  it('emits JSON without process metadata on each line and redacts password', () => {
     const { dest, dir, readLines } = makeSyncFileDest();
     initLogging({ level: 'info', redactPaths: [], destination: dest });
     getLogger({ component: 'test' }).info({ password: 'secret', ok: true }, 'hello');
     const lines = readLines();
     rmSync(dir, { recursive: true, force: true });
     expect(lines.length).toBeGreaterThan(0);
     const row = JSON.parse(lines[0]!) as Record<string, unknown>;
-    expect(row.service).toBe('kaytoo');
-    expect(row.env).toBe('test');
+    expect(row.service).toBeUndefined();
+    expect(row.env).toBeUndefined();
+    expect(row.version).toBeUndefined();
+    expect(row.pid).toBeUndefined();
+    expect(row.hostname).toBeUndefined();
     expect(row.component).toBe('test');
     expect(row.msg).toBe('hello');
     expect(row.password).toBe('[Redacted]');
@@ -55,6 +58,31 @@ describe('logging', () => {
     expect(row.component).toBe('insights');
   });
 
+  it('does not leak per-call fields from one log into later logs in the same context', () => {
+    const { dest, dir, readLines } = makeSyncFileDest();
+    initLogging({ level: 'info', redactPaths: [], destination: dest });
+    runWithLogContext({ pollId: 'pid-leak' }, () => {
+      const log = getLogger({ component: 'insights' });
+      log.warn({ degradedKey: 'alerting', degradedMsg: 'boom' }, 'insights degraded');
+      log.info({ findingCount: 3 }, 'second');
+    });
+    const lines = readLines();
+    rmSync(dir, { recursive: true, force: true });
+    const rows = lines.map((l) => JSON.parse(l) as Record<string, unknown>);
+    expect(rows[0]?.degradedKey).toBe('alerting');
+    expect(rows[1]?.degradedKey).toBeUndefined();
+    expect(rows[1]?.degradedMsg).toBeUndefined();
+    expect(rows[1]?.findingCount).toBe(3);
+    expect(rows[1]?.pollId).toBe('pid-leak');
+  });
+
+  it('exposes static service metadata for one-time startup logging', () => {
+    const meta = getServiceMetadata();
+    expect(meta.service).toBe('kaytoo');
+    expect(meta.env).toBe('test');
+    expect(typeof meta.version).toBe('string');
+  });
+
   it('supports custom redact paths', () => {
     const { dest, dir, readLines } = makeSyncFileDest();
     initLogging({ level: 'info', redactPaths: ['customSecret'], destination: dest });

test/matrixAdapter.test.ts

@@ -18,6 +18,8 @@ vi.mock('matrix-js-sdk', () => ({
     joinRoom: vi.fn().mockResolvedValue({}),
     getUserId: vi.fn().mockReturnValue('@bot:hs'),
     loginRequest: vi.fn().mockResolvedValue({ access_token: 'srv-tok', user_id: '@bot:hs' }),
+    setAccessToken: vi.fn(),
+    credentials: undefined as unknown,
   })),
 }));
 
@@ -254,7 +256,7 @@ describe('startMatrixAdapter', () => {
     await stop();
   });
 
-  it('uses matrix-js-sdk login with m.login.password and no token in createClient', async () => {
+  it('uses matrix-js-sdk login with m.login.password and installs returned credentials', async () => {
     const { stop, client } = await startMatrixAdapter({
       homeserverUrl: 'https://hs',
       auth: { user: 'kaytoo', password: 'pw' },
@@ -273,6 +275,8 @@ describe('startMatrixAdapter', () => {
       password: 'pw',
       initial_device_display_name: 'kaytoo',
     });
+    expect(client.setAccessToken).toHaveBeenCalledWith('srv-tok');
+    expect((client as unknown as { credentials: { userId: string } }).credentials).toEqual({ userId: '@bot:hs' });
     await stop();
   });
 
@@ -303,6 +307,8 @@ describe('startMatrixAdapter', () => {
           joinRoom: vi.fn().mockResolvedValue({}),
           getUserId: vi.fn().mockReturnValue('@bot:hs'),
           loginRequest: vi.fn().mockRejectedValue(new Error('M_FORBIDDEN: Invalid password')),
+          setAccessToken: vi.fn(),
+          credentials: undefined as unknown,
         }) as unknown as MatrixClient,
     );
     await expect(