Skip to content

Commit 79a7425

Browse files
zebbernzebbern
committed
fix(worker): gate workflow diagnostic logs
Signed-off-by: zebbern <185730623+zebbern@users.noreply.github.com>
1 parent 4fc2cd4 commit 79a7425

4 files changed

Lines changed: 111 additions & 26 deletions

File tree

‎worker/src/temporal/__tests__/workflow-runner.test.ts‎

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { beforeAll, describe, expect, it } from 'bun:test';
1+
import { beforeAll, describe, expect, it, vi } from 'bun:test';
22
import { z } from 'zod';
33
import {
44
componentRegistry,
@@ -40,6 +40,74 @@ describe('executeWorkflow', () => {
4040

4141
componentRegistry.register(component);
4242
}
43+
44+
if (!componentRegistry.has('test.silent.echo')) {
45+
const component: ComponentDefinition = {
46+
id: 'test.silent.echo',
47+
label: 'Test Silent Echo',
48+
category: 'transform',
49+
runner: { kind: 'inline' },
50+
inputs: inputs({
51+
value: withPortMeta(z.string(), { label: 'Value' }),
52+
}),
53+
outputs: outputs({
54+
echoed: withPortMeta(z.string(), { label: 'Echoed' }),
55+
}),
56+
async execute({ inputs }) {
57+
return { echoed: inputs.value };
58+
},
59+
};
60+
61+
componentRegistry.register(component);
62+
}
63+
});
64+
65+
it('does not write workflow diagnostics to console.log by default', async () => {
66+
const previousDebugFlag = process.env.SENTRIS_DEBUG_WORKFLOW;
67+
const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {});
68+
69+
try {
70+
delete process.env.SENTRIS_DEBUG_WORKFLOW;
71+
72+
const definition: WorkflowDefinition = {
73+
version: 1,
74+
title: 'Quiet workflow diagnostics',
75+
entrypoint: { ref: 'node-1' },
76+
config: {
77+
environment: 'test',
78+
timeoutSeconds: 30,
79+
},
80+
nodes: {
81+
'node-1': { ref: 'node-1' },
82+
},
83+
edges: [],
84+
dependencyCounts: {
85+
'node-1': 0,
86+
},
87+
actions: [
88+
{
89+
ref: 'node-1',
90+
componentId: 'test.silent.echo',
91+
params: {},
92+
inputOverrides: { value: 'quiet' },
93+
dependsOn: [],
94+
inputMappings: {},
95+
},
96+
],
97+
};
98+
99+
const result = await executeWorkflow(definition, {}, { runId: 'quiet-diagnostics' });
100+
101+
expect(result.success).toBe(true);
102+
expect(logSpy).not.toHaveBeenCalled();
103+
} finally {
104+
if (previousDebugFlag === undefined) {
105+
delete process.env.SENTRIS_DEBUG_WORKFLOW;
106+
} else {
107+
process.env.SENTRIS_DEBUG_WORKFLOW = previousDebugFlag;
108+
}
109+
logSpy.mockRestore();
110+
}
43111
});
44112

45113
it('records trace events with explicit levels in order of execution', async () => {

‎worker/src/temporal/activities/run-workflow.activity.ts‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import type {
88
import type { IFileStorageService, ITraceService, ISecretsService } from '@sentris/component-sdk';
99
import type { ArtifactServiceFactory } from '../artifact-factory';
1010
import { isTraceMetadataAware } from '../utils/trace-metadata';
11+
import { workflowDiagnosticLog } from '../workflow-diagnostics';
1112

1213
interface WorkflowActivityServices {
1314
storage: IFileStorageService;
@@ -48,14 +49,18 @@ function getWorkflowServices(): WorkflowActivityServices {
4849
export async function runWorkflowActivity(
4950
input: RunWorkflowActivityInput,
5051
): Promise<RunWorkflowActivityOutput> {
51-
console.log(`╔══════════════════════════════════════════════════════════════════════════════╗`);
52-
console.log(`🔧 [ACTIVITY START] runWorkflowActivity called`);
53-
console.log(`📋 Run ID: ${input.runId}`);
54-
console.log(`📋 Workflow ID: ${input.workflowId}`);
55-
console.log(`📋 Actions count: ${input.definition.actions.length}`);
56-
console.log(`📋 Action refs: ${input.definition.actions.map((a) => a.ref).join(', ')}`);
57-
console.log(`📋 Inputs keys: ${Object.keys(input.inputs || {}).join(', ')}`);
58-
console.log(`╚══════════════════════════════════════════════════════════════════════════════╝`);
52+
workflowDiagnosticLog(
53+
`╔══════════════════════════════════════════════════════════════════════════════╗`,
54+
);
55+
workflowDiagnosticLog(`🔧 [ACTIVITY START] runWorkflowActivity called`);
56+
workflowDiagnosticLog(`📋 Run ID: ${input.runId}`);
57+
workflowDiagnosticLog(`📋 Workflow ID: ${input.workflowId}`);
58+
workflowDiagnosticLog(`📋 Actions count: ${input.definition.actions.length}`);
59+
workflowDiagnosticLog(`📋 Action refs: ${input.definition.actions.map((a) => a.ref).join(', ')}`);
60+
workflowDiagnosticLog(`📋 Inputs keys: ${Object.keys(input.inputs || {}).join(', ')}`);
61+
workflowDiagnosticLog(
62+
`╚══════════════════════════════════════════════════════════════════════════════╝`,
63+
);
5964
const startTime = Date.now();
6065

6166
const svc = getWorkflowServices();
@@ -68,7 +73,7 @@ export async function runWorkflowActivity(
6873
});
6974
}
7075

71-
console.log(`⏳ [ACTIVITY] About to call executeWorkflow for ${input.runId}`);
76+
workflowDiagnosticLog(`⏳ [ACTIVITY] About to call executeWorkflow for ${input.runId}`);
7277
const result = await executeWorkflow(
7378
input.definition,
7479
{
@@ -88,10 +93,10 @@ export async function runWorkflowActivity(
8893
},
8994
);
9095
const duration = Date.now() - startTime;
91-
console.log(
96+
workflowDiagnosticLog(
9297
`✅ [ACTIVITY DONE] runWorkflow completed for run: ${input.runId} in ${duration}ms`,
9398
);
94-
console.log(`📊 [ACTIVITY] Result keys: ${Object.keys(result || {}).join(', ')}`);
99+
workflowDiagnosticLog(`📊 [ACTIVITY] Result keys: ${Object.keys(result || {}).join(', ')}`);
95100
return result;
96101
} catch (error: unknown) {
97102
const duration = Date.now() - startTime;
@@ -110,7 +115,7 @@ export async function runWorkflowActivity(
110115
throw error;
111116
} finally {
112117
if (isTraceMetadataAware(svc.trace) && typeof svc.trace.finalizeRun === 'function') {
113-
console.log(`🧹 [ACTIVITY] Finalizing trace metadata for ${input.runId}`);
118+
workflowDiagnosticLog(`🧹 [ACTIVITY] Finalizing trace metadata for ${input.runId}`);
114119
svc.trace.finalizeRun(input.runId);
115120
}
116121
}

‎worker/src/temporal/workflow-diagnostics.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
export function shouldLogWorkflowDiagnostics(): boolean {
2+
return process.env.SENTRIS_DEBUG_WORKFLOW === '1';
3+
}
4+
5+
export function workflowDiagnosticLog(...args: Parameters<typeof console.log>): void {
6+
if (shouldLogWorkflowDiagnostics()) {
7+
console.log(...args);
8+
}
9+
}

‎worker/src/temporal/workflow-runner.ts‎

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import { createLightweightSummary } from './utils/component-output';
2828
import { buildActionPayload } from './input-resolver';
2929
import { isComponentFailure, extractFailureMessage } from './workflows/workflow-helpers';
3030
import type { ArtifactServiceFactory } from './artifact-factory';
31+
import { workflowDiagnosticLog } from './workflow-diagnostics';
3132

3233
export interface ExecuteWorkflowOptions {
3334
runId?: string;
@@ -52,9 +53,9 @@ export async function executeWorkflow(
5253
options: ExecuteWorkflowOptions = {},
5354
): Promise<WorkflowRunResult> {
5455
const runId = options.runId ?? randomUUID();
55-
console.log(`🏃 [WORKFLOW RUNNER] executeWorkflow called for runId: ${runId}`);
56-
console.log(`📋 [WORKFLOW RUNNER] Definition has ${definition.actions.length} actions`);
57-
console.log(`📋 [WORKFLOW RUNNER] Entrypoint ref: ${definition.entrypoint.ref}`);
56+
workflowDiagnosticLog(`🏃 [WORKFLOW RUNNER] executeWorkflow called for runId: ${runId}`);
57+
workflowDiagnosticLog(`📋 [WORKFLOW RUNNER] Definition has ${definition.actions.length} actions`);
58+
workflowDiagnosticLog(`📋 [WORKFLOW RUNNER] Entrypoint ref: ${definition.entrypoint.ref}`);
5859

5960
const results = new Map<string, unknown>();
6061
const actionsByRef = new Map<string, (typeof definition.actions)[number]>(
@@ -85,7 +86,7 @@ export async function executeWorkflow(
8586
actionRef: string,
8687
schedulerContext: WorkflowSchedulerRunContext,
8788
): Promise<{ activePorts?: string[] | undefined } | null> => {
88-
console.log(
89+
workflowDiagnosticLog(
8990
`🎯 [WORKFLOW RUNNER] runAction called for: ${actionRef} (triggered by: ${schedulerContext.triggeredBy || 'root'})`,
9091
);
9192

@@ -242,7 +243,7 @@ export async function executeWorkflow(
242243
if (isEntrypointRef && request.inputs) {
243244
// Only apply inputs to the actual entrypoint component, not just any node matching the entrypoint ref
244245
if (isEntrypointComponent) {
245-
console.log(
246+
workflowDiagnosticLog(
246247
`[WorkflowRunner] Applying inputs to entrypoint component '${action.ref}' (${action.componentId})`,
247248
);
248249
inputs.__runtimeData = request.inputs;
@@ -311,14 +312,14 @@ export async function executeWorkflow(
311312
});
312313

313314
try {
314-
console.log(
315+
workflowDiagnosticLog(
315316
`⚡️ [WORKFLOW RUNNER] Executing component: ${action.componentId} for action: ${actionRef}`,
316317
);
317318
const rawOutput = await component.execute(
318319
{ inputs: parsedInputs, params: parsedParams },
319320
context,
320321
);
321-
console.log(
322+
workflowDiagnosticLog(
322323
`✅ [WORKFLOW RUNNER] Component execution completed: ${action.componentId} for action: ${actionRef}`,
323324
);
324325
let output = component.outputs.parse(rawOutput);
@@ -351,7 +352,7 @@ export async function executeWorkflow(
351352
}
352353
}
353354
results.set(action.ref, output);
354-
console.log(`💾 [WORKFLOW RUNNER] Result stored for: ${actionRef}`);
355+
workflowDiagnosticLog(`💾 [WORKFLOW RUNNER] Result stored for: ${actionRef}`);
355356
// Record node I/O completion
356357
await options.nodeIO?.recordCompletion({
357358
runId,
@@ -460,8 +461,8 @@ export async function executeWorkflow(
460461
run: runAction,
461462
});
462463

463-
console.log(`📊 [WORKFLOW RUNNER] runWorkflowWithScheduler completed for ${runId}`);
464-
console.log(`📊 [WORKFLOW RUNNER] Total results stored: ${results.size}`);
464+
workflowDiagnosticLog(`📊 [WORKFLOW RUNNER] runWorkflowWithScheduler completed for ${runId}`);
465+
workflowDiagnosticLog(`📊 [WORKFLOW RUNNER] Total results stored: ${results.size}`);
465466

466467
const outputsObject: Record<string, unknown> = {};
467468
let reportedFailure = false;
@@ -476,8 +477,10 @@ export async function executeWorkflow(
476477
}
477478
});
478479

479-
console.log(`📊 [WORKFLOW RUNNER] Output keys: ${Object.keys(outputsObject).join(', ')}`);
480-
console.log(`📊 [WORKFLOW RUNNER] Reported failure: ${reportedFailure}`);
480+
workflowDiagnosticLog(
481+
`📊 [WORKFLOW RUNNER] Output keys: ${Object.keys(outputsObject).join(', ')}`,
482+
);
483+
workflowDiagnosticLog(`📊 [WORKFLOW RUNNER] Reported failure: ${reportedFailure}`);
481484

482485
if (reportedFailure) {
483486
const baseMessage = 'One or more workflow actions failed';
@@ -492,7 +495,7 @@ export async function executeWorkflow(
492495
};
493496
}
494497

495-
console.log(`✅ [WORKFLOW RUNNER] Workflow completed successfully for ${runId}`);
498+
workflowDiagnosticLog(`✅ [WORKFLOW RUNNER] Workflow completed successfully for ${runId}`);
496499
return { outputs: outputsObject, success: true };
497500
} catch (error: unknown) {
498501
console.error(`❌ [WORKFLOW RUNNER] Workflow threw exception for ${runId}:`, error);

0 commit comments

Comments
 (0)