The unparseable-body arm takes its bytes off spool_bytes (#905 finding 3)

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

@@ -55,6 +55,14 @@ const GAP_BLOCK_TYPES = new Set([
  * @ref LLP 0252#project-then-delete [implements]: an unprojectable body is
  *   deleted and counted, not retried forever
  *
+ * `unparseableBytes` is what that immediate deletion took off the disk, and it
+ * is reported separately from `consumedBytes` because the caller subtracts the
+ * two at different moments: the consumed files are deleted only after the
+ * batch's writes land, while an unparseable one is already gone by the time
+ * this returns. Without it the published `spool_bytes` kept counting bytes
+ * that no longer existed until the next sweep restated the gauge a minute
+ * later.
+ *
  * @param {ClaudeTelemetryEvent[]} events
  * @param {{ spoolDir: string }} opts
  * @returns {Promise<{
@@ -63,6 +71,7 @@ const GAP_BLOCK_TYPES = new Set([
  *   consumedBytes: number,
  *   missing: number,
  *   unparseable: number,
+ *   unparseableBytes: number,
  *   refused: string[],
  * }>}
  */
@@ -74,6 +83,7 @@ export async function loadSpooledBodies(events, opts) {
   let consumedBytes = 0
   let missing = 0
   let unparseable = 0
+  let unparseableBytes = 0
   /** @type {string[]} */
   const refused = []
 
@@ -102,7 +112,11 @@ export async function loadSpooledBodies(events, opts) {
     const body = parseMaybeJson(raw.toString('utf8'))
     if (!isPlainObject(body)) {
       unparseable += 1
-      await fs.rm(file, { force: true }).catch(() => {})
+      // Only bytes that actually left the disk. An unlink that failed (a
+      // read-only spool, EPERM) leaves the file occupying the cap, and
+      // subtracting it here would under-report `spool_bytes` instead.
+      const removed = await fs.rm(file, { force: true }).then(() => true).catch(() => false)
+      if (removed) unparseableBytes += raw.length
       continue
     }
     bodies.set(ref, {
@@ -114,7 +128,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

@@ -636,12 +636,25 @@ function makeReceiveHandler({ ctx, deps, state, usageByRequestId, sessionBodyFac
         state.bodiesMissing += spooled.missing
         state.bodiesUnparseable += spooled.unparseable
         span.setAttribute('body_count', spooled.bodies.size)
+        // An unparseable body is deleted inside the read above, not by the
+        // post-write delete below, so its bytes have to come off the gauge
+        // here. Without this the only thing that ever corrected `spool_bytes`
+        // for those files was the next sweep, and `hyp status` read the gauge
+        // in between and reported bytes for content already off the disk.
+        // @ref LLP 0257#status-and-health [implements]: S16 - the published
+        //   number is the spool's CURRENT byte size, so every arm that removes
+        //   a file has to take its bytes off it
+        if (spooled.unparseableBytes > 0) {
+          state.spoolBytes = Math.max(0, state.spoolBytes - spooled.unparseableBytes)
+        }
         if (spooled.unparseable > 0) {
           span.setAttribute('bodies_unparseable', spooled.unparseable)
+          span.setAttribute('bodies_unparseable_bytes', spooled.unparseableBytes)
           ctx.log.warn('claude.telemetry.body_unparseable', {
             [Attr.PLUGIN]: PLUGIN_NAME,
             error_kind: 'body_unparseable',
             body_count: spooled.unparseable,
+            body_bytes: spooled.unparseableBytes,
           })
         }
         for (const ref of spooled.refused) {

test/plugins/claude-telemetry-spool-accounting.test.js

@@ -0,0 +1,236 @@
+// @ts-check
+
+/**
+ * What the Claude telemetry listener publishes as `spool_bytes` after a batch
+ * removed body files, driven through its real transport: a listener on an
+ * ephemeral port, a fake gateway behind it, and OTLP/JSON over the wire.
+ *
+ * The case here is a deferred review finding from PR #851 (issue #905): the
+ * unparseable-body arm deletes the file inside the read, so its bytes are gone
+ * from the disk but were never taken off the gauge, and `hyp status` reported
+ * them until the next sweep restated it a minute later.
+ *
+ * @ref LLP 0257#status-and-health [tests]: S16 - the details carry the spool's
+ *   CURRENT byte size, so every arm that removes a file has to correct 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 = '2b1f9c4e-7f3a-4a51-9a2c-0d6f8e5b1c33'
+const ASSISTANT_UUID = '9d2c7a10-5c4e-4a6b-8f0a-1e2d3c4b5a69'
+
+/** @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, and the next sweep is exactly
+ * what this test must not be allowed to do the correcting.
+ *
+ * @param {{ seed?: Array<{ name: string, content: string }> }} [opts]
+ */
+async function startListener(opts = {}) {
+  const hypHome = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-spool-bytes-'))
+  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: 1, 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 })
+    },
+  }
+}
+
+/** @param {string} dir */
+async function bytesOnDisk(dir) {
+  const names = await fsp.readdir(dir)
+  let total = 0
+  for (const name of names) {
+    const stat = await fsp.stat(path.join(dir, name))
+    if (stat.isFile()) total += stat.size
+  }
+  return total
+}
+
+// A truncated body file is the ordinary case here: Claude Code writes the
+// spool file itself, so a killed process leaves half a JSON object behind.
+const BROKEN = '{"model":"claude-haiku-4-5-20251001","messages":[{"role":"ass'
+const VALID = JSON.stringify({
+  model: 'claude-haiku-4-5-20251001',
+  content: [{ type: 'thinking', thinking: 'checking the tree', signature: 'sig' }],
+})
+
+test('an unparseable body takes its bytes off spool_bytes when it is deleted', async () => {
+  const listener = await startListener({
+    seed: [
+      { name: 'req-broken.json', content: BROKEN },
+      { name: 'res-valid.json', content: VALID },
+    ],
+  })
+  try {
+    // A recorded cwd, so the usage-policy gate resolves rather than withholds:
+    // a withheld batch never reaches the read that deletes the broken file.
+    await appendSessionContext(listener.stateFile, {
+      session_id: SESSION,
+      transcript_path: undefined,
+      git_branch: undefined,
+      cwd: listener.hypHome,
+      ts: '2026-08-17T19:30:00.000Z',
+    })
+
+    const seeded = await listener.details()
+    assert.equal(
+      seeded.spool_bytes,
+      BROKEN.length + VALID.length,
+      'the start-time sweep publishes what the seeded files occupy',
+    )
+
+    const posted = await listener.post(envelope([
+      record('api_request_body', {
+        body_ref: path.join(listener.spoolDir, 'req-broken.json'),
+      }, '2026-08-17T19:30:30.000Z'),
+      record('api_response_body', {
+        body_ref: path.join(listener.spoolDir, 'res-valid.json'),
+      }, '2026-08-17T19:30:31.000Z'),
+      record('assistant_response', {
+        response: 'This is a spike repo.',
+        'message.uuid': ASSISTANT_UUID,
+        model: 'claude-haiku-4-5-20251001',
+      }, '2026-08-17T19:30:31.009Z'),
+    ]))
+    assert.equal(posted.status, 200)
+
+    // Both files are gone: the valid one after its write landed, the broken
+    // one inside the read that failed to parse it.
+    assert.deepEqual(await fsp.readdir(listener.spoolDir), [])
+
+    const after = await listener.details()
+    assert.equal(
+      after.spool_bytes,
+      await bytesOnDisk(listener.spoolDir),
+      'spool_bytes has to match the disk without waiting for the next sweep',
+    )
+    assert.equal(after.spool_bytes, 0)
+    assert.equal(after.bodies_projected, 1)
+  } finally {
+    await listener.cleanup()
+  }
+})
+
+// The gauge is only correctable if the read reports what it removed, so the
+// reporting itself is pinned here rather than only through the listener.
+test('loadSpooledBodies reports the bytes its unparseable arm removed', async () => {
+  const spoolDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'hyp-claude-spool-load-'))
+  try {
+    const broken = path.join(spoolDir, 'broken.json')
+    const valid = path.join(spoolDir, 'valid.json')
+    await fsp.writeFile(broken, BROKEN, 'utf8')
+    await fsp.writeFile(valid, VALID, 'utf8')
+
+    const loaded = await loadSpooledBodies(/** @type {any} */ ([
+      { name: 'api_request_body', attributes: { body_ref: broken }, timestamp: undefined },
+      { name: 'api_response_body', attributes: { body_ref: valid }, timestamp: undefined },
+    ]), { spoolDir })
+
+    assert.equal(loaded.unparseable, 1)
+    assert.equal(loaded.unparseableBytes, BROKEN.length)
+    // The parseable one is still on disk: it is deleted only after the
+    // batch's writes land, and its bytes are `consumedBytes`, not these.
+    assert.equal(loaded.consumedBytes, VALID.length)
+    assert.deepEqual(await fsp.readdir(spoolDir), ['valid.json'])
+  } finally {
+    await fsp.rm(spoolDir, { recursive: true, force: true })
+  }
+})