[ai-harness] Rejected receipt resolves before durable rejection; recovery can rerun the input

Read session/inputs.js around admit() and reject(), then follow the post-admission rejection paths in session/requests.js and recovery in session/harness-session.js. Use the reproduction described in the issue to confirm that a rejected receipt can precede its durable log record and that recovery can rerun the input. Done means the rejection outcome is durably recorded with admission before the receipt resolves, and recovery does not rerun rejected work.

[ai-harness] A post-admission rejection's receipt resolves before harness.input.rejected is written, in a second append that is not awaited and can be lost

Versions: @tanstack/[email protected], @tanstack/[email protected], @tanstack/[email protected], [email protected], Node.js 24.22.1
Area: session input admission (session/inputs.js, session/requests.js) and log recovery (session/harness-session.js)

Summary

When a session rejects a work input after admitting it, the log receives two separate appends:

1. admit() awaits an append of harness.input (with any turn records).
2. The admission decision then runs against in-memory state. A rejection calls reject(), which appends harness.input.rejected without awaiting it and with its error swallowed (.catch(() => {})), and returns the rejected receipt synchronously.

The reproduction below shows, deterministically, on 0.1.0:

• The receipt resolves before the rejection is written. A log read made right after the receipt shows harness.input and no outcome for that id. This happens even with the plain memoryLogStore() and nothing held (scenario A).
• The two records land in separate appends. Holding the second append open shows a reader seeing the input with no outcome for as long as the append is in flight (scenario B).
• A crash between the appends turns a rejected input into one that ran. If the host stops before the rejection lands, the next host's recovery re-runs the logged follow-up as a turn. It fails with InterruptResumeValidationError and settles failed. The caller was told rejected, but the log never records a rejection (scenario C).

LogStore.append is documented as atomic per batch ("the store writes the whole batch … or it changes nothing"). The gap comes from the harness splitting one admission outcome across two batches.

Where (0.1.0 dist)

• session/inputs.js admit(): await this.writer.append([{ type: "harness.input", … }, ...records], { consumes }), then return "new".
• session/requests.js startTurn (prompt / steer / followUp): after accept() resolves "new", if (!running && this.waitsForInterrupts()) { this.reject(inputId, "pending_interrupts"); … }. The same method also rejects after admission with conflict, busy and no_human_wait. runTool and reset reject with pending_interrupts the same way. resolveForRun does so with conflict, use_answer, mixed_runs, unknown_question, no_pending_interrupts and busy.
• session/inputs.js reject(inputId, reason):

  this.writer.append([{ type: "harness.input.rejected", inputId, reason }]).catch(() => {});
  const receipt = { inputId, status: "rejected", reason };
  // …
  return receipt;
  

• session/harness-session.js recoverInputs: a pending chat input with a message is re-run (recoverDecision → run → enqueueTurn).

Reproduction

The reproduction is self-contained. It uses only the published packages above, memoryLogStore, and the fake adapter from @tanstack/ai/testing. It needs no network access, model provider or credentials. A LogStore proxy holds the one append that carries harness.input.rejected until the script releases it. No timer orders any step. The only timer is a 20 s watchdog that fails the run if a step never finishes.

Create these three files in an empty directory, then run:

npm install          # or `npm ci` with the lockfile this produces
npm run typecheck    # tsc --noEmit against the installed type definitions
npm run repro        # node repro.ts (Node.js >= 22.18 runs .ts directly)

{
  "name": "tanstack-harness-rejection-durability-repro",
  "private": true,
  "type": "module",
  "engines": { "node": ">=22.18" },
  "scripts": {
    "typecheck": "tsc --noEmit",
    "repro": "node repro.ts"
  },
  "dependencies": {
    "@opentelemetry/api": "1.9.1",
    "@tanstack/ai": "1.0.0",
    "@tanstack/ai-harness": "0.1.0",
    "@tanstack/ai-persistence": "1.0.0",
    "zod": "4.4.3"
  },
  "devDependencies": {
    "@types/node": "24.10.1",
    "typescript": "5.9.3"
  }
}

{
  "compilerOptions": {
    "target": "ES2023",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true,
    "verbatimModuleSyntax": true,
    "erasableSyntaxOnly": true,
    "skipLibCheck": true,
    "types": ["node"]
  },
  "include": ["repro.ts"]
}

/**
 * Reproduction: @tanstack/ai-harness 0.1.0 returns a post-admission
 * rejection's receipt before `harness.input.rejected` is durable, and logs
 * the input and its rejection in two separate appends.
 *
 * Only published packages, an in-memory log store, and the fake text adapter
 * from `@tanstack/ai/testing`. No network, no model provider, no timers that
 * order anything: a LogStore wrapper holds the one append that carries
 * `harness.input.rejected` until the script releases it.
 *
 * Run: npm ci && npm run typecheck && npm run repro
 */
import assert from 'node:assert/strict'

import { toolDefinition } from '@tanstack/ai'
import {
  createHarnessHost,
  defineHarness,
  logIdOf,
  type HarnessPersistence,
} from '@tanstack/ai-harness'
import {
  memoryLogState,
  memoryLogStore,
  memoryPersistence,
  type LogStore,
} from '@tanstack/ai-persistence'
import { fakeText } from '@tanstack/ai/testing'
import { z } from 'zod'

const THREAD = 'thread-1'
const LOG_ID = logIdOf({ logId: THREAD })
const REPLY = 'reply-1'

/** Every step, in the order it happened. */
const timeline: string[] = []
const note = (step: string) => {
  timeline.push(step)
  console.log(`  [${timeline.length}] ${step}`)
}

/** A promise and the function that resolves it. */
function deferred() {
  let resolve!: () => void
  const promise = new Promise<void>((done) => {
    resolve = done
  })
  return { promise, resolve }
}

/**
 * Wraps a LogStore so the first append that carries `harness.input.rejected`
 * after `arm()` waits until `release()`. Every other call passes through.
 */
function holdingRejections(inner: LogStore) {
  const reached = deferred()
  const released = deferred()
  const landed = deferred()
  let armed = false
  const log = new Proxy(inner, {
    get(target, prop, receiver) {
      if (prop !== 'append') {
        const value: unknown = Reflect.get(target, prop, receiver)
        return typeof value === 'function' ? value.bind(target) : value
      }
      const append: LogStore['append'] = async (batch) => {
        const isRejection = batch.records.some(
          (record) => record.type === 'harness.input.rejected',
        )
        if (!(armed && isRejection)) return target.append(batch)
        armed = false
        note('store: append of harness.input.rejected started (held)')
        reached.resolve()
        await released.promise
        await target.append(batch)
        note('store: append of harness.input.rejected landed')
        landed.resolve()
      }
      return append
    },
  })
  return {
    log,
    arm: () => {
      armed = true
    },
    reached: reached.promise,
    release: released.resolve,
    landed: landed.promise,
  }
}

/** The record types the log holds for one input id, in log order. */
async function recordsFor(log: LogStore, inputId: string) {
  const entries = await log.read({ logId: LOG_ID })
  return entries
    .filter(({ record }) => record['inputId'] === inputId)
    .map(({ record }) =>
      record.type === 'harness.input.settled'
        ? `${record.type}(${String(record['outcome'])})`
        : record.type,
    )
}

/** Resolves once `inputId` has an outcome record in the log. */
function outcomeOf(log: LogStore, inputId: string) {
  return new Promise<string[]>((resolve) => {
    const check = async () => {
      const types = await recordsFor(log, inputId)
      if (
        types.some(
          (type) =>
            type === 'harness.input.rejected' ||
            type.startsWith('harness.input.settled'),
        )
      ) {
        stop()
        resolve(types)
      }
    }
    const stop = log.subscribe({
      logId: LOG_ID,
      listener: () => void check(),
    })
    void check()
  })
}

const askPerson = toolDefinition({
  name: 'askPerson',
  description: 'Asks the person a question and waits for the answer.',
  inputSchema: z.object({ question: z.string() }),
  outputSchema: z.object({ text: z.string() }),
})

/** A harness whose first turn ends on a client-tool interrupt. */
function asker() {
  const adapter = fakeText()
  adapter.setResponses([
    {
      toolCalls: [{ name: 'askPerson', input: { question: 'Which day?' } }],
    },
    { text: 'Thanks.' },
  ])
  return defineHarness({
    name: 'repro/asker',
    adapter,
    tools: [askPerson.client()],
  })
}

/** Persistence over `log`, with the other stores in memory. */
function persistenceWith(log: LogStore): HarnessPersistence {
  const { metadata, sessions } = memoryPersistence().stores
  return { stores: { log, metadata, sessions } }
}

/** A thread that waits on a client-tool interrupt, on a host over `log`. */
async function waitingThread(log: LogStore) {
  const harness = asker()
  const host = createHarnessHost({ persistence: persistenceWith(log) })
  const session = await host.open(harness, { threadId: THREAD })
  const first = session.send('Start.', { inputId: 'turn-1' })
  await first.receipt
  await outcomeOf(log, 'turn-1')
  const pending = session.snapshot().pendingWaits
  assert.equal(
    pending.length,
    1,
    'the first turn ends on one open wait (its interrupt)',
  )
  return { harness, host, session }
}

async function baseline() {
  console.log('\nA. Plain memory store, nothing held')
  const log = memoryLogStore()
  const { host, session } = await waitingThread(log)
  const operation = session.send('Tuesday.', {
    mode: 'followUp',
    inputId: REPLY,
  })
  const receipt = await operation.receipt
  note(`caller: receipt ${JSON.stringify(receipt)}`)
  const atReceipt = await recordsFor(log, REPLY)
  note(`caller: log read right after the receipt: ${atReceipt.join(', ')}`)
  const settled = await outcomeOf(log, REPLY)
  note(`caller: log once the rejection landed: ${settled.join(', ')}`)
  assert.deepEqual(receipt, {
    inputId: REPLY,
    status: 'rejected',
    reason: 'pending_interrupts',
  })
  // Even the in-memory store, with nothing held, has not written the
  // rejection when the receipt resolves: a read the caller makes next sees
  // the input admitted with no outcome.
  assert.deepEqual(atReceipt, ['harness.input'])
  // It lands later, as a second append.
  assert.deepEqual(settled, ['harness.input', 'harness.input.rejected'])
  await host.close()
}

async function heldRejection() {
  console.log('\nB. The rejection append is held: receipt before durability')
  timeline.length = 0
  const held = holdingRejections(memoryLogStore())
  const { host, session } = await waitingThread(held.log)

  held.arm()
  const operation = session.send('Tuesday.', {
    mode: 'followUp',
    inputId: REPLY,
  })
  const receipt = await operation.receipt
  note(`caller: receipt ${JSON.stringify(receipt)}`)
  await held.reached
  const inWindow = await recordsFor(held.log, REPLY)
  note(`reader: log read while the rejection is held: ${inWindow.join(', ')}`)

  held.release()
  await held.landed
  const after = await recordsFor(held.log, REPLY)
  note(`reader: log read after release: ${after.join(', ')}`)

  // 1. The caller was told `rejected`...
  assert.deepEqual(receipt, {
    inputId: REPLY,
    status: 'rejected',
    reason: 'pending_interrupts',
  })
  // 2. ...before the rejection was durable: the receipt came first.
  assert.ok(
    timeline.findIndex((step) => step.startsWith('caller: receipt')) <
      timeline.findIndex((step) => step.includes('landed')),
    'the receipt resolved before harness.input.rejected landed',
  )
  // 3. A reader in that window sees the input admitted, with no outcome.
  assert.deepEqual(inWindow, ['harness.input'])
  // 4. The rejection is a second, separate append.
  assert.deepEqual(after, ['harness.input', 'harness.input.rejected'])
  await host.close()
}

async function crashBetweenAppends() {
  console.log(
    '\nC. The host stops between the two appends; a second host opens the thread',
  )
  timeline.length = 0
  let skippedMs = 0
  const state = memoryLogState({ now: () => Date.now() + skippedMs })
  const held = holdingRejections(memoryLogStore({ state }))
  const first = await waitingThread(held.log)

  held.arm()
  const receipt = await first.session.send('Tuesday.', {
    mode: 'followUp',
    inputId: REPLY,
  }).receipt
  note(`caller (host 1): receipt ${JSON.stringify(receipt)}`)
  await held.reached
  // Host 1 stops here: its rejection append never lands, and its log
  // ownership lapses (the store clock moves past the ownership TTL).
  skippedMs += 60 * 60 * 1000
  note('host 1: stopped before its rejection append landed')

  const secondLog = memoryLogStore({ state })
  const second = createHarnessHost({ persistence: persistenceWith(secondLog) })
  await second.open(asker(), { threadId: THREAD })
  note('host 2: opened the thread (its recovery runs)')
  const outcome = await outcomeOf(secondLog, REPLY)
  note(
    `reader: log for ${REPLY} after host 2's recovery: ${outcome.join(', ')}`,
  )

  assert.equal(receipt.status, 'rejected')
  // The caller was told `rejected`, but the log never records that: host 2
  // re-ran the logged input as a turn, which settled `failed`.
  assert.ok(!outcome.includes('harness.input.rejected'))
  assert.ok(outcome.includes('harness.input.applied'))
  assert.ok(outcome.includes('harness.input.settled(failed)'))
  await second.close()
}

const watchdog = setTimeout(() => {
  console.error('FAIL: a step did not finish within 20 s')
  process.exit(2)
}, 20_000)
watchdog.unref()

await baseline()
await heldRejection()
await crashBetweenAppends()
console.log('\nAll assertions passed.')
// Host 1 of C still holds its never-released append; end the process.
process.exit(0)

Output on 0.1.0

The run exits 0, so every assertion holds. Scenario C also logs TanStack's own InterruptResumeValidationError ("Thread has pending interrupts; resume is required before accepting new input. Resume entries must resolve or cancel the complete interrupt batch.") while host 2 runs the recovered turn. The step lines were identical across five consecutive runs:

A. Plain memory store, nothing held
  [1] caller: receipt {"inputId":"reply-1","status":"rejected","reason":"pending_interrupts"}
  [2] caller: log read right after the receipt: harness.input
  [3] caller: log once the rejection landed: harness.input, harness.input.rejected
B. The rejection append is held: receipt before durability
  [1] caller: receipt {"inputId":"reply-1","status":"rejected","reason":"pending_interrupts"}
  [2] store: append of harness.input.rejected started (held)
  [3] reader: log read while the rejection is held: harness.input
  [4] store: append of harness.input.rejected landed
  [5] reader: log read after release: harness.input, harness.input.rejected
C. The host stops between the two appends; a second host opens the thread
  [1] caller (host 1): receipt {"inputId":"reply-1","status":"rejected","reason":"pending_interrupts"}
  [2] store: append of harness.input.rejected started (held)
  [3] host 1: stopped before its rejection append landed
  [4] host 2: opened the thread (its recovery runs)
  [5] reader: log for reply-1 after host 2's recovery: harness.input, harness.run.node, harness.input.applied, harness.input.settled(failed)
All assertions passed.

What each assertion shows

• A, receipt equals rejected / pending_interrupts: the follow-up was admitted and then rejected, because the first turn ended on a client-tool interrupt.
• A, atReceipt equals ['harness.input']: with the plain in-memory store and nothing held, a read issued right after the receipt resolves does not yet contain the rejection. The receipt runs ahead of durability with no help from the test.
• A, settled equals ['harness.input', 'harness.input.rejected']: the rejection does land, later, as its own record.
• B, the receipt step comes before the "landed" step: the caller held rejected while the rejection append had not completed. The append waits on the store, and the receipt does not wait on the append.
• B, inWindow equals ['harness.input']: for as long as that append is in flight, any reader sees the input admitted with no outcome, the same state as an input that will run.
• B, after equals ['harness.input', 'harness.input.rejected']: the outcome is a second append, not part of the admission's batch.
• C, receipt.status equals rejected, and the log has no harness.input.rejected: host 1 stopped between the two appends. Its rejection never became durable, even though its caller was told rejected.
• C, the log has harness.input.applied and harness.input.settled(failed): host 2's recovery treated the logged input as pending work and ran it. With the interrupt still open, the turn failed. An input its sender was told was rejected is recorded as one that was applied and failed.

Scenario C simulates the stop. Host 1 stays in the process, but its held append is never released. Moving the shared store clock one hour ahead lets its 30 s log ownership lapse, so host 2 can take the log.

Expected semantics

An input id should never be durably visible in a state that contradicts the receipt its caller got:

1. A rejected receipt implies a durable rejection. When a caller holds { status: 'rejected' } for an admitted input, every subsequent read of the log should include harness.input.rejected for that id.
2. No observable "admitted but undecided" state for an input that is rejected. A reader should never see harness.input without an outcome for an input that will be rejected. Equivalently, harness.input without an outcome should mean "admitted, and it will run (or join, or queue)".
3. Crash safety. A crash at any point must leave the input either absent, so the caller may resend the same id, or rejected. It must never be pending in a way that recovery re-runs.
4. A failed rejection write is not silent. If the rejection cannot be written, the input must not be left looking admitted. The write failure should surface the way other log failures do: fail the operation and treat the log as failed.

Proposed fix

Root fix: atomic admission and rejection (crash-safe). Make the admission decision before the input record is written, and write the outcome in the same append:

• accepted: [harness.input, ...turnRecords], as today;
• rejected after the checks: [harness.input, harness.input.rejected] in one batch, with the same consumes. The input stays in the log, so knownAdmission dedupe and duplicate/conflict replay keep working.

The decision inputs (activeTurn, queue, interrupted, the runId checks, the open waits for resolve) are in-memory state of the single session that owns the log. Evaluating them just before the append is equivalent to today's check just after it, provided the writer chain orders the append before any later turn-end write. Alternatively, keep the decision after the input record but stage that record and flush it together with either the turn records or the rejection. Either way, there is exactly one append per admission outcome. This also gives the crash guarantee (3) for free: the batch lands entirely or not at all.

Mitigation only: await the rejection append (not crash-safe). Make reject() async, await the append, and fail the operation (and the log) when the append fails, instead of .catch(() => {}). This closes the "receipt before durability" window (1) and the silent-failure case (4). It does not close the crash window (3): a process that dies after the harness.input append and before the rejection append still leaves a pending input that recovery re-runs. Other hosts can also still observe the intermediate state (2) while the second append is in flight. It is a smaller change and a useful step, but not a full fix.

Suggested tests

1. Held rejection append. Use the holdingRejections wrapper from the reproduction. Assert that a post-admission rejection's receipt does not resolve before the log contains harness.input.rejected; today it does (scenario B). With the atomic fix, also assert that no read ever returns harness.input for the id without its outcome.
2. Crash between appends. Use scenario C as a test. With the atomic fix, host 2 sees the input either absent or rejected, and recovery runs no turn for it.
3. Failed rejection append. Make the store reject the append that carries the rejection, twice so the writer's single retry fails too. Assert that the operation and the log fail visibly and that the input is not left pending. (Not executed in the reproduction.)
4. Each post-admission reason. Repeat test 1 for pending_interrupts from prompt, runTool and reset, and for the resolve rejections (no_pending_interrupts, busy, use_answer, mixed_runs, unknown_question, conflict). The reproduction exercises only followUp with pending_interrupts. The other paths call the same reject().
5. Dedupe unchanged. After a rejection, resending the same id with the same payload returns duplicate, which replays the rejection. Resending it with a different payload returns conflict. Check both locally and through the offer lane (applyOffer / refuseOffer).

Impact

A caller that reads the log to decide what happened to an input cannot tell "admitted, will run" from "admitted, about to be rejected". Neither can another host that shares the log. One pattern this breaks is "resend under a new id once the old one was rejected, do nothing while it is pending". A consumer that treats a logged input with no outcome as taken can acknowledge an input that is then rejected, and it is never resent. After a crash between the appends, the input instead runs during recovery and fails. This report makes no claim about how often this happens in production. The reproduction only shows that the orderings are reachable and deterministic on 0.1.0.