Delete file store on cancel action

Author: huangjeff5

genkit-tools/common/src/manager/manager.ts

@@ -68,8 +68,6 @@ export class RuntimeManager {
   private filenameToDevUiMap: Record<string, DevToolsInfo> = {};
   private idToFileMap: Record<string, string> = {};
   private eventEmitter = new EventEmitter();
-  private watchers: chokidar.FSWatcher[] = [];
-  private healthCheckInterval?: NodeJS.Timeout;
 
   private constructor(
     readonly telemetryServerUrl: string | undefined,
@@ -93,27 +91,14 @@ export class RuntimeManager {
     await manager.setupRuntimesWatcher();
     await manager.setupDevUiWatcher();
     if (manager.manageHealth) {
-      manager.healthCheckInterval = setInterval(
+      setInterval(
         async () => await manager.performHealthChecks(),
         HEALTH_CHECK_INTERVAL
       );
     }
     return manager;
   }
 
-  /**
-   * Stops the runtime manager and cleans up resources.
-   */
-  async stop() {
-    if (this.healthCheckInterval) {
-      clearInterval(this.healthCheckInterval);
-    }
-    await Promise.all(this.watchers.map((watcher) => watcher.close()));
-    if (this.processManager) {
-      await this.processManager.kill();
-    }
-  }
-
   /**
    * Lists all active runtimes
    */
@@ -365,6 +350,10 @@ export class RuntimeManager {
           },
         }
       );
+
+      // Delete the trace from the telemetry server
+      await this.deleteTrace(input.traceId);
+
       return response.data;
     } catch (err) {
       const axiosError = err as AxiosError;
@@ -382,6 +371,18 @@ export class RuntimeManager {
     }
   }
 
+  /**
+   * Deletes a trace by ID
+   */
+  async deleteTrace(traceId: string): Promise<void> {
+    try {
+      await axios.delete(`${this.telemetryServerUrl}/api/traces/${traceId}`);
+    } catch (err) {
+      // Log but don't fail - trace deletion is best-effort
+      logger.debug(`Failed to delete trace ${traceId}: ${err}`);
+    }
+  }
+
   /**
    * Retrieves all traces
    */
@@ -538,7 +539,6 @@ export class RuntimeManager {
         persistent: true,
         ignoreInitial: false,
       });
-      this.watchers.push(watcher);
       watcher.on('add', (filePath) => this.handleNewRuntime(filePath));
       if (this.manageHealth) {
         watcher.on('unlink', (filePath) => this.handleRemovedRuntime(filePath));
@@ -563,7 +563,6 @@ export class RuntimeManager {
         persistent: true,
         ignoreInitial: false,
       });
-      this.watchers.push(watcher);
       watcher.on('add', (filePath) => this.handleNewDevUi(filePath));
       if (this.manageHealth) {
         watcher.on('unlink', (filePath) => this.handleRemovedDevUi(filePath));

genkit-tools/telemetry-server/src/file-trace-store.ts

@@ -197,6 +197,16 @@ export class LocalFileTraceStore implements TraceStore {
     };
   }
 
+  async delete(traceId: string): Promise<void> {
+    const filePath = path.resolve(this.storeRoot, `${traceId}`);
+    if (fs.existsSync(filePath)) {
+      fs.unlinkSync(filePath);
+      logger.debug(`[Telemetry Server] Deleted trace ${traceId}`);
+    }
+    // Index entry is left orphaned but will be filtered out on load
+    // and cleaned up on next reindex
+  }
+
   private async listFromFiles(query?: TraceQuery): Promise<TraceQueryResponse> {
     const files = fs.readdirSync(this.storeRoot);
     files.sort((a, b) => {

genkit-tools/telemetry-server/src/index.ts

@@ -65,6 +65,16 @@ export async function startTelemetryServer(params: {
     }
   });
 
+  api.delete('/api/traces/:traceId', async (request, response, next) => {
+    try {
+      const { traceId } = request.params;
+      await params.traceStore.delete(traceId);
+      response.status(200).send('OK');
+    } catch (e) {
+      next(e);
+    }
+  });
+
   // SSE endpoint for live trace streaming
   api.get('/api/traces/:traceId/stream', async (request, response, next) => {
     try {

genkit-tools/telemetry-server/src/types.ts

@@ -50,4 +50,5 @@ export interface TraceStore {
   save(traceId: string, trace: TraceData): Promise<void>;
   load(traceId: string): Promise<TraceData | undefined>;
   list(query?: TraceQuery): Promise<TraceQueryResponse>;
+  delete(traceId: string): Promise<void>;
 }

genkit-tools/telemetry-server/tests/broadcast_manager_test.ts

@@ -92,7 +92,8 @@ describe('BroadcastManager', () => {
     const event = createSpanEvent('trace-1');
     manager.broadcast('trace-1', event);
 
-    const expectedData = JSON.stringify(event) + '\n\n';
+    // SSE format: "data: <json>\n\n"
+    const expectedData = `data: ${JSON.stringify(event)}\n\n`;
     assert.strictEqual(response1.writtenData[0], expectedData);
     assert.strictEqual(response2.writtenData[0], expectedData);
   });

genkit-tools/telemetry-server/tests/file_store_test.ts

@@ -269,6 +269,86 @@ describe('local-file-store', () => {
     );
   });
 
+  it('deletes traces', async () => {
+    const traceData = {
+      traceId: TRACE_ID,
+      displayName: 'trace',
+      spans: { [SPAN_A]: span(TRACE_ID, SPAN_A, 100, 100) },
+    } as TraceData;
+
+    // Create the trace
+    const createRes = await fetch(`${url}/api/traces`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify(traceData),
+    });
+    assert.strictEqual(createRes.status, 200);
+
+    // Verify it exists
+    const getRes1 = await fetch(`${url}/api/traces/${TRACE_ID}`);
+    assert.strictEqual(getRes1.status, 200);
+    const retrieved = await getRes1.json();
+    assert.strictEqual(retrieved.traceId, TRACE_ID);
+
+    // Delete the trace
+    const deleteRes = await fetch(`${url}/api/traces/${TRACE_ID}`, {
+      method: 'DELETE',
+    });
+    assert.strictEqual(deleteRes.status, 200);
+
+    // Verify it's gone
+    const getRes2 = await fetch(`${url}/api/traces/${TRACE_ID}`);
+    assert.strictEqual(getRes2.status, 200);
+    const responseText = await getRes2.text();
+    assert.strictEqual(responseText, '');
+  });
+
+  it('deletes trace and filters from list', async () => {
+    // Create multiple traces
+    for (let i = 0; i < 3; i++) {
+      const spanId = `abc_${i}`;
+      const traceId = TRACE_ID + `_${i}`;
+      const spanData = span(traceId, spanId, 100 + i, 100 + i);
+      const trace = {
+        traceId: traceId,
+        displayName: 'trace',
+        spans: { [spanId]: spanData },
+      };
+
+      const res = await fetch(`${url}/api/traces`, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify(trace),
+      });
+      assert.strictEqual(res.status, 200);
+      await sleep(1);
+    }
+
+    // Delete the middle trace
+    const deleteRes = await fetch(`${url}/api/traces/${TRACE_ID}_1`, {
+      method: 'DELETE',
+    });
+    assert.strictEqual(deleteRes.status, 200);
+
+    // List should only return 2 traces
+    const listRes = await fetch(`${url}/api/traces`);
+    assert.strictEqual(listRes.status, 200);
+    const tracesResponse = await listRes.json();
+    assert.strictEqual(tracesResponse.traces.length, 2);
+    assert.ok(
+      tracesResponse.traces.every(
+        (t: TraceData) => t.traceId !== `${TRACE_ID}_1`
+      )
+    );
+  });
+
+  it('delete is idempotent for non-existent trace', async () => {
+    const deleteRes = await fetch(`${url}/api/traces/non-existent-trace`, {
+      method: 'DELETE',
+    });
+    assert.strictEqual(deleteRes.status, 200);
+  });
+
   async function assertTraceData(traceId: string, traceData: TraceData) {
     const getResp = await fetch(`${url}/api/traces/${traceId}`);
     assert.strictEqual(getResp.status, 200);