Add reusable dev session files to CLI Kit

packages/cli-kit/package.json

@@ -147,6 +147,7 @@
     "node-fetch": "3.3.2",
     "open": "8.4.2",
     "pathe": "1.1.2",
+    "proper-lockfile": "4.1.2",
     "react": "19.2.4",
     "semver": "7.8.4",
     "react-dom": "19.2.4",
@@ -163,6 +164,7 @@
     "@types/fs-extra": "9.0.13",
     "@types/gradient-string": "^1.1.2",
     "@types/lodash": "4.17.24",
+    "@types/proper-lockfile": "4.1.4",
     "@types/react": "^19.0.0",
     "@types/react-dom": "^19.0.0",
     "@types/semver": "^7.5.2",

packages/cli-kit/src/private/node/dev-instance-output.ts

@@ -0,0 +1,176 @@
+import {ConcurrentOutputContext, getConcurrentOutputContext, useConcurrentOutputContext} from './output-context.js'
+import {unstyled} from '../../public/node/output.js'
+import {
+  defineJsonOutputSchema,
+  formatJsonOutputTimestamp,
+  jsonOutputTimestampSchema,
+} from '../../public/node/json-output-schema.js'
+import {zod} from '../../public/node/schema.js'
+import {createWriteStream} from 'fs'
+import {Writable, type WritableOptions} from 'stream'
+import {finished} from 'stream/promises'
+import {StringDecoder} from 'string_decoder'
+import {once} from 'events'
+
+export const devInstanceLogOutputSchema = defineJsonOutputSchema({
+  name: 'DevInstanceLogEntry',
+  schema: zod
+    .object({
+      timestamp: jsonOutputTimestampSchema,
+      source: zod.string().describe('What produced the line, for example a process prefix like web or extensions.'),
+      level: zod.enum(['debug', 'info', 'warning', 'error']).optional(),
+      message: zod.string(),
+    })
+    .strict(),
+})
+
+/** One line of a dev instance's JSON Lines log. */
+export type DevInstanceLogEntry = zod.infer<typeof devInstanceLogOutputSchema.schema>
+
+interface DevInstanceOutputOptions {
+  logFile: string
+  logOutput?: Writable
+  isActive: () => boolean
+  onError: (error: unknown) => void
+}
+
+export function createDevInstanceOutput({logFile, logOutput, isActive, onError}: DevInstanceOutputOptions) {
+  // Each run starts a fresh log.
+  const logStream = createWriteStream(logFile, {flags: 'w', mode: 0o600})
+  logStream.on('error', onError)
+  logOutput?.on('error', onError)
+  const opened = once(logStream, 'open')
+  const outputs = new Set<Writable>()
+
+  const writeLogEntry = async (entry: Omit<DevInstanceLogEntry, 'timestamp'> & {timestamp?: string}) => {
+    if (!isActive()) return
+    const record = devInstanceLogOutputSchema.validate({
+      ...entry,
+      timestamp: entry.timestamp ?? formatJsonOutputTimestamp(new Date()),
+    })
+    const line = `${JSON.stringify(record)}\n`
+    try {
+      await writeToStream(logStream, line)
+      if (logOutput) await writeToStream(logOutput, line)
+    } catch (error) {
+      onError(error)
+      throw error
+    }
+  }
+
+  return {
+    opened,
+    log: writeLogEntry,
+    tee: (
+      target: Writable,
+      {source, stringWritesAreMessages = false}: {source: string; stringWritesAreMessages?: boolean},
+    ) => {
+      const decoder = new StringDecoder('utf8')
+      let pending = ''
+      let currentSource = source
+      const flushLines = async (text: string) => {
+        const lines = text.split(/\r?\n/)
+        pending = lines.pop() ?? ''
+        await Promise.all(
+          lines.map(async (line) => {
+            const message = unstyled(line).trimEnd()
+            if (message.length > 0) await writeLogEntry({source: currentSource, message})
+          }),
+        )
+      }
+      const flush = () => flushLines(`${pending}${decoder.end()}\n`)
+      const output = new DevInstanceOutput({
+        decodeStrings: !stringWritesAreMessages,
+        write(chunk: Buffer | string, encoding, callback) {
+          const nextSource = getConcurrentOutputContext()?.outputPrefix ?? source
+          const completeMessage = stringWritesAreMessages && typeof chunk === 'string'
+          const previous = nextSource !== currentSource || completeMessage ? flush() : Promise.resolve()
+          currentSource = nextSource
+          Promise.all([
+            previous,
+            writeToStream(target, chunk, encoding),
+            flushLines(
+              pending +
+                decoder.write(typeof chunk === 'string' ? Buffer.from(chunk, encoding) : chunk) +
+                (completeMessage ? '\n' : ''),
+            ),
+          ])
+            .then(() => callback())
+            .catch(callback)
+        },
+        final(callback) {
+          flush()
+            .then(() => callback())
+            .catch(callback)
+        },
+      })
+      const onTargetError = (error: Error) => output.destroy(error)
+      target.on('error', onTargetError)
+      output.on('error', (error) => onError(error))
+      output.once('close', () => {
+        target.off('error', onTargetError)
+      })
+      outputs.add(output)
+      return output
+    },
+    flush: () => Promise.allSettled([...outputs].map((output) => endStream(output))),
+    close: async () => {
+      try {
+        await endStream(logStream)
+        if (logOutput && !logOutput.destroyed) await writeToStream(logOutput, '')
+      } finally {
+        logOutput?.off('error', onError)
+      }
+    },
+  }
+}
+
+// Writable processes queued chunks in an earlier write's context. Capture context when each chunk is submitted.
+class DevInstanceOutput extends Writable {
+  private readonly contexts: (ConcurrentOutputContext | undefined)[] = []
+
+  constructor(options: WritableOptions & {write: NonNullable<WritableOptions['write']>}) {
+    super({
+      ...options,
+      write: (chunk, encoding, callback) => {
+        useConcurrentOutputContext(this.contexts.shift() ?? {}, () =>
+          options.write.call(this, chunk, encoding, callback),
+        )
+      },
+    })
+  }
+
+  write(
+    chunk: string | Uint8Array,
+    encoding?: BufferEncoding | ((error?: Error | null) => void),
+    callback?: (error?: Error | null) => void,
+  ): boolean {
+    this.contexts.push(getConcurrentOutputContext())
+    return typeof encoding === 'string'
+      ? super.write(chunk, encoding, callback)
+      : super.write(chunk, encoding ?? callback)
+  }
+
+  end(
+    chunk?: string | Uint8Array | null | (() => void),
+    encoding?: BufferEncoding | (() => void),
+    callback?: () => void,
+  ): this {
+    if (chunk !== undefined && chunk !== null && typeof chunk !== 'function')
+      this.contexts.push(getConcurrentOutputContext())
+    if (typeof chunk === 'function') return super.end(chunk)
+    return typeof encoding === 'string' ? super.end(chunk, encoding, callback) : super.end(chunk, encoding ?? callback)
+  }
+}
+
+async function writeToStream(stream: Writable, chunk: Buffer | string, encoding: BufferEncoding = 'utf8') {
+  await new Promise<void>((resolve, reject) => {
+    stream.write(chunk, encoding, (error) => (error ? reject(error) : resolve()))
+  })
+}
+
+async function endStream(stream: Writable) {
+  const completed = finished(stream, {cleanup: true})
+  stream.end()
+  await completed
+}

packages/cli-kit/src/private/node/output-context.ts

@@ -0,0 +1,18 @@
+import {AsyncLocalStorage} from 'node:async_hooks'
+
+export interface ConcurrentOutputContext {
+  outputPrefix?: string
+  /** Controls ANSI stripping for terminal output. JSON diagnostics are always unstyled. */
+  stripAnsi?: boolean
+}
+
+const outputContextStore = new AsyncLocalStorage<ConcurrentOutputContext>()
+
+export function useConcurrentOutputContext<T>(context: ConcurrentOutputContext, callback: () => T): T {
+  return outputContextStore.run(context, callback)
+}
+
+/** Returns the source and styling context for the current output write. */
+export function getConcurrentOutputContext(): ConcurrentOutputContext | undefined {
+  return outputContextStore.getStore()
+}

packages/cli-kit/src/private/node/ui/components/ConcurrentOutput.tsx

@@ -1,3 +1,4 @@
+import {ConcurrentOutputContext, getConcurrentOutputContext, useConcurrentOutputContext} from '../../output-context.js'
 import {outputInfo, type OutputProcess} from '../../../../public/node/output.js'
 import {AbortSignal} from '../../../../public/node/abort.js'
 import {useComplete} from '../../ui.js'
@@ -7,7 +8,6 @@ import figures from 'figures'
 import stripAnsi from 'strip-ansi'
 
 import {Writable} from 'stream'
-import {AsyncLocalStorage} from 'node:async_hooks'
 import {StringDecoder} from 'node:string_decoder'
 
 export interface ConcurrentOutputProps {
@@ -45,18 +45,6 @@ function currentTime() {
   return `${hours}:${minutes}:${seconds}`
 }
 
-interface ConcurrentOutputContext {
-  outputPrefix?: string
-  /** Controls ANSI stripping for terminal output. JSON diagnostics are always unstyled. */
-  stripAnsi?: boolean
-}
-
-const outputContextStore = new AsyncLocalStorage<ConcurrentOutputContext>()
-
-function useConcurrentOutputContext<T>(context: ConcurrentOutputContext, callback: () => T): T {
-  return outputContextStore.run(context, callback)
-}
-
 /** Runs finite processes concurrently and routes their output through the shared diagnostic context. */
 export async function runConcurrentProcessesForJson({
   processes,
@@ -79,7 +67,7 @@ export async function runConcurrentProcessesForJson({
               next()
               return
             }
-            const currentPrefix = outputContextStore.getStore()?.outputPrefix ?? process.prefix
+            const currentPrefix = getConcurrentOutputContext()?.outputPrefix ?? process.prefix
             if (!hasPendingLine) prefix = currentPrefix
             const lines = (pending + decoder.write(chunk)).split(/\r?\n/)
             pending = lines.pop() ?? ''
@@ -197,7 +185,7 @@ const ConcurrentOutput: FunctionComponent<ConcurrentOutputProps> = ({
     (process: OutputProcess, prefixes: string[]) => {
       return new Writable({
         write(chunk, _encoding, next) {
-          const context = outputContextStore.getStore()
+          const context = getConcurrentOutputContext()
           const prefix = context?.outputPrefix ?? process.prefix
           const shouldStripAnsi = context?.stripAnsi ?? true
           const log = chunk.toString('utf8').replace(/(\n)$/, '')
@@ -291,4 +279,4 @@ const ConcurrentOutput: FunctionComponent<ConcurrentOutputProps> = ({
     </Static>
   )
 }
-export {ConcurrentOutput, ConcurrentOutputContext, useConcurrentOutputContext}
+export {ConcurrentOutput, ConcurrentOutputContext, getConcurrentOutputContext, useConcurrentOutputContext}