feat(monitoring): add log flushing before status changes

- Introduced `flush` method in `AgentLoggerService` to ensure log data is flushed to disk for improved readability.
- Updated status methods (`complete`, `fail`, `pause`, `markSkipped`) to invoke log flushing prior to state updates.
- Enhanced stream closing logic to wait for flush completion before releasing file locks.

Author: hadywalied

src/agents/monitoring/logger.ts

@@ -105,6 +105,24 @@ export class AgentLoggerService {
     stream.write(data);
   }
 
+  /**
+   * Flush buffered data to disk for an agent's log
+   * Call this before status changes to ensure data is readable by the UI
+   */
+  async flush(agentId: number): Promise<void> {
+    const stream = this.activeStreams.get(agentId);
+    if (!stream || !stream.writable) {
+      return;
+    }
+
+    return new Promise<void>((resolve) => {
+      // Write empty string with callback - ensures buffer is flushed to OS
+      stream.write('', 'utf-8', () => {
+        resolve();
+      });
+    });
+  }
+
   /**
    * Close an agent's log stream and release file lock
    */
@@ -114,15 +132,19 @@ export class AgentLoggerService {
       const monitor = AgentMonitorService.getInstance();
       const agent = monitor.getAgent(agentId);
 
-      // Release lock FIRST
-      if (agent) {
-        await this.lockService.releaseLock(agent.logPath);
-      }
+      // First, close stream and wait for it to finish flushing
+      await new Promise<void>((resolve) => {
+        stream.on('finish', resolve);
+        stream.end();
+      });
 
-      // Then close stream
-      stream.end();
       this.activeStreams.delete(agentId);
       logger.debug(`Closed log stream for agent ${agentId}`);
+
+      // Release lock AFTER stream is fully closed
+      if (agent) {
+        await this.lockService.releaseLock(agent.logPath);
+      }
     }
   }
 

src/agents/monitoring/status.ts

@@ -1,4 +1,5 @@
 import { AgentMonitorService } from './monitor.js';
+import { AgentLoggerService } from './logger.js';
 import type { AgentStatus } from './types.js';
 import type { WorkflowEventEmitter } from '../../workflows/events/emitter.js';
 import type { ParsedTelemetry } from '../../shared/telemetry/index.js';
@@ -62,22 +63,27 @@ export class StatusService {
   // Mark skipped by monitoring ID (for cleanup coordination)
   async markSkipped(monitoringId: number): Promise<void> {
     this.skippedIds.add(monitoringId);
+    await AgentLoggerService.getInstance().flush(monitoringId);
     await this.monitor.markSkipped(monitoringId);
     this.emitStatus(monitoringId, 'skipped');
   }
 
   // Status updates (DB + UI)
+  // All status changes flush the log first to ensure data is readable
   async complete(id: number, telemetry?: ParsedTelemetry): Promise<void> {
+    await AgentLoggerService.getInstance().flush(id);
     await this.monitor.complete(id, telemetry);
     this.emitStatus(id, 'completed');
   }
 
   async fail(id: number, error: Error | string): Promise<void> {
+    await AgentLoggerService.getInstance().flush(id);
     await this.monitor.fail(id, error);
     this.emitStatus(id, 'failed');
   }
 
   async pause(id: number): Promise<void> {
+    await AgentLoggerService.getInstance().flush(id);
     await this.monitor.markPaused(id);
     this.emitStatus(id, 'paused');
   }