Add JSON streaming to app function replay watch mode

.changeset/function-replay-json-stream.md

@@ -0,0 +1,6 @@
+---
+'@shopify/cli': minor
+'@shopify/cli-kit': patch
+---
+
+Stream structured execution and rebuild records from app function replay --json in watch mode.

packages/app/src/cli/commands/app/function/replay.test.ts

@@ -5,6 +5,9 @@ import {testAppLinked, testFunctionExtension} from '../../../models/app/app.test
 import {renderReplay} from '../../../services/function/ui.js'
 import {selectFunctionRunPrompt} from '../../../prompts/function/replay.js'
 import {functionRunJsonOutputSchema} from '../../../services/function/runner/types.js'
+import {functionReplayJsonOutputSchema} from '../../../services/function/replay/types.js'
+import {watchReplay} from '../../../services/function/replay/watch.js'
+import {AppEventWatcher} from '../../../services/dev/app-events/app-event-watcher.js'
 import {Config} from '@oclif/core'
 import {afterEach, expect, test, vi} from 'vitest'
 import {captureOutputWithExitCode, exec, terminalSupportsPrompting} from '@shopify/cli-kit/node/system'
@@ -17,6 +20,7 @@ import {isTerminalInteractive} from '@shopify/cli-kit/node/context/local'
 vi.mock('../../../services/app-context.js')
 vi.mock('@shopify/cli-kit/node/system')
 vi.mock('../../../services/function/ui.js')
+vi.mock('../../../services/function/replay/watch.js')
 vi.mock('../../../prompts/function/replay.js')
 vi.mock('../../../services/function/common.js', async (importOriginal) => ({
   ...(await importOriginal<typeof import('../../../services/function/common.js')>()),
@@ -96,14 +100,15 @@ test('preserves native numeric tokens in finite replay output', async () => {
 })
 
 test.each([{argv: ['--json', '--log', 'abcdef']}, {argv: ['--json', '--watch', '--log', 'abcdef']}])(
-  'rejects JSON watch mode before linking the app: $argv',
+  'streams JSON watch mode: $argv',
   async ({argv}) => {
-    const instance = new FunctionReplay(argv, await Config.load())
-    await withCapturedStandardStreams(async ({stdout}) => {
-      await expect(instance.run()).rejects.toThrow('Use --no-watch with --json')
-      expect(stdout()).toBe('')
+    await inTemporaryDirectory(async (directory) => {
+      const instance = await command(argv, directory)
+      await instance.run()
+      expect(watchReplay).toHaveBeenCalledOnce()
+      expect(renderReplay).not.toHaveBeenCalled()
+      expect(captureOutputWithExitCode).not.toHaveBeenCalled()
     })
-    expect(linkedAppContext).not.toHaveBeenCalled()
   },
 )
 
@@ -116,6 +121,51 @@ test('keeps the default text watch workflow', async () => {
   })
 })
 
+test.each([
+  {signal: 'SIGINT' as const, exitCode: 130},
+  {signal: 'SIGTERM' as const, exitCode: 143},
+  {signal: 'SIGQUIT' as const, exitCode: 131},
+])(
+  'writes watch records and cleans up on $signal through the command and real output writer',
+  async ({signal, exitCode}) => {
+    await inTemporaryDirectory(async (directory) => {
+      const {watchReplay: realWatchReplay} = await vi.importActual<
+        typeof import('../../../services/function/replay/watch.js')
+      >('../../../services/function/replay/watch.js')
+      vi.mocked(watchReplay).mockImplementation(realWatchReplay)
+      vi.spyOn(AppEventWatcher.prototype, 'start').mockResolvedValue(undefined)
+      const processEvents: NodeJS.EventEmitter = process
+      vi.mocked(captureOutputWithExitCode).mockImplementation(async () => {
+        setImmediate(() => processEvents.emit('shopify:signal', signal))
+        return {stdout: JSON.stringify(nativeResult), stderr: 'Runner warning', exitCode: 0}
+      })
+      const previousExitCode = process.exitCode
+      const instance = await command(['--json', '--no-input', '--log', 'abcdef'], directory)
+      try {
+        await withCapturedStandardStreams(async ({stdout, stderr}) => {
+          await runWithCommandEventsForCommand(instance.argv, () => instance.run())
+          expect(
+            stdout()
+              .trim()
+              .split('\n')
+              .map((line) => JSON.parse(line)),
+          ).toEqual([
+            expect.objectContaining({type: 'status', status: 'watching'}),
+            expect.objectContaining({type: 'execution', result: nativeResult, exitCode: 0}),
+            expect.objectContaining({type: 'status', status: 'cancelled'}),
+          ])
+          expect(JSON.parse(stderr())).toMatchObject({type: 'diagnostic', message: 'Runner warning'})
+          expect(process.exitCode).toBe(exitCode)
+        })
+      } finally {
+        // The signal handler runs inside this test; restore the worker's exit status.
+        // eslint-disable-next-line require-atomic-updates
+        process.exitCode = previousExitCode
+      }
+    })
+  },
+)
+
 test('captures finite text replay and preserves native output', async () => {
   await inTemporaryDirectory(async (directory) => {
     const nativeText = '  Function output:\n{ operations: [] }\n'
@@ -174,8 +224,11 @@ test('requires --log when --no-input is enabled', async () => {
   expect(selectFunctionRunPrompt).not.toHaveBeenCalled()
 })
 
-test('shares the native runner schema and preserves synchronous analytics', () => {
-  expect(FunctionReplay.jsonOutputSchema).toBe(functionRunJsonOutputSchema)
+test('discovers finite and streaming schemas and preserves synchronous analytics', () => {
+  expect(FunctionReplay.jsonOutputSchema).toBe(functionReplayJsonOutputSchema)
+  expect(functionReplayJsonOutputSchema.validate(nativeResult)).toEqual(
+    functionRunJsonOutputSchema.validate(nativeResult),
+  )
   expect(FunctionReplay.descriptionForHelp()).toContain('FunctionRunResult')
   expect(FunctionReplay.requiresSyncAnalytics).toBe(true)
 })

packages/app/src/cli/commands/app/function/replay.ts

@@ -1,26 +1,27 @@
 import {chooseFunction, functionFlags} from '../../../services/function/common.js'
 import {replay} from '../../../services/function/replay.js'
-import {functionRunJsonOutputSchema} from '../../../services/function/runner/types.js'
+import {functionReplayJsonOutputSchema} from '../../../services/function/replay/types.js'
 import {presentFunctionExecution} from '../../../services/function/runner/result.js'
 import {appFlags} from '../../../flags.js'
 import AppLinkedCommand, {AppLinkedCommandOutput} from '../../../utilities/app-linked-command.js'
 import {linkedAppContext} from '../../../services/app-context.js'
 import {globalFlags, jsonFlag, requiredIfNonInteractive} from '@shopify/cli-kit/node/cli'
 import {Flags} from '@oclif/core'
-import {AbortError} from '@shopify/cli-kit/node/error'
 
 export default class FunctionReplay extends AppLinkedCommand {
   public static get requiresSyncAnalytics(): boolean {
     return true
   }
 
-  static get jsonOutputSchema() {
-    return functionRunJsonOutputSchema
-  }
+  static jsonOutputSchema = functionReplayJsonOutputSchema
 
   static summary = 'Replays a function run from an app log.'
 
-  static descriptionWithMarkdown = `Runs the function from your current directory for [testing purposes](https://shopify.dev/docs/apps/functions/testing-and-debugging). To learn how you can monitor and debug functions when errors occur, refer to [Shopify Functions error handling](https://shopify.dev/docs/api/functions/errors).`
+  static descriptionWithMarkdown = `Runs the function from your current directory for [testing purposes](https://shopify.dev/docs/apps/functions/testing-and-debugging). To learn how you can monitor and debug functions when errors occur, refer to [Shopify Functions error handling](https://shopify.dev/docs/api/functions/errors).
+
+Watch mode is enabled by default. With --json, writes one JSONL record per lifecycle update, rebuild, and execution. Execution records preserve the native Function runner result under result. Diagnostics and prompts use stderr. Recoverable build and Function execution failures keep watching; fatal CLI failures end the stream with the standard error document. Ctrl-C stops watching and exits with code 130. Use --no-input for unattended replay.
+
+With --no-watch --json, writes one native Function runner result and exits, preserving the finite result contract.`
 
   static description = this.descriptionForHelp()
 
@@ -49,9 +50,6 @@ export default class FunctionReplay extends AppLinkedCommand {
 
   public async run(): Promise<AppLinkedCommandOutput> {
     const {flags} = await this.parse(FunctionReplay)
-    if (flags.json && flags.watch) {
-      throw new AbortError('JSON output requires a finite replay. Use --no-watch with --json.')
-    }
 
     const {app} = await linkedAppContext({
       directory: flags.path,

packages/app/src/cli/services/build/extension-cancellation.test.ts

@@ -0,0 +1,168 @@
+import {buildFunctionExtension} from './extension.js'
+import {testAppLinked, testFunctionExtension} from '../../models/app/app.test-data.js'
+import {AbortController} from '@shopify/cli-kit/node/abort'
+import {AbortSilentError} from '@shopify/cli-kit/node/error'
+import {fileExists, inTemporaryDirectory, writeFile} from '@shopify/cli-kit/node/fs'
+import {joinPath} from '@shopify/cli-kit/node/path'
+import {expect, test, vi} from 'vitest'
+import lockfile from 'proper-lockfile'
+import {Writable} from 'stream'
+
+vi.mock('../function/schema-version.js')
+
+test('cancels a build waiting for a real lock without starting its command', async () => {
+  await inTemporaryDirectory(async (directory) => {
+    const extension = await testFunctionExtension({dir: directory})
+    extension.configuration.build!.command = 'node build.cjs'
+    await writeFile(joinPath(directory, 'build.cjs'), "require('node:fs').writeFileSync('started', 'yes')")
+    const lockfilePath = joinPath(directory, '.build-lock')
+    const release = await lockfile.lock(directory, {lockfilePath})
+    const lock = vi.spyOn(lockfile, 'lock')
+    const controller = new AbortController()
+    let settled = false
+    const building = buildFunctionExtension(extension, {
+      app: testAppLinked({directory}),
+      stdout: process.stderr,
+      stderr: process.stderr,
+      environment: 'development',
+      signal: controller.signal,
+    })
+    const result = building
+      .catch((error: unknown) => error)
+      .finally(() => {
+        settled = true
+      })
+
+    try {
+      await vi.waitFor(() => expect(lock).toHaveBeenCalled())
+      await expect(lock.mock.results[0]!.value).rejects.toMatchObject({code: 'ELOCKED'})
+      controller.abort()
+      await vi.waitFor(() => expect(settled).toBe(true))
+      await expect(result).resolves.toBeInstanceOf(AbortSilentError)
+      await expect(lockfile.check(directory, {lockfilePath})).resolves.toBe(true)
+      await expect(fileExists(joinPath(directory, 'started'))).resolves.toBe(false)
+    } finally {
+      controller.abort()
+      await release()
+      await result
+      lock.mockRestore()
+    }
+  })
+})
+
+test('builds after the current lock holder releases the lock', async () => {
+  await inTemporaryDirectory(async (directory) => {
+    const extension = await testFunctionExtension({dir: directory})
+    extension.configuration.build!.command = 'node build.cjs'
+    await writeFile(joinPath(directory, 'build.cjs'), "require('node:fs').writeFileSync('started', 'yes')")
+    const lockfilePath = joinPath(directory, '.build-lock')
+    const release = await lockfile.lock(directory, {lockfilePath})
+    const lock = vi.spyOn(lockfile, 'lock')
+    const controller = new AbortController()
+    const building = buildFunctionExtension(extension, {
+      app: testAppLinked({directory}),
+      stdout: process.stderr,
+      stderr: process.stderr,
+      environment: 'development',
+      signal: controller.signal,
+    })
+    const result = building.catch((error: unknown) => error)
+
+    try {
+      try {
+        await vi.waitFor(() => expect(lock).toHaveBeenCalled())
+        await expect(lock.mock.results[0]!.value).rejects.toMatchObject({code: 'ELOCKED'})
+      } finally {
+        await release()
+      }
+
+      await expect(result).resolves.toMatchObject({status: 'success'})
+      await expect(fileExists(joinPath(directory, 'started'))).resolves.toBe(true)
+      await expect(lockfile.check(directory, {lockfilePath})).resolves.toBe(false)
+    } finally {
+      controller.abort()
+      await result
+      lock.mockRestore()
+    }
+  })
+})
+
+test('releases a newly acquired lock if cancellation arrives before the build starts', async () => {
+  await inTemporaryDirectory(async (directory) => {
+    const extension = await testFunctionExtension({dir: directory})
+    extension.configuration.build!.command = 'node build.cjs'
+    await writeFile(joinPath(directory, 'build.cjs'), "require('node:fs').writeFileSync('started', 'yes')")
+    const controller = new AbortController()
+    const acquireLock = lockfile.lock.bind(lockfile)
+    vi.spyOn(lockfile, 'lock').mockImplementationOnce(async (file, options) => {
+      const release = await acquireLock(file, options)
+      controller.abort()
+      return release
+    })
+
+    await expect(
+      buildFunctionExtension(extension, {
+        app: testAppLinked({directory}),
+        stdout: process.stderr,
+        stderr: process.stderr,
+        environment: 'development',
+        signal: controller.signal,
+      }),
+    ).rejects.toBeInstanceOf(AbortSilentError)
+
+    await expect(fileExists(joinPath(directory, 'started'))).resolves.toBe(false)
+    await expect(lockfile.check(directory, {lockfilePath: joinPath(directory, '.build-lock')})).resolves.toBe(false)
+  })
+})
+
+test('silently cancels a running build and releases its lock', async () => {
+  await inTemporaryDirectory(async (directory) => {
+    const extension = await testFunctionExtension({dir: directory})
+    extension.configuration.build!.command = 'node build.cjs'
+    await writeFile(joinPath(directory, 'build.cjs'), "process.stdout.write('ready'); setInterval(() => {}, 1000)")
+    const controller = new AbortController()
+    const stdout = new Writable({
+      write(chunk, _encoding, callback) {
+        if (chunk.toString().includes('ready')) controller.abort()
+        callback()
+      },
+    })
+
+    try {
+      await expect(
+        buildFunctionExtension(extension, {
+          app: testAppLinked({directory}),
+          stdout,
+          stderr: process.stderr,
+          environment: 'development',
+          signal: controller.signal,
+        }),
+      ).rejects.toBeInstanceOf(AbortSilentError)
+      expect(controller.signal.aborted).toBe(true)
+      await expect(lockfile.check(directory, {lockfilePath: joinPath(directory, '.build-lock')})).resolves.toBe(false)
+    } finally {
+      controller.abort()
+      stdout.end()
+    }
+  })
+})
+
+test('silently skips a build whose signal is already aborted', async () => {
+  await inTemporaryDirectory(async (directory) => {
+    const extension = await testFunctionExtension({dir: directory})
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(
+      buildFunctionExtension(extension, {
+        app: testAppLinked({directory}),
+        stdout: process.stderr,
+        stderr: process.stderr,
+        environment: 'development',
+        signal: controller.signal,
+      }),
+    ).rejects.toBeInstanceOf(AbortSilentError)
+
+    await expect(fileExists(joinPath(directory, '.build-lock'))).resolves.toBe(false)
+  })
+})

packages/app/src/cli/services/build/extension.test.ts

@@ -7,6 +7,7 @@ import {describe, expect, test, vi} from 'vitest'
 import {exec} from '@shopify/cli-kit/node/system'
 import lockfile from 'proper-lockfile'
 import {AbortError} from '@shopify/cli-kit/node/error'
+import {AbortController} from '@shopify/cli-kit/node/abort'
 import * as fs from '@shopify/cli-kit/node/fs'
 import {dirname, joinPath} from '@shopify/cli-kit/node/path'
 import {Writable} from 'stream'
@@ -37,7 +38,7 @@ describe('buildFunctionExtension', () => {
     const releaseLock = vi.fn()
     const stdout = {write: vi.fn()} as unknown as Writable
     const stderr = {write: vi.fn()} as unknown as Writable
-    const signal = vi.fn() as any
+    const signal = new AbortController().signal
     const app = {} as any
     const extension = await testFunctionExtension({
       config,