Repository navigation
fix(workflows): make replay journal durable #8735
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 22 commits
9af160e
a562d8b
2c735cf
66424a8
d34f64d
d51e4d1
78ea279
5fcf726
fdc3cad
a5a4366
33f3885
3aef43e
23bbe16
c4862ae
8bfd654
0172874
b8d1500
ec532c6
5cc7242
c2c0eb3
e09b429
56b13a0
a40ba51
f8c1278
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||
|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -11,7 +11,13 @@ import path from 'node:path'; | |||||||||
| import { workflowsCommand } from './workflowsCommand.js'; | ||||||||||
| import { type CommandContext } from './types.js'; | ||||||||||
| import { createMockCommandContext } from '../../test-utils/mockCommandContext.js'; | ||||||||||
| import type { WorkflowTask, WorkflowSnapshot } from '@qwen-code/qwen-code-core'; | ||||||||||
| import { | ||||||||||
| Storage, | ||||||||||
| writeWorkflowManifest, | ||||||||||
| type Config, | ||||||||||
| type WorkflowTask, | ||||||||||
| type WorkflowSnapshot, | ||||||||||
| } from '@qwen-code/qwen-code-core'; | ||||||||||
|
|
||||||||||
| function entry(overrides: Partial<WorkflowTask> = {}): WorkflowTask { | ||||||||||
| return { | ||||||||||
|
|
@@ -536,7 +542,9 @@ describe('workflowsCommand', () => { | |||||||||
|
|
||||||||||
| // Write snapshot JSON files into a fresh temp dir and return a context | ||||||||||
| // whose `config.storage.getWorkflowRunsDir()` points at it. This drives | ||||||||||
| // the real `listWorkflowSnapshots` (no module mocking). | ||||||||||
| // the real `listWorkflowRunRecords` (no module mocking) — note the | ||||||||||
| // different bad-record semantics: an unparseable file surfaces as a | ||||||||||
| // visible `interrupted · not recoverable` row rather than being skipped. | ||||||||||
| async function ctxWithSnapshots( | ||||||||||
| snaps: Array<Partial<WorkflowSnapshot>>, | ||||||||||
| mode: 'interactive' | 'non_interactive' = 'interactive', | ||||||||||
|
|
@@ -562,6 +570,51 @@ describe('workflowsCommand', () => { | |||||||||
| } as unknown as Parameters<typeof createMockCommandContext>[0]); | ||||||||||
| } | ||||||||||
|
|
||||||||||
| async function ctxWithManifest( | ||||||||||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Deferred (test hygiene). The stale ctxWithSnapshots harness constant after the listing migration; cleaned with the CLI test follow-up. 中文说明推迟(测试卫生)。列表迁移后遗留的 ctxWithSnapshots 脚手架常量;随 CLI 测试后续轮次清理。 |
||||||||||
| run: Partial<WorkflowTask>, | ||||||||||
| journal: { | ||||||||||
| integrity: 'complete' | 'failed'; | ||||||||||
| error?: string; | ||||||||||
| } = { integrity: 'complete' }, | ||||||||||
| ): Promise<CommandContext> { | ||||||||||
| const root = await fs.mkdtemp(path.join(os.tmpdir(), 'wf-manifest-cli-')); | ||||||||||
| tmpDirs.push(root); | ||||||||||
| const storage = new Storage(root, path.join(root, 'runtime')); | ||||||||||
| const config = { | ||||||||||
| storage, | ||||||||||
| getWorkflowRunRegistry: () => ({ | ||||||||||
| list: listMock, | ||||||||||
| get: getMock, | ||||||||||
| pause: pauseMock, | ||||||||||
| resume: resumeMock, | ||||||||||
| }), | ||||||||||
| }; | ||||||||||
| await writeWorkflowManifest( | ||||||||||
| config as unknown as Config, | ||||||||||
| entry({ | ||||||||||
| runId: 'wf_deadbeef', | ||||||||||
| status: 'running', | ||||||||||
| script: 'return 1;', | ||||||||||
| ...run, | ||||||||||
| }), | ||||||||||
| { | ||||||||||
| args: null, | ||||||||||
| journal: { | ||||||||||
| version: 1, | ||||||||||
| keyVersion: 'v2', | ||||||||||
| byteLength: 0, | ||||||||||
| sha256: | ||||||||||
| 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855', | ||||||||||
| ...journal, | ||||||||||
| }, | ||||||||||
| }, | ||||||||||
| ); | ||||||||||
| return createMockCommandContext({ | ||||||||||
| services: { config }, | ||||||||||
| executionMode: 'interactive', | ||||||||||
| } as unknown as Parameters<typeof createMockCommandContext>[0]); | ||||||||||
| } | ||||||||||
|
|
||||||||||
| afterEach(async () => { | ||||||||||
| await Promise.all( | ||||||||||
| tmpDirs | ||||||||||
|
|
@@ -585,6 +638,23 @@ describe('workflowsCommand', () => { | |||||||||
| expect(result.content).toContain('oldrun'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| it.runIf(process.platform !== 'win32')( | ||||||||||
| 'does not render a control-character legacy filename', | ||||||||||
| async () => { | ||||||||||
| listMock.mockReturnValue([]); | ||||||||||
| const ctx = await ctxWithSnapshots([ | ||||||||||
| { runId: 'wf_bad\nCONTROL_SENTINEL' }, | ||||||||||
| ]); | ||||||||||
| const result = await workflowsCommand.action!(ctx, ''); | ||||||||||
| expect(result).toMatchObject({ | ||||||||||
| type: 'message', | ||||||||||
| messageType: 'info', | ||||||||||
| content: 'No workflow runs recorded yet.', | ||||||||||
| }); | ||||||||||
| expect(JSON.stringify(result)).not.toContain('CONTROL_SENTINEL'); | ||||||||||
| }, | ||||||||||
| ); | ||||||||||
|
|
||||||||||
| it('live registry entry shadows a same-runId snapshot (no duplicate row)', async () => { | ||||||||||
| listMock.mockReturnValue([ | ||||||||||
| entry({ | ||||||||||
|
|
@@ -668,5 +738,138 @@ describe('workflowsCommand', () => { | |||||||||
| content: 'Unknown live workflow runId: wf_ghost', | ||||||||||
| }); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| it('classifies an orphaned active manifest as interrupted and recoverable', async () => { | ||||||||||
| listMock.mockReturnValue([]); | ||||||||||
| const ctx = await ctxWithManifest({ | ||||||||||
| runId: 'wf_deadbeef', | ||||||||||
| status: 'paused', | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| const result = await workflowsCommand.action!(ctx, ''); | ||||||||||
| if (!result || result.type !== 'message') throw new Error('no result'); | ||||||||||
| expect(result.content).toContain('Workflow runs (1 total · 0 active)'); | ||||||||||
|
Comment on lines
+749
to
+751
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [Suggestion] The
Suggested change
中文说明[Suggestion] 详情视图的 — DeepSeek/deepseek-v4-flash via Qwen Code /review (v0.21.8)
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Deferred (test gap, CLI detail view). The recoverable/not-recoverable arm of the detail view has no direct assertion; tracked with the CLI test follow-up. 中文说明推迟(测试缺口,CLI 详情视图)。详情视图的 recoverable/not-recoverable 分支无直接断言;随 CLI 测试后续轮次跟进。 |
||||||||||
| expect(result.content).toContain('wf_deadbeef'); | ||||||||||
| expect(result.content).toContain('interrupted'); | ||||||||||
| expect(result.content).toContain('· recoverable'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| it('shows a durable recovery failure reason in detail', async () => { | ||||||||||
| getMock.mockReturnValue(undefined); | ||||||||||
| const ctx = await ctxWithManifest( | ||||||||||
| { runId: 'wf_badc0de', status: 'running' }, | ||||||||||
| { integrity: 'failed', error: 'disk write failed' }, | ||||||||||
| ); | ||||||||||
|
|
||||||||||
| const result = await workflowsCommand.action!(ctx, 'wf_badc0de'); | ||||||||||
| if (!result || result.type !== 'message') throw new Error('no result'); | ||||||||||
| expect(result.content).toContain('status : interrupted'); | ||||||||||
| expect(result.content).toContain('recovery : not recoverable'); | ||||||||||
| expect(result.content).toContain('reason : disk write failed'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| // The reason string is content another process wrote, not a runId, so | ||||||||||
| // the legacy-filename guard never sees it. Both the listing row (which | ||||||||||
| // slices to 80 chars, keeping any ESC) and the detail view render it. | ||||||||||
| it('strips terminal control bytes from a durable recovery reason', async () => { | ||||||||||
| getMock.mockReturnValue(undefined); | ||||||||||
| const ctx = await ctxWithManifest( | ||||||||||
| { runId: 'wf_badc0de', status: 'running' }, | ||||||||||
| { | ||||||||||
| integrity: 'failed', | ||||||||||
| error: '\u001b]0;pwned\u0007disk write failed', | ||||||||||
| }, | ||||||||||
| ); | ||||||||||
|
|
||||||||||
| const listing = await workflowsCommand.action!(ctx, ''); | ||||||||||
| const detail = await workflowsCommand.action!(ctx, 'wf_badc0de'); | ||||||||||
| if (!listing || listing.type !== 'message') throw new Error('no result'); | ||||||||||
| if (!detail || detail.type !== 'message') throw new Error('no result'); | ||||||||||
|
|
||||||||||
| expect(listing.content).not.toContain('\u001b'); | ||||||||||
| expect(detail.content).not.toContain('\u001b'); | ||||||||||
| expect(detail.content).toContain('disk write failed'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| // Same vector as the recovery-reason case, through the rest of the | ||||||||||
| // rendered family. `parseManifest` type-checks these as strings only, | ||||||||||
| // so a crafted manifest in the project's workflows dir carries ESC/OSC | ||||||||||
| // bytes all the way into both render paths. | ||||||||||
| it('strips terminal control bytes from every rendered manifest string', async () => { | ||||||||||
| getMock.mockReturnValue(undefined); | ||||||||||
| const esc = '\u001b]0;pwned\u0007'; | ||||||||||
| const ctx = await ctxWithManifest({ | ||||||||||
| runId: 'wf_badc0de', | ||||||||||
| status: 'failed', | ||||||||||
| meta: { | ||||||||||
| name: `${esc}evil-name`, | ||||||||||
| description: `${esc}evil-description`, | ||||||||||
| whenToUse: `${esc}evil-when`, | ||||||||||
| }, | ||||||||||
| error: `${esc}boom`, | ||||||||||
| recentLogs: [`${esc}log line`], | ||||||||||
| phases: [`${esc}phase one`], | ||||||||||
| perPhaseTokens: new Map([[`${esc}phase one`, 42]]), | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| const listing = await workflowsCommand.action!(ctx, ''); | ||||||||||
| const detail = await workflowsCommand.action!(ctx, 'wf_badc0de'); | ||||||||||
| if (!listing || listing.type !== 'message') throw new Error('no result'); | ||||||||||
| if (!detail || detail.type !== 'message') throw new Error('no result'); | ||||||||||
|
|
||||||||||
| expect(listing.content).not.toContain('\u001b'); | ||||||||||
| expect(detail.content).not.toContain('\u001b'); | ||||||||||
| // Sanitizing must not swallow the content itself. | ||||||||||
| expect(detail.content).toContain('evil-name'); | ||||||||||
| expect(detail.content).toContain('evil-description'); | ||||||||||
| expect(detail.content).toContain('evil-when'); | ||||||||||
| expect(detail.content).toContain('boom'); | ||||||||||
| expect(detail.content).toContain('log line'); | ||||||||||
| expect(detail.content).toContain('phase one'); | ||||||||||
| // The token chip still resolves, so the map keys went through the | ||||||||||
| // same sanitizing pass as the phase list. | ||||||||||
| expect(detail.content).toContain('42t'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| // A kept LF would let a crafted record forge extra rows and extra | ||||||||||
| // detail fields, which is why the single-line helper flattens it. | ||||||||||
| it('flattens newlines in single-line manifest fields', async () => { | ||||||||||
| getMock.mockReturnValue(undefined); | ||||||||||
| const ctx = await ctxWithManifest({ | ||||||||||
| runId: 'wf_badc0de', | ||||||||||
| status: 'failed', | ||||||||||
| error: 'boom\n status : completed', | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| const detail = await workflowsCommand.action!(ctx, 'wf_badc0de'); | ||||||||||
| if (!detail || detail.type !== 'message') throw new Error('no result'); | ||||||||||
|
|
||||||||||
| const lines = detail.content.split('\n'); | ||||||||||
| // The payload stays inside the error row instead of becoming a row | ||||||||||
| // of its own, so it cannot impersonate a real status field. | ||||||||||
| expect(lines.some((l) => l.startsWith(' error : boom'))).toBe( | ||||||||||
| true, | ||||||||||
| ); | ||||||||||
| expect(lines).not.toContain(' status : completed'); | ||||||||||
| expect(lines).toContain(' status : failed'); | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| it('never routes pause or resume to an interrupted persisted run', async () => { | ||||||||||
| getMock.mockReturnValue(undefined); | ||||||||||
| const ctx = await ctxWithManifest({ | ||||||||||
| runId: 'wf_deadbeef', | ||||||||||
| status: 'paused', | ||||||||||
| }); | ||||||||||
|
|
||||||||||
| const result = await workflowsCommand.action!(ctx, 'p wf_deadbeef'); | ||||||||||
| expect(result).toMatchObject({ | ||||||||||
| type: 'message', | ||||||||||
| messageType: 'error', | ||||||||||
| content: | ||||||||||
| 'Workflow wf_deadbeef is interrupted and has no live owner; it cannot be paused or resumed.', | ||||||||||
| }); | ||||||||||
| expect(pauseMock).not.toHaveBeenCalled(); | ||||||||||
| expect(resumeMock).not.toHaveBeenCalled(); | ||||||||||
| }); | ||||||||||
| }); | ||||||||||
| }); | ||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Deferred (test-only). The bad-record semantic the R6-4 harness comment promises; tracked with the CLI test follow-up round.
中文说明
推迟(仅测试)。R6-4 脚手架注释承诺的坏记录语义;随 CLI 测试后续轮次跟进。