fix(daemon): drain and finalize traces deterministically

src/callLifecycle.ts

@@ -337,18 +337,22 @@ function finishAgentSpan(
   call: TracedAgent,
   outcome: ToolResult,
   failureType?: string,
+  endTime?: Date,
 ): void {
   const output = outcome.ok ? outcome.output : outcome.error;
   if (output !== undefined && output !== null && output !== '') {
     const text = typeof output === 'string' ? output : jsonStr(output);
     call.span.setAttributes({ [ATTR.OUTPUT_MESSAGES]: assistantOutputMessages([text]) });
   }
   if (outcome.ok) {
-    call.span.end();
+    call.span.end(endTime ? { endTime } : undefined);
     return;
   }
   call.span.setAttributes({ [ATTR.ERROR_TYPE]: failureType ?? errorType(outcome.error) });
-  call.span.end({ error: new Error(outcome.error) });
+  call.span.end({
+    error: new Error(outcome.error),
+    ...(endTime ? { endTime } : {}),
+  });
 }
 
 function errorType(error: string): string {
@@ -434,19 +438,21 @@ export function finalizeOpenCalls(
   state: CallState,
   roots: Iterable<TurnTrace>,
   reason: string,
+  endTime?: Date,
 ): string[] {
   const closed: string[] = [];
   const closeChildren = (parent: CallParent) => {
     for (const call of [...parent.children].reverse()) {
       if (call.kind === 'agent') closeChildren(call);
       if (call.kind === 'agent' && call.outcome) {
-        finishAgentSpan(call, call.outcome);
+        finishAgentSpan(call, call.outcome, undefined, endTime);
       } else if (call.kind === 'agent' && !call.toolUseId && call.stopSeen) {
-        call.span.end();
+        call.span.end(endTime ? { endTime } : undefined);
       } else {
         call.span.setAttributes({ [ATTR.WEAVE_ORPHAN_REASON]: reason });
         call.span.end({
           error: new Error(`call did not complete (${reason})`),
+          ...(endTime ? { endTime } : {}),
         });
       }
       completeCall(state, call, false);

src/daemon.ts

@@ -45,6 +45,9 @@ const MAX_SOCKET_PAYLOAD_BYTES = 4 * 1024 * 1024;
 
 export class Daemon {
   private server?: net.Server;
+  /** Socket inode bound by this process; a successor may reuse the path while
+   * this daemon drains. */
+  private ownedSocketInode?: number;
   private running = false;
   private lastActivity = Date.now();
   private readonly inactivityMs =
@@ -64,6 +67,13 @@ export class Daemon {
   }
 
   async start(): Promise<void> {
+    // Install cleanup before any await can expose a partially started daemon.
+    this.running = true;
+    process.on('SIGTERM', () => void this.shutdown('SIGTERM'));
+    process.on('SIGINT', () => void this.shutdown('SIGINT'));
+    process.on('SIGHUP', () => void this.shutdown('SIGHUP'));
+    process.on('exit', () => this.releaseOwnedSocket());
+
     if (this.config.weaveProject && this.config.apiKey) {
       try {
         await this.initWeave();
@@ -87,20 +97,8 @@ export class Daemon {
     }
 
     await this.bindSocketWithHerdProtection();
-    this.running = true;
     this.log('INFO', `Daemon started — socket: ${this.socketPath}`);
 
-    process.on('SIGTERM', () => void this.shutdown('SIGTERM'));
-    process.on('SIGINT', () => void this.shutdown('SIGINT'));
-    process.on('SIGHUP', () => void this.shutdown('SIGHUP'));
-    process.on('exit', () => {
-      try {
-        if (fs.existsSync(this.socketPath)) fs.unlinkSync(this.socketPath);
-      } catch {
-        // The next hook's socket probe handles stale-socket cleanup.
-      }
-    });
-
     const checkEveryMs = Math.min(
       60_000,
       Math.max(500, Math.floor(this.inactivityMs / 4)),
@@ -135,10 +133,11 @@ export class Daemon {
         reject(err);
       };
       server.once('error', onError);
+      this.server = server;
       server.listen(this.socketPath, () => {
         process.umask(previousUmask);
         server.removeListener('error', onError);
-        this.server = server;
+        this.captureSocketOwnership();
         resolve();
       });
     });
@@ -170,6 +169,31 @@ export class Daemon {
     }
   }
 
+  /** Capture an early bind only while this process's server is still live. */
+  private captureSocketOwnership(): void {
+    if (this.ownedSocketInode !== undefined || !this.server?.listening) return;
+    try {
+      this.ownedSocketInode = fs.statSync(this.socketPath).ino;
+    } catch {
+      // The socket was already removed.
+    }
+  }
+
+  /** Remove only the socket inode this daemon created. */
+  private releaseOwnedSocket(): void {
+    this.captureSocketOwnership();
+    const owned = this.ownedSocketInode;
+    this.ownedSocketInode = undefined;
+    if (owned === undefined) return;
+    try {
+      if (fs.statSync(this.socketPath).ino === owned) {
+        fs.unlinkSync(this.socketPath);
+      }
+    } catch {
+      // The socket was already removed.
+    }
+  }
+
   private async initWeave(): Promise<void> {
     if (!this.config.weaveProject) {
       throw new Error('weaveProject required to init tracer');
@@ -311,7 +335,21 @@ export class Daemon {
 
   private async drain(reason: string): Promise<void> {
     this.log('INFO', `Shutdown: ${reason}`);
-    this.server?.close();
+    // The path can become visible just before listen's callback records it.
+    this.captureSocketOwnership();
+    let serverClosed: Promise<void> | undefined;
+    if (this.server?.listening) {
+      // Node stops listening and unlinks a Unix socket when close() is called;
+      // the callback only waits for already accepted connections.
+      serverClosed = new Promise<void>(resolve =>
+        this.server!.close(() => resolve()));
+      // A successor may now bind this path, so never inspect it again.
+      this.ownedSocketInode = undefined;
+    } else {
+      this.releaseOwnedSocket();
+    }
+    await serverClosed;
+    await this.hookHandler.waitForPendingEvents();
     this.hookHandler.finalizeForShutdown();
     if (this.tracingEnabled) {
       try {
@@ -321,9 +359,6 @@ export class Daemon {
       }
     }
     this.hookHandler.closeTranscripts();
-    if (fs.existsSync(this.socketPath)) {
-      fs.unlinkSync(this.socketPath);
-    }
   }
 
   private log(level: 'DEBUG' | 'INFO' | 'ERROR', message: string): void {

src/hookHandler.ts

@@ -732,6 +732,11 @@ export class HookHandler {
     return false;
   }
 
+  /** Admission must be stopped before taking this snapshot. */
+  async waitForPendingEvents(): Promise<void> {
+    await Promise.all([...this.sessionQueues.values()]);
+  }
+
   finalizeForShutdown(): void {
     for (const session of this.sessions.values()) {
       try {

src/session.ts

@@ -30,6 +30,11 @@ import { TranscriptFile, readFirstTranscriptLine } from './transcriptFile.js';
 
 type TraceLog = (level: 'DEBUG' | 'INFO' | 'ERROR', message: string) => void;
 
+/** Keep forced parent closes after children opened during this clock tick. */
+function spanCloseTime(): Date {
+  return new Date(Date.now() + 1);
+}
+
 export type TurnTrace = {
   kind: 'turn';
   span: weave.Turn;
@@ -190,21 +195,22 @@ export class Session {
   finishAtSessionEnd(promptId: string | undefined): number {
     const parsed = this.parseTranscript();
     this.reconcileFinalTurn(promptId, parsed);
-    return this.finishTurns('session_ended', parsed);
+    return this.finishTurns('session_ended', parsed, spanCloseTime());
   }
 
   finishOpenTurns(orphanReason: string): number {
-    return this.finishTurns(orphanReason, this.parseTranscript());
+    return this.finishTurns(orphanReason, this.parseTranscript(), spanCloseTime());
   }
 
   private finishTurns(
     orphanReason: string,
     parsed: ParsedSession | null,
+    endTime: Date,
   ): number {
     const turnCount = this.turns.size;
     for (const turn of [...this.turns]) {
       this.recordFinalTurnOutput(turn, orphanReason, parsed);
-      this.closeTurn(turn, orphanReason);
+      this.closeTurn(turn, orphanReason, endTime);
     }
     return turnCount;
   }
@@ -414,14 +420,27 @@ export class Session {
 
   private finalizeTurn(turn: TurnTrace, orphanReason: string): void {
     this.recordFinalTurnOutput(turn, orphanReason, this.parseTranscript());
-    this.closeTurn(turn, orphanReason);
+    this.closeTurn(
+      turn,
+      orphanReason,
+      turn.children.size ? spanCloseTime() : undefined,
+    );
   }
 
-  private closeTurn(turn: TurnTrace, orphanReason: string): void {
-    for (const toolUseId of finalizeOpenCalls(this.calls, [turn], orphanReason)) {
+  private closeTurn(
+    turn: TurnTrace,
+    orphanReason: string,
+    endTime?: Date,
+  ): void {
+    for (const toolUseId of finalizeOpenCalls(
+      this.calls,
+      [turn],
+      orphanReason,
+      endTime,
+    )) {
       this.log('DEBUG', `Closed pending call: ${toolUseId}`);
     }
-    turn.span.end();
+    turn.span.end(endTime ? { endTime } : undefined);
     this.turns.delete(turn);
     if (this.currentTurn === turn) this.currentTurn = undefined;
   }

tests/daemon-idle-inflight.test.ts

@@ -92,3 +92,48 @@ test('an open tool keeps a stopped turn alive', async () => {
     await d.stop();
   }
 });
+
+test('shutdown drains queued hooks before finalizing', async () => {
+  const d = await startTestDaemon();
+  try {
+    const sessionId = 'shutdown-queued-hooks';
+    const transcript = writeTranscript(d.home, sessionId);
+    await d.send({
+      hook_event_name: 'SessionStart',
+      session_id: sessionId,
+      transcript_path: transcript,
+    });
+    await d.send({
+      hook_event_name: 'UserPromptSubmit',
+      session_id: sessionId,
+      transcript_path: transcript,
+      prompt: 'queue work',
+    });
+    assert.ok(await d.waitForLog(/Created turn span/));
+
+    // Stop remains in its transcript retry loop while the tool queues behind it.
+    await d.send({
+      hook_event_name: 'Stop',
+      session_id: sessionId,
+      transcript_path: transcript,
+      last_assistant_message: 'not flushed yet',
+    });
+    await d.send({
+      hook_event_name: 'PreToolUse',
+      session_id: sessionId,
+      transcript_path: transcript,
+      tool_use_id: 'queued-tool',
+      tool_name: 'Read',
+      tool_input: { file_path: '/x' },
+    });
+    d.proc.kill('SIGTERM');
+
+    assert.ok(
+      await d.waitForExit(5000),
+      `daemon did not exit; log was:\n${d.readLog()}`,
+    );
+    assert.match(d.readLog(), /Closed pending call: queued-tool/);
+  } finally {
+    await d.stop();
+  }
+});