diff --git a/CHANGELOG.md b/CHANGELOG.md index 8660b28..4df83bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,23 @@ platform API version. `elk-server-0.3.0-all.jar`**, and the server now needs **Java 17** (ELK 0.10 is built for it); without it layouts stay on elkjs, as before. +### Fixed + +- **The running ring and the queue sizes no longer blink through a run.** The + runtime writes `run.wf-viewer.live.json` at every firing and the server + re-reads it every 400ms, so now and then it read the file halfway written, + took that for no run at all and fell through to the previous run: the ring + went off, the previous run's queue sizes came on, and the next read put + things back. A run-output file that does not parse is now read as it last + did (the queue trace too), and from the end of a run until its record is in + `run-log.jsonl` its last live overlay stands, as finished, where the + previous run showed for a moment. +- **The eye on the palette, off, takes the run's tokens and queue sizes off + the edges.** It took only the step counter away: the carried colour, the + token pills and the sizes stayed, so it looked as if it did nothing. The + nodes keep the run's state (the ring is the live run, with its own + setting). A queue size the trace no longer has is taken off the edge too. + ### Tooling - `scripts/render-svg --select ` selects an element (the diagram's id or diff --git a/packages/diagram-client/src/tool-palette-header-controls.ts b/packages/diagram-client/src/tool-palette-header-controls.ts index a5c4d22..acc9aa6 100644 --- a/packages/diagram-client/src/tool-palette-header-controls.ts +++ b/packages/diagram-client/src/tool-palette-header-controls.ts @@ -235,7 +235,7 @@ function patchHeaderTools(headerTools: HTMLElement): void { queueIcon.classList.toggle('codicon-eye', visible); queueIcon.classList.toggle('codicon-eye-closed', !visible); queueIcon.classList.toggle('cal-queue-hidden-icon', !visible); - queueIcon.title = visible ? 'Hide queue overlay' : 'Show queue overlay'; + queueIcon.title = visible ? 'Hide what the run put on the edges (tokens, queue sizes)' : 'Show what the run put on the edges (tokens, queue sizes)'; queueIcon.setAttribute('aria-label', queueIcon.title); }; diff --git a/packages/diagram-server/src/server/source-model-storage.ts b/packages/diagram-server/src/server/source-model-storage.ts index 5080885..0bb2564 100644 --- a/packages/diagram-server/src/server/source-model-storage.ts +++ b/packages/diagram-server/src/server/source-model-storage.ts @@ -143,6 +143,25 @@ export function countCoincidentNodes(positions: Iterable<{ x: number; y: number * Source model storage for Workflow diagrams. * Acquires the graph via the injected DiagramModelSource seam and transforms it to GModel. */ +/** Take off an edge what a run put on it: the token it carried, its queue size at a step. */ +function clearEdgeOverlay(edge: GEdge): void { + const args = edge.args as Record | undefined; + if (args) { + delete args[WorkflowDiagramMetadata.VIEWER_LAST_TOKEN]; + delete args[WorkflowDiagramMetadata.QUEUE_SIZE]; + } +} + +/** Call `fn` on every edge under `element`. */ +function walkGEdges(element: { children?: unknown[] }, fn: (edge: GEdge) => void): void { + for (const child of element.children ?? []) { + if (child instanceof GEdge) { + fn(child); + } + walkGEdges(child as { children?: unknown[] }, fn); + } +} + @injectable() export class WorkflowSourceModelStorage implements SourceModelStorage { @@ -195,6 +214,24 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { */ private readonly runLogEntryCache = new Map> }>(); + /** + * What each run-output JSON file last parsed to, by path, keyed on the file's size and mtime + * (see {@link readJsonFileMemo}). A run writes `run.wf-viewer.live.json` at every firing and the + * diagram re-reads it every 400ms, so now and then it reads the file halfway written, which does + * not parse. That used to count as no file: the loader fell through to the previous run, the + * ring went off and the previous run's queue sizes came on, and the next read put things back. + * A file that does not parse is now read as it last did. + */ + private readonly jsonFileMemo = new Map(); + + /** + * The live overlay last read from each run-output dir, as the file held it. When a run ends + * its live file goes (or says it is not running) before the run's record is in `run-log.jsonl`, + * and in between the loader found only the previous run. While the log has no record of the + * run, its last live overlay stands, as finished. + */ + private readonly lastLiveOverlay = new Map(); + /** * Short-TTL memo for the run-output *workspace* fallback scans (see * {@link findRunOutputFallbackFiles}). The fallback only fires when the primary per-out-dir scan @@ -1076,6 +1113,7 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { console.log(`[WorkflowSourceModelStorage] applyViewerOverlay: no overlay found for workflowFile=${workflowFilePath} workflow=${workflowName}`); // Clear any stale glow state left from a previous cycle. this.clearExecutionGlowState(root); + walkGEdges(root, clearEdgeOverlay); return; } const activeNamesLog = overlay.active @@ -1093,6 +1131,10 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { const queueTrace = queueTraceVisible === false ? undefined : await this.tryLoadQueueTraceAtStep(overlay.outDir, queueTraceStep); + // The eye on the palette, off: nothing the run put on the edges is drawn, neither the + // token an edge carried nor the queue size at a step. The nodes keep the run's state; + // the ring is the live run, with its own setting. + const edgeOverlayHidden = queueTraceVisible === false; if (queueTrace) { (root as any).args['wf:queueTraceStep'] = queueTrace.selectedStep; (root as any).args['wf:queueTraceStepCount'] = queueTrace.stepCount; @@ -1219,7 +1261,9 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { if (!element) { return; } - if (element instanceof GEdge) { + if (element instanceof GEdge && edgeOverlayHidden) { + clearEdgeOverlay(element); + } else if (element instanceof GEdge) { const args = element.args as Record | undefined; let queueSizeForEdge: number | undefined; let queueTokenForEdge: unknown; @@ -1277,6 +1321,8 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { if (queueSizeForEdge !== undefined) { element.args = element.args || {}; (element.args as any)[WorkflowDiagramMetadata.QUEUE_SIZE] = queueSizeForEdge; + } else if (element.args) { + delete (element.args as any)[WorkflowDiagramMetadata.QUEUE_SIZE]; } } if (element instanceof GNode) { @@ -2188,6 +2234,39 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { return runs; } + /** + * The JSON a run-output file holds, through {@link jsonFileMemo}: `undefined` when there is no + * file. A file that fails to parse is one being written, and is answered with what it last + * parsed to, or `undefined` when it never did. + */ + private async readJsonFileMemo(filePath: string): Promise { + const fs = await import('fs/promises'); + let stat: import('fs').Stats; + try { + stat = await fs.stat(filePath); + } catch { + this.jsonFileMemo.delete(filePath); + return undefined; + } + const cached = this.jsonFileMemo.get(filePath); + if (cached && cached.mtimeMs === stat.mtimeMs && cached.size === stat.size) { + return cached.value; + } + let value: unknown; + try { + value = JSON.parse(await fs.readFile(filePath, 'utf-8')); + } catch { + return cached?.value; + } + this.jsonFileMemo.set(filePath, { mtimeMs: stat.mtimeMs, size: stat.size, value }); + return value; + } + + /** Whether `run-log.jsonl` at `runLogPath` records the run `runId`. */ + private async runLogHasRun(runLogPath: string, runId: string): Promise { + return (await this.readRunLogEntries(runLogPath)).some(entry => entry.runId === runId); + } + private async tryLoadQueueTraceAtStep( outDir: string, requestedStep?: number @@ -2207,23 +2286,10 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { lastToken?: unknown; }>; } | undefined> { - const fs = await import('fs/promises'); const path = await import('path'); const queueTracePath = path.join(outDir, 'run.wf-queues.json'); - let text: string; - try { - text = await fs.readFile(queueTracePath, 'utf-8'); - } catch { - return undefined; - } - - let trace: any; - try { - trace = JSON.parse(text); - } catch { - return undefined; - } + const trace: any = await this.readJsonFileMemo(queueTracePath); if (!trace || typeof trace !== 'object' || trace.version !== 1 || !Array.isArray(trace.steps) || trace.steps.length === 0) { return undefined; @@ -2335,30 +2401,18 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { return typeof candidate === 'string' && wantWorkflowNames.has(candidate.trim()); }; - const readOverlayFromPath = async ( - overlayPath: string, - requireRunning = false - ): Promise<{ + type ViewerOverlay = { runId: string; outDir: string; running: boolean; edges: Record; active?: Array<{ entityInstanceName?: string; entityInstancePath?: string[] }>; error?: { entityInstanceName?: string; entityInstancePath?: string[]; message?: string }; - } | undefined> => { - let overlayText: string; - try { - overlayText = await fs.readFile(overlayPath, 'utf-8'); - } catch { - return undefined; - } + }; - let overlay: any; - try { - overlay = JSON.parse(overlayText); - } catch { - return undefined; - } + // The overlay `overlay` (a file's JSON) is, for this view: undefined when it is not one, or + // is another run's, another workflow's, or (`requireRunning`) a finished run's. + const acceptOverlay = (overlay: any, overlayPath: string, requireRunning = false): ViewerOverlay | undefined => { if (!overlay || typeof overlay !== 'object') { return undefined; } @@ -2425,14 +2479,10 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { }; }; - const readOverlay = async (entry: { workflowName?: string; sourcePath?: string; outDir?: string; runId?: string }): Promise<{ - runId: string; - outDir: string; - running: boolean; - edges: Record; - active?: Array<{ entityInstanceName?: string; entityInstancePath?: string[] }>; - error?: { entityInstanceName?: string; entityInstancePath?: string[]; message?: string }; - } | undefined> => { + const readOverlayFromPath = async (overlayPath: string, requireRunning = false): Promise => + acceptOverlay(await this.readJsonFileMemo(overlayPath), overlayPath, requireRunning); + + const readOverlay = async (entry: { workflowName?: string; sourcePath?: string; outDir?: string; runId?: string }): Promise => { if (preferredRunId && entry.runId !== preferredRunId) { return undefined; } @@ -2464,11 +2514,22 @@ export class WorkflowSourceModelStorage implements SourceModelStorage { for (const baseOutDir of baseOutDirs) { if (liveGlowEnabled) { const liveOverlayPath = path.join(baseOutDir, 'run.wf-viewer.live.json'); - const liveOverlay = await readOverlayFromPath(liveOverlayPath, true); + const liveJson = await this.readJsonFileMemo(liveOverlayPath); + const liveOverlay = acceptOverlay(liveJson, liveOverlayPath, true); console.log(`[WorkflowSourceModelStorage] viewer live overlay: ${liveOverlayPath} found=${!!liveOverlay}`); if (liveOverlay) { + this.lastLiveOverlay.set(baseOutDir, liveJson); return liveOverlay; } + // The run has ended: its live file says so, or is gone. Until the log records the + // run the loader would find the previous one, so the run's last live overlay + // stands, finished and with nothing active (what the run's final overlay says + // of it comes with the record). + const ended = acceptOverlay(liveJson ?? this.lastLiveOverlay.get(baseOutDir), liveOverlayPath); + if (ended && !(await this.runLogHasRun(path.join(baseOutDir, 'run-log.jsonl'), ended.runId))) { + console.log(`[WorkflowSourceModelStorage] viewer live overlay: run ${ended.runId} ended, not yet in the log`); + return { ...ended, running: false, active: undefined }; + } } const runLogPath = path.join(baseOutDir, 'run-log.jsonl'); const entries = await this.readRunLogEntries(runLogPath); diff --git a/packages/diagram-server/test/run-overlay-steady-during-run.test.ts b/packages/diagram-server/test/run-overlay-steady-during-run.test.ts new file mode 100644 index 0000000..0e10bb2 --- /dev/null +++ b/packages/diagram-server/test/run-overlay-steady-during-run.test.ts @@ -0,0 +1,172 @@ +/** + * The run overlay holds steady while a run is on, and at its end. + * + * The runtime writes `run.wf-viewer.live.json` at every firing and the diagram + * re-reads it every 400ms, so now and then it reads the file halfway written. + * That used to count as no file: the loader fell through to the previous run, + * the ring went off and the previous run's queue sizes came on, and the next + * read put things back -- the ring and the sizes blinked all through a run. + * At the end of a run the live file went before the run was in the log, and + * for a moment the previous run showed again. + * + * And the eye on the palette, off, took only the step counter away: the tokens + * and queue sizes stayed on the edges, so it looked as if it did nothing. + */ +import { afterEach, describe, expect, it } from 'vitest'; +import * as fs from 'node:fs/promises'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { GEdge, GModelRoot } from '@eclipse-glsp/server'; +import { WorkflowDiagramMetadata } from '@dialogram/shared'; +import { WorkflowSourceModelStorage } from '../src/server/source-model-storage'; + +const tempDirs: string[] = []; +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map(dir => fs.rm(dir, { recursive: true, force: true }))); +}); + +function makeStorage(): any { + const storage = new (WorkflowSourceModelStorage as any)(); + storage.storageOptions = { settingsNamespace: 'wfLang', operationPrefix: 'wfpy' }; + return storage; +} + +/** `/wf.py` with `wf-out/`: a finished run `old` in the log, and the files `run` wrote. */ +async function seed(): Promise<{ workflowFile: string; outDir: string; livePath: string; logPath: string; sourcePath: string }> { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'overlay-steady-')); + tempDirs.push(root); + const workflowFile = path.join(root, 'wf.py'); + await fs.writeFile(workflowFile, '# wf stub\n', 'utf-8'); + const sourcePath = path.resolve(workflowFile); + const outDir = path.join(root, 'wf-out'); + const oldDir = path.join(outDir, 'old'); + await fs.mkdir(oldDir, { recursive: true }); + await fs.writeFile(path.join(oldDir, 'run.wf-viewer.json'), JSON.stringify({ + version: 1, runId: 'old', outDir: oldDir, workflowName: 'wf', sourcePath, running: false, edges: { ab: { lastToken: 'old-token' } } + })); + const logPath = path.join(outDir, 'run-log.jsonl'); + await fs.writeFile(logPath, JSON.stringify({ runId: 'old', workflowName: 'wf', sourcePath, outDir: oldDir, finishedAt: '2026-07-20T00:00:00Z' }) + '\n'); + return { workflowFile, outDir, livePath: path.join(outDir, 'run.wf-viewer.live.json'), logPath, sourcePath }; +} + +const liveOverlay = (sourcePath: string, outDir: string, running: boolean, active?: string) => ({ + version: 1, runId: 'run', outDir: path.join(outDir, 'run'), workflowName: 'wf', sourcePath, running, + edges: { ab: { lastToken: 'run-token' } }, + ...(active ? { active: [{ entityInstanceName: active, fireCount: 3 }] } : {}) +}); + +describe('the live overlay while a run is on', () => { + it('is read as it last parsed when the file is halfway written', async () => { + const { workflowFile, outDir, livePath, sourcePath } = await seed(); + const storage = makeStorage(); + await fs.writeFile(livePath, JSON.stringify(liveOverlay(sourcePath, outDir, true, 'b'))); + const first = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(first?.runId).toBe('run'); + expect(first?.running).toBe(true); + + // The runtime is rewriting the file: a read sees the first half of it. + await fs.writeFile(livePath, JSON.stringify(liveOverlay(sourcePath, outDir, true, 'c')).slice(0, 40)); + const during = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(during?.runId).toBe('run'); + expect(during?.running).toBe(true); + expect(during?.active?.[0]?.entityInstanceName).toBe('b'); + + // Written in full: the new state. + await fs.writeFile(livePath, JSON.stringify(liveOverlay(sourcePath, outDir, true, 'c'))); + const after = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(after?.active?.[0]?.entityInstanceName).toBe('c'); + }); + + it('stands, finished, from the end of the run until the log records it', async () => { + const { workflowFile, outDir, livePath, logPath, sourcePath } = await seed(); + const storage = makeStorage(); + await fs.writeFile(livePath, JSON.stringify(liveOverlay(sourcePath, outDir, true, 'b'))); + expect((await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']))?.runId).toBe('run'); + + // The run stops: the live file says so first ... + await fs.writeFile(livePath, JSON.stringify(liveOverlay(sourcePath, outDir, false))); + const stopped = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(stopped?.runId).toBe('run'); + expect(stopped?.running).toBe(false); + expect(stopped?.active).toBeUndefined(); + expect(stopped?.edges.ab.lastToken).toBe('run-token'); + + // ... then goes, before the run is in the log. + await fs.rm(livePath); + const gone = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(gone?.runId).toBe('run'); + expect(gone?.running).toBe(false); + + // The log records the run: its final overlay is what shows from now on. + const runDir = path.join(outDir, 'run'); + await fs.mkdir(runDir, { recursive: true }); + await fs.writeFile(path.join(runDir, 'run.wf-viewer.json'), JSON.stringify({ + version: 1, runId: 'run', outDir: runDir, workflowName: 'wf', sourcePath, running: false, edges: { ab: { lastToken: 'final-token' } } + })); + await fs.appendFile(logPath, JSON.stringify({ runId: 'run', workflowName: 'wf', sourcePath, outDir: runDir, finishedAt: '2026-07-21T00:00:00Z' }) + '\n'); + const recorded = await storage.tryLoadLatestViewerOverlay(workflowFile, 'wf', [workflowFile], ['wf']); + expect(recorded?.runId).toBe('run'); + expect(recorded?.edges.ab.lastToken).toBe('final-token'); + }); + + it('a queue trace halfway written is read as it last parsed', async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'overlay-steady-trace-')); + tempDirs.push(dir); + const storage = makeStorage(); + const tracePath = path.join(dir, 'run.wf-queues.json'); + const trace = (steps: number) => ({ version: 1, steps: Array.from({ length: steps }, (_, i) => ({ step: i + 1, queueSizes: [{ overlayAstPath: 'ab', size: i }] })) }); + await fs.writeFile(tracePath, JSON.stringify(trace(2))); + expect((await storage.tryLoadQueueTraceAtStep(dir))?.stepCount).toBe(2); + await fs.writeFile(tracePath, JSON.stringify(trace(3)).slice(0, 30)); + expect((await storage.tryLoadQueueTraceAtStep(dir))?.stepCount).toBe(2); + await fs.writeFile(tracePath, JSON.stringify(trace(3))); + expect((await storage.tryLoadQueueTraceAtStep(dir))?.stepCount).toBe(3); + }); +}); + +describe('the eye on the palette, off', () => { + function model(): { root: GModelRoot; ab: GEdge } { + const root = new GModelRoot(); + root.id = 'root'; + const ab = new GEdge(); + ab.id = 'ab'; + // Stale from the last model: a size the trace no longer has. + ab.args = { astPath: 'ab', [WorkflowDiagramMetadata.QUEUE_SIZE]: 7 }; + root.children = [ab]; + return { root, ab }; + } + + async function apply(visible: boolean): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'overlay-eye-')); + tempDirs.push(dir); + await fs.writeFile(path.join(dir, 'run.wf-queues.json'), JSON.stringify({ + version: 1, steps: [{ step: 1, actorInstanceName: 'a', queueSizes: [{ overlayAstPath: 'ab', size: 2, lastToken: 'tok' }] }] + })); + const storage = makeStorage(); + storage.isLiveExecutionGlowEnabled = async () => true; + storage.tryLoadLatestViewerOverlay = async () => ({ runId: 'r', outDir: dir, running: false, active: [], edges: { ab: { lastToken: 'tok' } } }); + storage.ensureModelSource = () => ({ + analysis: { + buildOverlayAstPathCandidates: (args: any) => (args?.astPath ? [args.astPath] : []), + buildOverlaySignatureCandidates: () => [], + buildOverlayNodeIdentityCandidates: () => [], + resolveOverlayActiveEntityName: () => undefined + } + }); + const m = model(); + await storage.applyViewerOverlayToEdges(m.root, '/w/flow.py', 'flow', [], ['flow'], undefined, visible, ['/w/flow.py']); + return m.ab; + } + + it('takes the token and the queue size off the edges', async () => { + const ab = await apply(false); + expect((ab.args as any)[WorkflowDiagramMetadata.VIEWER_LAST_TOKEN]).toBeUndefined(); + expect((ab.args as any)[WorkflowDiagramMetadata.QUEUE_SIZE]).toBeUndefined(); + }); + + it('on, draws them, the size the trace has and not a stale one', async () => { + const ab = await apply(true); + expect((ab.args as any)[WorkflowDiagramMetadata.VIEWER_LAST_TOKEN]).toBe('tok'); + expect((ab.args as any)[WorkflowDiagramMetadata.QUEUE_SIZE]).toBe(2); + }); +});