fix(dsh): retain safe admission stages and probe causes

packages/dsh-loopx-plugin/src/cli.ts

@@ -1,4 +1,5 @@
 import { spawn } from 'node:child_process'
+import { statSync } from 'node:fs'
 
 // 1.2.4 includes the released Windows peer-file fix. Bootstrap and every
 // runtime consumer must use the same floor, including an existing global CLI.
@@ -21,12 +22,57 @@ export class LoopXCliError extends Error {
     message: string,
     readonly retryable: boolean,
     readonly exitCode?: number | undefined,
+    readonly diagnostics: LoopXCliDiagnostics = {},
   ) {
     super(message)
     this.name = 'LoopXCliError'
   }
 }
 
+export interface LoopXCliDiagnostics {
+  osCode?: string | undefined
+  candidate?: 'configured-command' | 'managed-launcher' | 'path-loopx' | 'python-module' | undefined
+  cwdExists?: boolean | undefined
+  causeKind?: LoopXCliErrorKind | undefined
+  probes?: readonly LoopXCliError[] | undefined
+}
+
+export function spawnCwdExists(cwd?: string): boolean | undefined {
+  try {
+    return statSync(cwd ?? process.cwd(), { throwIfNoEntry: false })?.isDirectory() ?? false
+  } catch {
+    return undefined
+  }
+}
+
+/** Only closed labels and scalars belong in the driver's diagnostic record. */
+function cliFailureFields(error: unknown): string {
+  const failure = error instanceof LoopXCliError ? error : undefined
+  const details = failure?.diagnostics
+  const kind = (value: unknown): string => [
+    'aborted', 'missing', 'timeout', 'transport', 'output_limit', 'exit',
+    'typed_failure', 'invalid_json', 'invalid_schema',
+  ].includes(value as string) ? String(value) : 'internal'
+  const code = [
+    'ENOENT', 'EPERM', 'EACCES', 'ETIMEDOUT', 'EAGAIN', 'EBUSY', 'EMFILE',
+    'ENFILE', 'ENOMEM', 'ENOTDIR', 'E2BIG', 'ENOEXEC', 'EINVAL', 'ECANCELED', 'EIO',
+  ].includes(details?.osCode ?? '') ? details!.osCode : 'none'
+  const candidate = [
+    'configured-command', 'managed-launcher', 'path-loopx', 'python-module',
+  ].includes(details?.candidate ?? '') ? details!.candidate : 'unknown'
+  const exitCode = Number.isSafeInteger(failure?.exitCode) ? failure!.exitCode : 'none'
+  const cwd = typeof details?.cwdExists === 'boolean' ? details.cwdExists : 'unknown'
+  return `kind=${kind(failure?.kind)} cause_kind=${kind(details?.causeKind ?? failure?.kind)} code=${code} exit_code=${exitCode} candidate=${candidate} cwd_exists=${cwd}`
+}
+
+export function cliErrorFields(error: unknown): string {
+  const probes = error instanceof LoopXCliError
+    ? error.diagnostics.probes ?? [] : []
+  const selected = probes.length > 8 ? [probes[0]!, ...probes.slice(-7)] : probes
+  const fields = selected.map(probe => cliFailureFields(probe).replaceAll(' ', ',')).join(';')
+  return `${cliFailureFields(error)} probes=${fields || 'none'} probes_omitted=${probes.length - selected.length}`
+}
+
 export interface FileResult {
   readonly exitCode: number
   readonly stdout: string
@@ -52,6 +98,7 @@ export interface LoopXCommand {
   readonly prefix: readonly string[]
   readonly skillCommand: string
   readonly version: string
+  readonly candidate?: LoopXCliDiagnostics['candidate']
 }
 
 export interface ManagedLoopXLauncher {
@@ -72,9 +119,9 @@ const FORCE_KILL_GRACE_MS = 250
 
 function transportKind(error: NodeJS.ErrnoException): LoopXCliError {
   if (error.code === 'ENOENT') {
-    return new LoopXCliError('missing', 'required executable is unavailable', false)
+    return new LoopXCliError('missing', 'required executable is unavailable', false, undefined, { osCode: error.code })
   }
-  return new LoopXCliError('transport', 'could not start the LoopX command', true)
+  return new LoopXCliError('transport', 'could not start the LoopX command', true, undefined, { osCode: error.code })
 }
 
 export const runFile: FileRunner = (file, args, options) => new Promise((resolve, reject) => {
@@ -266,6 +313,7 @@ export async function runJsonCommand(
   let lastError: LoopXCliError | undefined
 
   for (let attempt = 0; attempt < attempts; attempt += 1) {
+    const cwdExists = spawnCwdExists(options.cwd)
     try {
       const result = await runner(
         command.file,
@@ -296,6 +344,8 @@ export async function runJsonCommand(
       const failure = error instanceof LoopXCliError
         ? error
         : new LoopXCliError('transport', 'LoopX command failed to execute', true)
+      failure.diagnostics.cwdExists = cwdExists
+      failure.diagnostics.candidate ??= command.candidate
       lastError = failure
       if (!failure.retryable || attempt + 1 >= attempts) throw failure
       await wait(retryDelays[attempt] ?? retryDelays.at(-1) ?? 0, options.signal)
@@ -367,7 +417,12 @@ export async function resolveLoopXCommand(
         })),
       ]
 
+  const probes: LoopXCliError[] = []
   for (const candidate of candidates) {
+    const label = configured ? 'configured-command'
+      : managedCandidates.includes(candidate) ? 'managed-launcher'
+        : candidate.file === 'loopx' ? 'path-loopx' : 'python-module'
+    const cwdExists = spawnCwdExists()
     try {
       const result = await runner(
         candidate.file,
@@ -383,13 +438,25 @@ export async function resolveLoopXCommand(
         throw new LoopXCliError('aborted', 'operation was cancelled', false)
       }
       const version = result.exitCode === 0 ? supportedLoopXVersion(result.stdout) : undefined
-      if (version) return { ...candidate, version }
+      if (version) return { ...candidate, version, candidate: label }
+      throw new LoopXCliError(
+        result.exitCode === 0 ? 'invalid_schema' : 'exit',
+        'LoopX version probe did not verify', false, result.exitCode,
+      )
     } catch (error: unknown) {
       if (options.signal?.aborted) {
         throw new LoopXCliError('aborted', 'operation was cancelled', false)
       }
       if (error instanceof LoopXCliError && error.kind === 'aborted') throw error
+      const failure = error instanceof LoopXCliError ? error
+        : new LoopXCliError('transport', 'LoopX version probe failed to execute', false)
+      probes.push(new LoopXCliError(failure.kind, 'version probe failed', false, failure.exitCode, {
+        osCode: failure.diagnostics.osCode, candidate: label, cwdExists,
+      }))
     }
   }
-  throw new LoopXCliError('missing', `LoopX CLI ${MINIMUM_LOOPX_VERSION} or newer is unavailable`, false)
+  const last = probes.at(-1)
+  throw new LoopXCliError('missing', `LoopX CLI ${MINIMUM_LOOPX_VERSION} or newer is unavailable`, false, last?.exitCode, {
+    ...last?.diagnostics, causeKind: last?.kind, probes,
+  })
 }

packages/dsh-loopx-plugin/src/driver.ts

@@ -9,6 +9,7 @@ import type {} from '@deepseek-ai/dsh-session'
 import type { SessionEvent, UserMessage } from '@deepseek-ai/dsh-session'
 import {
   LoopXCliError,
+  cliErrorFields,
   resolveLoopXCommand,
   runJsonCommand,
   supportedLoopXVersion,
@@ -112,6 +113,18 @@ interface WaitPlan {
 }
 
 type EvaluationPlan = QueuePlan | WaitPlan
+type AdmissionStage = 'resolve-command' | 'resolve-binding' | 'quota' | 'heartbeat'
+  | 'pre-step-recheck' | 'maintenance-slot'
+
+function preStepRejection(
+  reason: string, messages: UserMessage[], submitted: UserMessage, startedAt: number,
+): string {
+  const kinds = [...new Set(messages.filter(message => message.id !== submitted.id).map(message => {
+    const kind = message.source?.kind
+    return typeof kind === 'string' && /^[a-z][a-z0-9-]{0,47}$/u.test(kind) ? kind : 'unknown'
+  }))].sort().slice(0, 16)
+  return `pre_step_rejected stage=pre-step-recheck reason=${reason} other_kinds=${kinds.join(',') || 'none'} elapsed_ms=${Math.round(performance.now() - startedAt)}`
+}
 
 export interface LoopXDriverOptions {
   readonly isLiveAgent: (agent: Agent) => boolean
@@ -632,7 +645,11 @@ export class LoopXContinuationDriver {
       ?? (() => `dsh-loopx-${randomUUID()}`)
     this.retryDelaysMs = options.retryDelaysMs ?? [200, 750]
     this.runDetached = options.runDetached ?? (operation => operation())
-    this.warn = options.warn ?? (() => {})
+    this.warn = message => {
+      try { options.warn?.(message) } catch {
+        // Diagnostic exporters must not interrupt admission or message restoration.
+      }
+    }
   }
 
   observeAgent(agent: Agent): void {
@@ -807,21 +824,25 @@ export class LoopXContinuationDriver {
       : messages.find(message => message.id === reservation.messageId)
         ?? messages.find(message => isDriverMessage(message))
     if (submitted === undefined) return next()
+    const startedAt = performance.now()
     if (reservation === undefined || !sameReservation(submitted, reservation)
       || reservation.phase !== 'claimed') {
+      this.warn(`dsh-loopx-driver: ${preStepRejection(reservation === undefined ? 'missing-reservation' : !sameReservation(submitted, reservation) ? 'reservation-mismatch' : 'reservation-not-claimed', messages, submitted, startedAt)}`)
       if (reservation !== undefined && submitted.id === reservation.messageId) {
         this.retireReservation(state)
       }
       this.restoreOtherMessages(agent, messages, submitted)
       return { kind: 'reject' }
     }
     if (messages.some(message => message.id !== submitted.id && !isHostContextMessage(message))) {
+      this.warn(`dsh-loopx-driver: ${preStepRejection('competing-messages', messages, submitted, startedAt)}`)
       this.retireReservation(state)
       this.restoreOtherMessages(agent, messages, submitted)
       return { kind: 'reject' }
     }
 
     if (!(await this.authorityStillAllows(state, reservation, signal))) {
+      this.warn(`dsh-loopx-driver: ${preStepRejection('authority-before-next', messages, submitted, startedAt)}`)
       this.retireReservation(state)
       this.restoreOtherMessages(agent, messages, submitted)
       this.requestEvaluation(state)
@@ -842,6 +863,7 @@ export class LoopXContinuationDriver {
       return decision
     }
     if (decision.kind === 'reject') {
+      this.warn(`dsh-loopx-driver: ${preStepRejection('downstream-reject', messages, submitted, startedAt)}`)
       state.pauseAfterTurnError = true
       this.retireReservation(state)
       return decision
@@ -850,11 +872,13 @@ export class LoopXContinuationDriver {
     if (decision.messages.filter(message => !isHostContextMessage(message)).length !== 1
       || admitted === undefined
       || !sameReservation(admitted, reservation)) {
+      this.warn(`dsh-loopx-driver: ${preStepRejection('downstream-batch', decision.messages, submitted, startedAt)}`)
       this.retireReservation(state)
       this.restoreOtherMessages(agent, decision.messages, submitted)
       return { kind: 'reject' }
     }
     if (!(await this.authorityStillAllows(state, reservation, signal))) {
+      this.warn(`dsh-loopx-driver: ${preStepRejection('authority-after-next', decision.messages, submitted, startedAt)}`)
       this.retireReservation(state)
       this.restoreOtherMessages(agent, decision.messages, submitted)
       return { kind: 'reject' }
@@ -982,6 +1006,9 @@ export class LoopXContinuationDriver {
 
   private async evaluate(state: DriverState): Promise<boolean> {
     if (!this.ready(state)) return false
+    const startedAt = performance.now()
+    let stage: AdmissionStage = 'maintenance-slot'
+    let admissionSignal: AbortSignal | undefined
     const controller = new AbortController()
     state.controller = controller
     let evaluationStarted = false
@@ -991,7 +1018,8 @@ export class LoopXContinuationDriver {
       plan = await state.agent.runMaintenance(async maintenanceSignal => {
         evaluationStarted = true
         const signal = AbortSignal.any([controller.signal, maintenanceSignal])
-        return this.buildPlan(state, signal)
+        admissionSignal = signal
+        return this.buildPlan(state, signal, value => { stage = value })
       })
     } catch (error: unknown) {
       if (!evaluationStarted && !controller.signal.aborted
@@ -1016,7 +1044,7 @@ export class LoopXContinuationDriver {
         this.command = undefined
       }
       if (!controller.signal.aborted && !cancelled) {
-        this.warn(`dsh-loopx-driver: LoopX admission failed: ${renderThrown(error)}`)
+        this.warn(`dsh-loopx-driver: admission_failed stage=${stage} ${cliErrorFields(error)} signal_aborted=${admissionSignal?.aborted ?? controller.signal.aborted} elapsed_ms=${Math.round(performance.now() - startedAt)}`)
       }
       return evaluationStarted
     } finally {
@@ -1067,11 +1095,14 @@ export class LoopXContinuationDriver {
   private async buildPlan(
     state: DriverState,
     signal: AbortSignal,
+    setStage: (stage: AdmissionStage) => void,
   ): Promise<EvaluationPlan> {
     const session = state.agent.session
     this.ensureReady(state, signal, session)
+    setStage('resolve-command')
     const command = await this.loopXCommand(signal)
     this.ensureReady(state, signal, session)
+    setStage('resolve-binding')
     const bindingPayload = await this.resolveBinding(command, state.agent, signal)
     this.ensureReady(state, signal, session)
     if (bindingPayload.ok !== true) {
@@ -1087,6 +1118,7 @@ export class LoopXContinuationDriver {
     }
 
     const turnInstanceId = this.makeTurnInstanceId()
+    setStage('quota')
     const quota = await this.readQuota(command, state.agent, binding, turnInstanceId, { signal })
     this.ensureReady(state, signal, session)
     if (quota.ok !== true) {
@@ -1107,6 +1139,7 @@ export class LoopXContinuationDriver {
     if (quota.should_run !== true) {
       return schedulerWait(quota, state.schedulerToken, state.unchangedPolls)
     }
+    setStage('heartbeat')
     const heartbeat = await this.readHeartbeat(
       command,
       state.agent,
@@ -1145,6 +1178,7 @@ export class LoopXContinuationDriver {
     signal: AbortSignal,
   ): Promise<boolean> {
     if (!this.reservationIsCurrent(state, reservation, signal)) return false
+    const startedAt = performance.now()
     try {
       const command = await this.loopXCommand(signal)
       const bindingPayload = await this.resolveBinding(command, state.agent, signal)
@@ -1170,7 +1204,7 @@ export class LoopXContinuationDriver {
         this.command = undefined
       }
       if (!(error instanceof LoopXCliError && error.kind === 'aborted')) {
-        this.warn(`dsh-loopx-driver: pre-step authority check failed: ${renderThrown(error)}`)
+        this.warn(`dsh-loopx-driver: admission_failed stage=pre-step-recheck ${cliErrorFields(error)} elapsed_ms=${Math.round(performance.now() - startedAt)}`)
       }
       return false
     }

packages/dsh-loopx-plugin/tests/admission-closeout.integration.spec.ts

@@ -554,7 +554,10 @@ describe('DSH admission-to-closeout integration', () => {
       expect(reachedNext).toBe(true)
       expect(admission).toEqual({ kind: 'reject' })
       expect(guards).toBe(3)
-      expect(warnings).toEqual([])
+      expect(warnings).toHaveLength(1)
+      expect(warnings[0]).toContain(
+        'pre_step_rejected stage=pre-step-recheck reason=authority-after-next other_kinds=none',
+      )
       expect(testFixture.queued).toHaveLength(0)
     } finally {
       await driver.dispose()
@@ -624,9 +627,11 @@ describe('DSH admission-to-closeout integration', () => {
         if (scenario === 'digest mismatch' || scenario === 'inline digest mismatch') {
           expect(reads).toBe(1)
           expect(testFixture.queued).toHaveLength(0)
-          expect(warnings).toEqual([expect.stringContaining(
-            scenario === 'digest mismatch' ? 'invalid page' : 'selection digest did not match',
-          )])
+          expect(warnings).toHaveLength(1)
+          expect(warnings[0]).toContain(
+            'admission_failed stage=quota kind=invalid_schema cause_kind=invalid_schema',
+          )
+          expect(warnings[0]).not.toMatch(/invalid page|selection digest did not match/u)
         } else if (scenario === 'held') {
           expect(reads).toBe(2)
           expect(testFixture.queued).toHaveLength(0)