fix(app): fence prompt admission against session revert

Author: deepagent-ai

packages/app/src/pages/session.tsx

@@ -53,6 +53,7 @@ import {
   sendFollowupDraft,
 } from "@/components/prompt-input/submit"
 import { createSessionComposerState, SessionComposerRegion } from "@/pages/session/composer"
+import { createFollowupSubmissionRegistry } from "@/pages/session/followup-submission"
 import {
   createForkAction,
   createOpenReviewFile,
@@ -1445,14 +1446,17 @@ export default function Page() {
     return followup.edit[id]
   })
 
+  const followupSubmissions = createFollowupSubmissionRegistry()
+
   const followupMutation = useMutation(() => ({
     mutationFn: async (input: { sessionID: string; id: string }) => {
       const item = (followup.items[input.sessionID] ?? []).find((entry) => entry.id === input.id)
       if (!item) return
 
       setFollowup("failed", input.sessionID, undefined)
 
-      const ok = await sendFollowupDraft({
+      const controller = new AbortController()
+      const promise = sendFollowupDraft({
         client: sdk.client,
         sync,
         serverSync,
@@ -1461,11 +1465,18 @@ export default function Page() {
         intentSource: "followup",
         optimisticBusy: item.sessionDirectory === sdk.directory,
         confirmPromptDraft,
-      }).catch((err) => {
-        setFollowup("failed", input.sessionID, input.id)
-        fail(err)
-        return false
+        promptPrepareSignal: controller.signal,
       })
+      followupSubmissions.register({ ...input, controller, promise })
+      const ok = await promise
+        .catch((err) => {
+          setFollowup("failed", input.sessionID, input.id)
+          fail(err)
+          return false
+        })
+        .finally(() => {
+          followupSubmissions.clear(input.sessionID, input.id)
+        })
       if (!ok) return
 
       setFollowup("items", input.sessionID, (items) => (items ?? []).filter((entry) => entry.id !== input.id))
@@ -1509,6 +1520,7 @@ export default function Page() {
   }
 
   const queueFollowup = (draft: FollowupDraft) => {
+    if (reverting()) return
     setFollowup("items", draft.sessionID, (items) => [
       ...(items ?? []),
       { id: Identifier.ascending("message"), ...draft },
@@ -1520,6 +1532,7 @@ export default function Page() {
   const followupDock = createMemo(() => queuedFollowups().map((item) => ({ id: item.id, text: followupText(item) })))
 
   const sendFollowup = (sessionID: string, id: string) => {
+    if (reverting()) return Promise.resolve()
     if (sync.session.get(sessionID)?.parentID) return Promise.resolve()
     const item = (followup.items[sessionID] ?? []).find((entry) => entry.id === id)
     if (!item) return Promise.resolve()
@@ -1531,7 +1544,7 @@ export default function Page() {
   const editFollowup = (id: string) => {
     const sessionID = params.id
     if (!sessionID) return
-    if (followupBusy(sessionID)) return
+    if (reverting() || followupBusy(sessionID)) return
 
     const item = queuedFollowups().find((entry) => entry.id === id)
     if (!item) return
@@ -1548,7 +1561,7 @@ export default function Page() {
   const deleteFollowup = (id: string) => {
     const sessionID = params.id
     if (!sessionID) return
-    if (followupBusy(sessionID)) return
+    if (reverting() || followupBusy(sessionID)) return
 
     setFollowup("items", sessionID, (items) => (items ?? []).filter((entry) => entry.id !== id))
     setFollowup("failed", sessionID, (value) => (value === id ? undefined : value))
@@ -1563,10 +1576,15 @@ export default function Page() {
   const halt = (sessionID: string) =>
     busy(sessionID) ? sdk.client.session.abort({ sessionID }).catch(() => {}) : Promise.resolve()
 
+  const cancelFollowup = async (sessionID: string) => {
+    await followupSubmissions.cancel(sessionID)
+  }
+
   const revertMutation = useMutation(() => ({
     mutationFn: async (input: { sessionID: string; messageID: string }) => {
       const value = draft(input.messageID)
-      await (promptInputControl?.cancelPending() ?? Promise.resolve())
+      await cancelFollowup(input.sessionID)
+        .then(() => promptInputControl?.cancelPending() ?? Promise.resolve())
         .then(() => halt(input.sessionID))
         .then(() => sdk.client.session.revert(input))
         .then((result) => {
@@ -1595,7 +1613,8 @@ export default function Page() {
               messageID: next.id,
             })
 
-      await (promptInputControl?.cancelPending() ?? Promise.resolve())
+      await cancelFollowup(sessionID)
+        .then(() => promptInputControl?.cancelPending() ?? Promise.resolve())
         .then(() => halt(sessionID))
         .then(request)
         .then((result) => {
@@ -1748,6 +1767,7 @@ export default function Page() {
               queue: queueEnabled,
               items: followupDock(),
               sending: sendingFollowup(),
+              disabled: reverting(),
               edit: editingFollowup(),
               onQueue: queueFollowup,
               onAbort: () => {

packages/app/src/pages/session/composer/session-composer-region.tsx

@@ -37,6 +37,7 @@ export function SessionComposerRegion(props: {
     queue: () => boolean
     items: { id: string; text: string }[]
     sending?: string
+    disabled?: boolean
     edit?: { id: string; prompt: FollowupDraft["prompt"]; context: FollowupDraft["context"] }
     onQueue: (draft: FollowupDraft) => void
     onAbort: () => void
@@ -280,6 +281,7 @@ export function SessionComposerRegion(props: {
                 <SessionFollowupDock
                   items={props.followup!.items}
                   sending={props.followup!.sending}
+                  disabled={props.followup!.disabled}
                   onSend={props.followup!.onSend}
                   onEdit={props.followup!.onEdit}
                   onDelete={props.followup!.onDelete}

packages/app/src/pages/session/composer/session-followup-dock.tsx

@@ -8,6 +8,7 @@ import { useLanguage } from "@/context/language"
 export function SessionFollowupDock(props: {
   items: { id: string; text: string }[]
   sending?: string
+  disabled?: boolean
   onSend: (id: string) => void
   onEdit: (id: string) => void
   onDelete: (id: string) => void
@@ -86,7 +87,7 @@ export function SessionFollowupDock(props: {
                   size="small"
                   variant="secondary"
                   class="shrink-0"
-                  disabled={!!props.sending}
+                  disabled={props.disabled || !!props.sending}
                   onClick={() => props.onSend(item.id)}
                 >
                   {language.t("session.followupDock.sendNow")}
@@ -95,7 +96,7 @@ export function SessionFollowupDock(props: {
                   size="small"
                   variant="ghost"
                   class="shrink-0"
-                  disabled={!!props.sending}
+                  disabled={props.disabled || !!props.sending}
                   onClick={() => props.onEdit(item.id)}
                 >
                   {language.t("session.followupDock.edit")}
@@ -104,7 +105,7 @@ export function SessionFollowupDock(props: {
                   icon="close"
                   size="small"
                   variant="ghost"
-                  disabled={!!props.sending}
+                  disabled={props.disabled || !!props.sending}
                   onClick={() => props.onDelete(item.id)}
                   title={language.t("session.followupDock.delete")}
                   aria-label={language.t("session.followupDock.delete")}

packages/app/src/pages/session/followup-submission.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, test } from "bun:test"
+import { createFollowupSubmissionRegistry } from "./followup-submission"
+
+const deferred = () => {
+  let resolve!: (value: boolean) => void
+  const promise = new Promise<boolean>((done) => {
+    resolve = done
+  })
+  return { promise, resolve }
+}
+
+describe("follow-up submission registry", () => {
+  test("cancel aborts and joins only the targeted session", async () => {
+    const registry = createFollowupSubmissionRegistry()
+    const first = deferred()
+    const second = deferred()
+    const firstController = new AbortController()
+    const secondController = new AbortController()
+    registry.register({ sessionID: "session-a", id: "a", controller: firstController, promise: first.promise })
+    registry.register({ sessionID: "session-b", id: "b", controller: secondController, promise: second.promise })
+
+    let joined = false
+    const cancel = registry.cancel("session-a").then(() => {
+      joined = true
+    })
+    await Promise.resolve()
+    expect(firstController.signal.aborted).toBe(true)
+    expect(secondController.signal.aborted).toBe(false)
+    expect(joined).toBe(false)
+    first.resolve(false)
+    await cancel
+    expect(joined).toBe(true)
+    second.resolve(true)
+  })
+
+  test("an old completion cannot clear a replacement submission", async () => {
+    const registry = createFollowupSubmissionRegistry()
+    const current = deferred()
+    const controller = new AbortController()
+    registry.register({
+      sessionID: "session-a",
+      id: "replacement",
+      controller,
+      promise: current.promise,
+    })
+    registry.clear("session-a", "old")
+
+    const cancel = registry.cancel("session-a")
+    expect(controller.signal.aborted).toBe(true)
+    current.resolve(false)
+    await cancel
+  })
+})

packages/app/src/pages/session/followup-submission.ts

@@ -0,0 +1,39 @@
+export type FollowupSubmission = {
+  readonly sessionID: string
+  readonly id: string
+  readonly controller: AbortController
+  readonly promise: Promise<boolean>
+}
+
+export function createFollowupSubmissionRegistry() {
+  // Keyed by sessionID → Map<id, entry> so multiple concurrent followups per session
+  // are all tracked and cancelled on revert (fixes single-entry overwrite gap).
+  const submissions = new Map<string, Map<string, Omit<FollowupSubmission, "sessionID">>>()
+
+  return {
+    register(input: FollowupSubmission) {
+      let slot = submissions.get(input.sessionID)
+      if (!slot) {
+        slot = new Map()
+        submissions.set(input.sessionID, slot)
+      }
+      slot.set(input.id, {
+        id: input.id,
+        controller: input.controller,
+        promise: input.promise,
+      })
+    },
+    clear(sessionID: string, id: string) {
+      const slot = submissions.get(sessionID)
+      if (!slot) return
+      slot.delete(id)
+      if (slot.size === 0) submissions.delete(sessionID)
+    },
+    async cancel(sessionID: string) {
+      const slot = submissions.get(sessionID)
+      if (!slot) return
+      for (const sub of slot.values()) sub.controller.abort()
+      await Promise.all([...slot.values()].map((sub) => sub.promise.catch(() => false)))
+    },
+  }
+}

packages/core/src/database/migration.gen.ts

@@ -0,0 +1,18 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260806060000_session_mutation_epoch",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run("ALTER TABLE session ADD COLUMN mutation_epoch INTEGER NOT NULL DEFAULT 0")
+      yield* tx.run("ALTER TABLE session_intent ADD COLUMN mutation_epoch INTEGER NOT NULL DEFAULT 0")
+      yield* tx.run("ALTER TABLE session_steer ADD COLUMN mutation_epoch INTEGER NOT NULL DEFAULT 0")
+      yield* tx.run("ALTER TABLE session_steer ADD COLUMN superseded_at INTEGER")
+      yield* tx.run(`
+        CREATE INDEX session_steer_session_epoch_pending_idx
+        ON session_steer (session_id, mutation_epoch, delivery, consumed_seq, superseded_at, seq)
+      `)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260806060000_session_mutation_epoch.ts

@@ -1,4 +1,13 @@
-import { sqliteTable, text, integer, index, primaryKey, real, uniqueIndex, type AnySQLiteColumn } from "drizzle-orm/sqlite-core"
+import {
+  sqliteTable,
+  text,
+  integer,
+  index,
+  primaryKey,
+  real,
+  uniqueIndex,
+  type AnySQLiteColumn,
+} from "drizzle-orm/sqlite-core"
 import { sql } from "drizzle-orm"
 import * as DatabasePath from "../database/path"
 import { ProjectTable } from "../project/sql"
@@ -46,6 +55,7 @@ export const SessionTable = sqliteTable(
     tokens_reasoning: integer().notNull().default(0),
     tokens_cache_read: integer().notNull().default(0),
     tokens_cache_write: integer().notNull().default(0),
+    mutation_epoch: integer().notNull().default(0),
     revert: text({ mode: "json" }).$type<{ messageID: MessageID; partID?: PartID; snapshot?: string; diff?: string }>(),
     permission: text({ mode: "json" }).$type<PermissionV1.Ruleset>(),
     agent: text(),
@@ -182,10 +192,9 @@ export const SessionInputTable = sqliteTable(
 // loop (SessionPrompt.runLoop), where it is persisted as an ordinary tail user message. This is a
 // PLAIN durable buffer (direct row writes, NOT event-sourced) — deliberately distinct from
 // SessionInputTable, which is projected only by the dormant experimentalEventSystem V2 runner and
-// feeds a different (V2) history store. Consume-once is enforced by `consumed_seq`: `drainSteer`
-// atomically stamps every pending row it returns in one transaction, so a second drain (or a
-// concurrent one) sees no pending rows. `seq` is a per-session monotonic admission order (autoincrement
-// PK) so a drain returns steers in the exact order the user sent them.
+// feeds a different (V2) history store. Chat materialization writes the V1 message and consume stamp in
+// one transaction. `mutation_epoch` and `superseded_at` fence every pending row against revert/rewrite.
+// `seq` is a per-session monotonic admission order so a drain preserves exact send order.
 export const SessionSteerTable = sqliteTable(
   "session_steer",
   {
@@ -201,7 +210,9 @@ export const SessionSteerTable = sqliteTable(
     correlation_id: text(),
     prompt: text({ mode: "json" }).notNull().$type<Prompt>(),
     delivery: text().$type<SessionInput.Delivery>().notNull(),
+    mutation_epoch: integer().notNull().default(0),
     consumed_seq: integer(),
+    superseded_at: integer(),
     time_created: integer()
       .notNull()
       .$default(() => Date.now()),
@@ -221,16 +232,15 @@ export const SessionIntentTable = sqliteTable(
       .notNull()
       .references(() => SessionTable.id, { onDelete: "cascade" }),
     source: text().$type<"composer" | "intelligence" | "followup" | "rewrite">().notNull(),
-    state: text()
-      .$type<"preparing" | "admitting" | "admitted" | "canceled" | "superseded" | "failed">()
-      .notNull(),
+    state: text().$type<"preparing" | "admitting" | "admitted" | "canceled" | "superseded" | "failed">().notNull(),
     selected_variant: text().$type<"original" | "rewritten">(),
     selected_payload_hash: text(),
     delivery: text().$type<"turn" | SessionInput.Delivery>(),
     admitted_message_id: text(),
     correlation_id: text(),
     owner_token: text(),
     lease_expires_at: integer(),
+    mutation_epoch: integer().notNull().default(0),
     version: integer().notNull().default(0),
     time_created: integer().notNull(),
     time_selected: integer(),
@@ -272,9 +282,7 @@ export const TaskRunTable = sqliteTable(
     child_session_id: text().$type<SessionSchema.ID>().notNull(),
     generation: integer().notNull(),
     delivery_mode: text().$type<"foreground" | "background">().notNull(),
-    phase: text()
-      .$type<"admission" | "research" | "finalize" | "settled" | "queue" | "provision">()
-      .notNull(),
+    phase: text().$type<"admission" | "research" | "finalize" | "settled" | "queue" | "provision">().notNull(),
     state: text()
       .$type<
         | "admitted"
@@ -388,7 +396,13 @@ export const TaskRunTable = sqliteTable(
       .where(sql`${table.state} IN ('admitted', 'provisioning', 'running', 'researching', 'finalizing')`),
     index("task_run_parent_state_idx").on(table.parent_session_id, table.state, table.time_updated),
     index("task_run_root_idx").on(table.root_run_id),
-    index("task_run_queue_idx").on(table.state, table.available_at, table.priority, table.time_created, table.generation),
+    index("task_run_queue_idx").on(
+      table.state,
+      table.available_at,
+      table.priority,
+      table.time_created,
+      table.generation,
+    ),
     index("task_run_goal_idx").on(table.goal_id, table.goal_tick_seq, table.goal_role, table.goal_ordinal),
   ],
 )