An unparseable body leaves spool_bytes reporting bytes it already deleted (#904)

hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js

@@ -52,7 +52,10 @@ const GAP_BLOCK_TYPES = new Set([
  *
  * A file that fails to parse is deleted immediately and counted: the
  * same session is recoverable from transcript backfill, and an
- * undeleted body is a raw prompt sitting on disk.
+ * undeleted body is a raw prompt sitting on disk. Its size is reported
+ * alongside the count, because a deletion the caller cannot size leaves
+ * the published `spool_bytes` gauge high until the next sweep restates
+ * it - reporting bytes for content that is already off the disk.
  * @ref LLP 0252#project-then-delete [implements]: an unprojectable body is
  *   deleted and counted, not retried forever
  *
@@ -64,6 +67,7 @@ const GAP_BLOCK_TYPES = new Set([
  *   consumedBytes: number,
  *   missing: number,
  *   unparseable: number,
+ *   unparseableBytes: number,
  *   refused: string[],
  * }>}
  */
@@ -75,6 +79,7 @@ export async function loadSpooledBodies(events, opts) {
   let consumedBytes = 0
   let missing = 0
   let unparseable = 0
+  let unparseableBytes = 0
   /** @type {string[]} */
   const refused = []
 
@@ -103,7 +108,26 @@ export async function loadSpooledBodies(events, opts) {
     const body = parseMaybeJson(raw.toString('utf8'))
     if (!isPlainObject(body)) {
       unparseable += 1
-      await fs.rm(file, { force: true }).catch(() => {})
+      try {
+        // `unlink`, not `fs.rm(..., { force: true })`: a forced remove RESOLVES
+        // for a path that is already gone, and the bytes below are only ours to
+        // report if this call is the one that took the file off the disk. Two
+        // reads of the same `body_ref` can be in flight at once (the handler is
+        // not serialized, so an exporter retry overlaps the original it is
+        // retrying), and a forced remove would let both subtract those bytes.
+        await fs.unlink(file)
+        // Sized only once the file is gone: one that is still there (EPERM, a
+        // read-only spool) is still spooled, so subtracting its bytes would
+        // publish a spool smaller than the one the next sweep finds.
+        // @ref LLP 0257#status-and-health [implements]: S16 - `spool_bytes` is
+        //   the spool's current size, so a deletion only the reader can see has
+        //   to be sized for the caller
+        unparseableBytes += raw.length
+      } catch {
+        // Already gone, or not removable at all: either way this batch has no
+        // byte movement to report, and the content is recoverable from
+        // transcript backfill.
+      }
       continue
     }
     bodies.set(ref, {
@@ -115,7 +139,7 @@ export async function loadSpooledBodies(events, opts) {
     consumedBytes += raw.length
   }
 
-  return { bodies, consumedFiles, consumedBytes, missing, unparseable, refused }
+  return { bodies, consumedFiles, consumedBytes, missing, unparseable, unparseableBytes, refused }
 }
 
 /**

hypaware-core/plugins-workspace/claude/src/telemetry/source.js

@@ -638,6 +638,15 @@ function makeReceiveHandler({ ctx, deps, state, usageByRequestId, sessionBodyFac
         const spooled = await loadSpooledBodies(events, { spoolDir })
         state.bodiesMissing += spooled.missing
         state.bodiesUnparseable += spooled.unparseable
+        // The read arm deletes an unparseable body on the spot, so the gauge
+        // has to come down here too: a file the last sweep sized otherwise
+        // stays published for up to a sweep interval after it was removed.
+        // Clamped and sweep-corrected like the projected arm below. The
+        // policy-drop arm still lacks the same subtraction, because the
+        // delete-unread path reports no bytes to subtract.
+        // @ref LLP 0257#status-and-health [implements]: S16 - `spool_bytes` is
+        //   the spool's current size, not its size at the last sweep
+        state.spoolBytes = Math.max(0, state.spoolBytes - spooled.unparseableBytes)
         span.setAttribute('body_count', spooled.bodies.size)
         if (spooled.unparseable > 0) {
           span.setAttribute('bodies_unparseable', spooled.unparseable)

test/plugins/claude-telemetry-unparseable-body.test.js

@@ -0,0 +1,225 @@
+// @ts-check
+
+/**
+ * The unparseable-body arm of the Claude telemetry listener, driven through
+ * its real transport: a listener on an ephemeral port, a fake gateway behind
+ * it, and OTLP/JSON over the wire.
+ *
+ * A body file that does not parse is deleted immediately (an undeleted body is
+ * a raw prompt sitting on disk), and the published `spool_bytes` gauge has to
+ * come down with it, exactly as it does for a body that projected.
+ *
+ * @ref LLP 0257#status-and-health [tests]: S16 - `spool_bytes` is the spool's
+ *   current size, so the reader's own deletion has to move it
+ */
+
+import test from 'node:test'
+import assert from 'node:assert/strict'
+import fsp from 'node:fs/promises'
+import os from 'node:os'
+import path from 'node:path'
+
+import { appendSessionContext } from '../../hypaware-core/plugins-workspace/claude/src/session_context.js'
+import { createStartClaudeTelemetrySource } from '../../hypaware-core/plugins-workspace/claude/src/telemetry/source.js'
+import { claudeBodySpoolDir } from '../../hypaware-core/plugins-workspace/claude/src/telemetry/spool.js'
+import { loadSpooledBodies } from '../../hypaware-core/plugins-workspace/claude/src/telemetry/bodies.js'
+
+const SESSION = 'd41f0b2e-7c8a-4f19-9f61-6a1c2f7d0e33'
+const REQUEST_ID = 'req_011Ce8sjpb8Uzvot2JMvFkKe'
+
+/** @param {Record<string, unknown>} attrs */
+function kvAttributes(attrs) {
+  return Object.entries(attrs).map(([key, value]) => {
+    if (typeof value === 'number') {
+      return Number.isInteger(value)
+        ? { key, value: { intValue: value } }
+        : { key, value: { doubleValue: value } }
+    }
+    return { key, value: { stringValue: String(value) } }
+  })
+}
+
+/**
+ * @param {string} name
+ * @param {Record<string, unknown>} attrs
+ * @param {string} timestamp
+ */
+function record(name, attrs, timestamp) {
+  return {
+    timeUnixNano: String(BigInt(Date.parse(timestamp)) * 1_000_000n),
+    body: { stringValue: `claude_code.${name}` },
+    attributes: kvAttributes({
+      'session.id': SESSION,
+      'event.name': name,
+      'event.timestamp': timestamp,
+      ...attrs,
+    }),
+  }
+}
+
+/** @param {Array<ReturnType<typeof record>>} records */
+function envelope(records) {
+  return {
+    resourceLogs: [
+      {
+        resource: { attributes: kvAttributes({ 'service.name': 'claude-code' }) },
+        scopeLogs: [{ scope: { name: 'com.anthropic.claude_code.events' }, logRecords: records }],
+      },
+    ],
+  }
+}
+
+/**
+ * Start a real listener on an ephemeral port with a fake gateway behind it.
+ *
+ * `seed` files are written BEFORE the start-time sweep, which is what primes
+ * the published `spool_bytes`: a file dropped in afterwards is invisible to
+ * the gauge until the next sweep restates it.
+ *
+ * @param {{ seed?: Array<{ name: string, content: string }> }} [opts]
+ */
+async function startListener(opts = {}) {
+  const hypHome = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-unparseable-'))
+  const spoolDir = claudeBodySpoolDir(hypHome)
+  await fsp.mkdir(spoolDir, { recursive: true })
+  for (const file of opts.seed ?? []) {
+    await fsp.writeFile(path.join(spoolDir, file.name), file.content, 'utf8')
+  }
+  const stateFile = path.join(hypHome, 'claude-sessions.json')
+  const noop = () => {}
+  const start = createStartClaudeTelemetrySource({
+    gateway: /** @type {any} */ ({
+      recordProjectedExchange: async () => ({ rowsWritten: 0, rowsSkipped: 0 }),
+    }),
+    clientName: 'claude',
+    stateFile,
+  })
+  const ctx = /** @type {any} */ ({
+    config: { telemetry: { listen_host: '127.0.0.1', listen_port: 0 } },
+    env: { HYP_HOME: hypHome },
+    log: { info: noop, warn: noop, error: noop, debug: noop },
+    storage: {
+      cacheTablePath: () => path.join(hypHome, 'cache', 'claude_telemetry_events'),
+      appendRows: async () => {},
+    },
+  })
+  // `status` and `stop` are optional on the kernel's StartedSource; this one
+  // publishes both, and this test reads them.
+  const source = /** @type {any} */ (await start(ctx))
+  const first = /** @type {any} */ ((await source.status()).details)
+  const port = /** @type {number} */ (first.listen_port)
+
+  return {
+    hypHome,
+    spoolDir,
+    stateFile,
+    /** @param {ReturnType<typeof envelope>} body */
+    async post(body) {
+      return fetch(`http://127.0.0.1:${port}/v1/logs`, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify(body),
+      })
+    },
+    async details() {
+      return /** @type {Record<string, unknown>} */ ((await source.status()).details ?? {})
+    },
+    async cleanup() {
+      await source.stop()
+      await fsp.rm(hypHome, { recursive: true, force: true })
+    },
+  }
+}
+
+// The reader deletes an unparseable body and counts it, but before this fix it
+// never told the caller how many bytes went with it, so `spool_bytes` stayed
+// at its pre-batch value until the next sweep restated it (up to a minute) and
+// `hyp status` reported bytes for a file that was already off the disk.
+test('an unparseable body brings the published spool_bytes down with it', async () => {
+  const content = 'not json at all'
+  const listener = await startListener({ seed: [{ name: 'broken.request.json', content }] })
+  try {
+    // A recorded cwd, so the usage-policy gate resolves rather than withholds:
+    // this has to reach the READ path, not the drop path.
+    await appendSessionContext(listener.stateFile, {
+      session_id: SESSION,
+      transcript_path: undefined,
+      git_branch: undefined,
+      cwd: listener.hypHome,
+      ts: '2026-08-17T19:30:00.000Z',
+    })
+    const body = path.join(listener.spoolDir, 'broken.request.json')
+    // The start-time sweep is what primes the gauge, so the file has to be on
+    // disk before it runs.
+    assert.equal(
+      (await listener.details()).spool_bytes,
+      content.length,
+      'the start sweep publishes what is already spooled'
+    )
+
+    const res = await listener.post(envelope([
+      record('api_request_body', { body_ref: body, request_id: REQUEST_ID }, '2026-08-17T19:31:00.000Z'),
+    ]))
+    assert.equal(res.status, 200)
+
+    await assert.rejects(fsp.stat(body), 'an unparseable body is deleted, not left on disk')
+    const details = await listener.details()
+    assert.equal(details.spool_bytes, 0, 'the gauge has to come down by what the read arm removed')
+  } finally {
+    await listener.cleanup()
+  }
+})
+
+// The reader half on its own: the byte total is what lets the call site
+// subtract, so it has to be reported even when nothing projects.
+test('loadSpooledBodies reports the bytes an unparseable body took with it', async () => {
+  const dir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-unparseable-unit-'))
+  try {
+    const content = 'not json at all'
+    const file = path.join(dir, 'broken.request.json')
+    await fsp.writeFile(file, content, 'utf8')
+    const events = [{
+      name: 'api_request_body',
+      timestamp: '2026-08-17T19:31:00.000Z',
+      attributes: { body_ref: file, request_id: REQUEST_ID },
+    }]
+    const loaded = await loadSpooledBodies(/** @type {any} */ (events), { spoolDir: dir })
+    assert.equal(loaded.unparseable, 1)
+    assert.equal(loaded.consumedBytes, 0, 'nothing projected, so nothing was consumed')
+    assert.equal(loaded.unparseableBytes, content.length)
+    await assert.rejects(fsp.stat(file))
+  } finally {
+    await fsp.rm(dir, { recursive: true, force: true })
+  }
+})
+
+// Two reads of the same `body_ref` overlapping in the handler: both are issued
+// before either resolves, so both find the file and both call it unparseable,
+// but only one of them can be the call that removed it. `fs.rm(..., { force:
+// true })` resolves for a path that is already gone, so it reported the bytes
+// twice and brought `spool_bytes` down by 2x one deletion.
+test('two overlapping reads of one unparseable body report its bytes once', async () => {
+  const dir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-unparseable-race-'))
+  try {
+    const content = 'not json at all'
+    const file = path.join(dir, 'broken.request.json')
+    await fsp.writeFile(file, content, 'utf8')
+    const events = [{
+      name: 'api_request_body',
+      timestamp: '2026-08-17T19:31:00.000Z',
+      attributes: { body_ref: file, request_id: REQUEST_ID },
+    }]
+    const both = await Promise.all([
+      loadSpooledBodies(/** @type {any} */ (events), { spoolDir: dir }),
+      loadSpooledBodies(/** @type {any} */ (events), { spoolDir: dir }),
+    ])
+    assert.equal(both[0].unparseable + both[1].unparseable, 2, 'both reads saw it')
+    assert.equal(
+      both[0].unparseableBytes + both[1].unparseableBytes,
+      content.length,
+      'one file left the disk, so its bytes are reported once'
+    )
+  } finally {
+    await fsp.rm(dir, { recursive: true, force: true })
+  }
+})