fix(ai-harness): store a rejected input with its rejection

.changeset/harness-atomic-input-rejection.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-harness': patch
+---
+
+Store a rejected input and its `harness.input.rejected` record in one log append. The rejected receipt resolves after that append is stored, and a crash no longer runs the input as pending work.

docs/config.json

@@ -1122,7 +1122,7 @@
           "label": "Send inputs safely",
           "to": "harness/inputs",
           "addedAt": "2026-10-13",
-          "updatedAt": "2026-10-22"
+          "updatedAt": "2026-10-30"
         },
         {
           "label": "Run a tool without a model",
@@ -1133,7 +1133,7 @@
           "label": "Durable sessions",
           "to": "harness/durable-sessions",
           "addedAt": "2026-10-13",
-          "updatedAt": "2026-10-22"
+          "updatedAt": "2026-10-30"
         },
         {
           "label": "Run background work on workers",

docs/harness/durable-sessions.md

@@ -78,7 +78,7 @@ A host that stops also stops renewing its ownership of the thread. When the owne
 4. The time limit passed: the input ends `failed`, with the error code `timeout`.
 5. Else the turn runs again as the next attempt, and clients get a `harness.operation.resumed` event.
 
-An input that was stored but never ran, runs now. `session.settled({ inputId })` resolves to how an input ended. See [Send inputs safely](./inputs).
+An input that was stored but never ran, runs now. A rejected input is not pending work. When the receipt resolves, the log has `harness.input.rejected`. Recovery does not run that input. `session.settled({ inputId })` resolves to how an input ended. See [Send inputs safely](./inputs).
 
 A background agent that you started with `runAgainAfterCrash: true` runs again on the next host. A background child that the model started with `background: true` runs again too, unless you set `durability: { resumeBackgroundAgents: false }`. See [Keep a background agent going after a crash](./subagents#keep-a-background-agent-going-after-a-crash). The children and groups that it started before the stop do not start again: they give their stored results. See [Children of a run that runs again](./subagents#children-of-a-run-that-runs-again). For any other background agent whose host stopped, the next host:
 

docs/harness/inputs.md

@@ -38,6 +38,7 @@ const receipt = await turn.receipt
 console.log(receipt.status) // 'accepted' or 'queued'
 ```
 
+- When the session rejects a stored input, the receipt resolves after the log has `harness.input.rejected`.
 - The same id and the same message: you get the first input's operation, and the model is not called again.
 - The same id and another message: the receipt has `status: 'rejected'` and `reason: 'conflict'`, and `await turn` rejects with `InputRejectedError`.
 - An input to a thread that was deleted: the receipt has `status: 'rejected'` and `reason: 'deleted'`, and `await turn` rejects with `InputRejectedError`. Nothing runs.

packages/ai-harness/src/session/inputs.ts

@@ -44,7 +44,8 @@ import type { QueuedTurn, WaitingSteer } from './types'
  * How `accept` took an input. `offered`: another host owns the log, and the
  * input waits for that host in the input lane of the log. `refused`: the
  * `acceptInput` hook of the host refused it, with this reason. Nothing is
- * stored.
+ * stored. `rejected`: the input and `harness.input.rejected` landed in one
+ * append, with this reason.
  */
 export type Admission =
   | 'new'
@@ -54,6 +55,7 @@ export type Admission =
   | 'offered'
   | 'deleted'
   | { refused: string }
+  | { rejected: string }
 
 /** What `accept` stores with an input besides the input itself. */
 interface AcceptOptions {
@@ -74,6 +76,12 @@ interface AcceptOptions {
   offer?: () => Promise<OfferedHarnessInput>
   /** More records for the append of a new input. */
   records?: Array<HarnessRecord>
+  /**
+   * Called once the input is new, before its append. A reason stores
+   * `harness.input.rejected` in that same append and returns
+   * `{ rejected: reason }`. Turn records are left out.
+   */
+  rejectIf?: (input: HarnessInput) => string | undefined
 }
 
 /** The offer that the next `accept` applies. See `applyOffer`. */
@@ -625,7 +633,8 @@ export abstract class SessionInputs<
    * tick is a duplicate too. `user` is who sent it. The access check of
    * a `checked` call runs first: a refused input is not stored, and the
    * error rejects this call. The `turn` record goes in the append of the
-   * input, so a crash keeps both or neither.
+   * input, so a crash keeps both or neither. A rejection from `rejectIf`
+   * is in that append too: the receipt waits until the log has it.
    *
    * When another host owns the log, the input goes to that host through the
    * input lane (`offered`). An input from the lane (see `applyOffer`)
@@ -655,7 +664,12 @@ export abstract class SessionInputs<
         offer,
       })
       // A retry with the input as its caller sent it is the same input.
-      if (admission === 'new' && accepted !== input) {
+      // A rejection stores `accepted`, so the caller's payload is screened
+      // the same way.
+      const wasStored =
+        admission === 'new' ||
+        (typeof admission === 'object' && 'rejected' in admission)
+      if (wasStored && accepted !== input) {
         this.inputRecords.screened.set(
           inputId,
           inputKey(input, storedUser(user).user),
@@ -712,11 +726,17 @@ export abstract class SessionInputs<
     // for the next owner.
     if (this.closing && WORK_OPS.has(input.op)) return 'closed'
     this.admitted.set(inputId, inputKey(input, sender.user))
+    // Before the append, so the decision and the record are one batch.
+    // A crash then leaves the input absent, or rejected, never pending.
+    const reason = options.rejectIf?.(input)
     try {
-      const records = [
-        ...(options.turn ? turnRecords(inputId, options.turn) : []),
-        ...(options.records ?? []),
-      ]
+      const records =
+        reason !== undefined
+          ? [{ type: 'harness.input.rejected' as const, inputId, reason }]
+          : [
+              ...(options.turn ? turnRecords(inputId, options.turn) : []),
+              ...(options.records ?? []),
+            ]
       await this.writer.append(
         [
           {
@@ -742,15 +762,11 @@ export abstract class SessionInputs<
     }
     // An input with no work to run (a cancel, a setting) ends once stored.
     if (!WORK_OPS.has(input.op)) this.inputRecords.end(inputId)
-    this.feed.publish(
-      'session',
-      // The feed scrubs the session's secrets from the message.
-      customEvent(HARNESS_EVENTS.inputAccepted, {
-        inputId,
-        op: input.op,
-        ...('message' in input ? { message: input.message } : {}),
-      }),
-    )
+    this.publishAccepted(inputId, input)
+    if (reason !== undefined) {
+      this.recordRejection(inputId, reason)
+      return { rejected: reason }
+    }
     return 'new'
   }
 
@@ -1174,11 +1190,28 @@ export abstract class SessionInputs<
     )
   }
 
-  /** Refuse an input. Returns its rejected receipt. */
+  /** Refuse an input that is already in the log. Returns its rejected receipt. */
   protected reject(inputId: string, reason: string) {
     void this.writer
       .append([{ type: 'harness.input.rejected', inputId, reason }])
       .catch(() => {})
+    return this.recordRejection(inputId, reason)
+  }
+
+  /** Tell listeners the input is in the log. The feed scrubs secrets. */
+  private publishAccepted(inputId: string, input: HarnessInput) {
+    this.feed.publish(
+      'session',
+      customEvent(HARNESS_EVENTS.inputAccepted, {
+        inputId,
+        op: input.op,
+        ...('message' in input ? { message: input.message } : {}),
+      }),
+    )
+  }
+
+  /** Keep the rejected receipt and publish it. The log write is the caller's. */
+  private recordRejection(inputId: string, reason: string): Receipt {
     const receipt: Receipt = { inputId, status: 'rejected', reason }
     this.inputRecords.reject(receipt)
     this.feed.publish(
@@ -1348,7 +1381,9 @@ export abstract class SessionInputs<
     admission: Exclude<Admission, 'new'>,
   ): Receipt {
     if (typeof admission === 'object') {
-      return { inputId, status: 'rejected', reason: admission.refused }
+      const reason =
+        'rejected' in admission ? admission.rejected : admission.refused
+      return { inputId, status: 'rejected', reason }
     }
     if (admission !== 'duplicate' && admission !== 'offered') {
       return { inputId, status: 'rejected', reason: admission }