diff --git a/packages/app/src/app.tsx b/packages/app/src/app.tsx
index 7acbad26..0da7afc6 100644
--- a/packages/app/src/app.tsx
+++ b/packages/app/src/app.tsx
@@ -45,7 +45,7 @@ import { PromptProvider } from "@/context/prompt"
 import { ServerConnection, ServerProvider, serverName, useServer } from "@/context/server"
 import { SettingsProvider } from "@/context/settings"
 import { TerminalProvider } from "@/context/terminal"
-import { TabsProvider, useTabs } from "@/context/tabs"
+import { startupTab, TabsProvider, useTabs } from "@/context/tabs"
 import { WslServersProvider } from "@/wsl/context"
 import DirectoryLayout from "@/pages/directory-layout"
 import Layout from "@/pages/layout"
@@ -144,20 +144,22 @@ function SessionProviders(props: ParentProps) {
 
 function RouterRoot(props: ParentProps<{ appChildren?: JSX.Element }>) {
   const tabs = useTabs()
+  const server = useServer()
   const navigate = useNavigate()
   const location = useLocation()
 
-  // On startup: as soon as persisted tabs are loaded (local disk, no HTTP calls),
-  // navigate to the last visited project's session list. This matches opencode's approach —
-  // use local/cached data to drive the first frame, let the server data fill in asynchronously.
-  // We navigate to /:dir/session (no specific session ID) so autoselecting can handle
-  // the final session selection independently without conflicting navigation.
+  // Restore the explicitly persisted active tab. A cross-server directory is never
+  // navigated until the target server has become active.
   createEffect(() => {
     if (!tabs.ready()) return
     if (location.pathname !== "/") return
-    const first = tabs.store[0]
-    if (!first) return
-    navigate(`/${first.dirBase64}/session`, { replace: true })
+    const tab = startupTab(tabs.store, tabs.active.key, server.list)
+    if (!tab) return
+    if (server.key !== tab.server) {
+      server.setActive(tab.server)
+      return
+    }
+    navigate(`/${tab.dirBase64}/session`, { replace: true })
   })
 
   return (
diff --git a/packages/app/src/components/terminal.tsx b/packages/app/src/components/terminal.tsx
index 8897c050..ef79fd36 100644
--- a/packages/app/src/components/terminal.tsx
+++ b/packages/app/src/components/terminal.tsx
@@ -35,6 +35,10 @@ export interface TerminalProps extends ComponentProps<"div"> {
   runtimeId?: string
   onSubmit?: () => void
   onStatusChange?: (status: TerminalStatus, error?: TerminalFailure) => void
+  /** When true (restored PTY), show the terminal immediately after xterm initialises
+   *  instead of waiting for the WebSocket handshake. The WebSocket still connects in
+   *  the background; input typed before it's ready is buffered. */
+  optimisticReady?: boolean
 }
 
 let shared: Promise<{ mod: typeof import("ghostty-web"); ghostty: Ghostty }> | undefined
@@ -159,6 +163,7 @@ export const Terminal = (props: TerminalProps) => {
     "runtimeId",
     "onSubmit",
     "onStatusChange",
+    "optimisticReady",
   ])
   const id = local.pty.ptyId
   let ws: WebSocket | undefined
@@ -435,6 +440,8 @@ export const Terminal = (props: TerminalProps) => {
       })
       cleanups.push(() => disposeIfDisposable(onResize))
       const onData = t.onData((data) => {
+        // When optimisticReady is active the buffering handler below takes over.
+        if (local.optimisticReady) return
         if (ws?.readyState === WebSocket.OPEN) ws.send(data)
       })
       cleanups.push(() => disposeIfDisposable(onData))
@@ -456,6 +463,28 @@ export const Terminal = (props: TerminalProps) => {
       scheduleSize(t.cols, t.rows)
       startResize()
 
+      // For restored PTYs: show the terminal surface immediately after xterm is
+      // initialised instead of waiting for the full WebSocket handshake (which can
+      // take 2-4 s). Input typed before the socket opens is buffered and flushed
+      // once the connection is established.
+      let inputBuffer = local.optimisticReady ? "" : undefined
+      if (local.optimisticReady) {
+        markReady()
+        // Intercept onData to buffer keystrokes until the WebSocket is open.
+        const onDataOpt = t.onData((data) => {
+          if (ws?.readyState === WebSocket.OPEN) {
+            if (inputBuffer) {
+              ws.send(inputBuffer)
+              inputBuffer = undefined
+            }
+            ws.send(data)
+          } else {
+            inputBuffer = (inputBuffer ?? "") + data
+          }
+        })
+        cleanups.push(() => disposeIfDisposable(onDataOpt))
+      }
+
       const once = { value: false }
       const decoder = new TextDecoder()
 
diff --git a/packages/app/src/context/layout.tsx b/packages/app/src/context/layout.tsx
index fca1b7b4..9c296052 100644
--- a/packages/app/src/context/layout.tsx
+++ b/packages/app/src/context/layout.tsx
@@ -938,6 +938,11 @@ export const { use: useLayout, provider: LayoutProvider } = createSimpleContext(
           const session = key()
           const current = store.sessionView[session]
           const bottom = next.bottomPanel
+          // Keep the project-level terminal flag in sync so that switching to a
+          // session that has never explicitly opened the panel doesn't close it.
+          // `bottomPanel` memo falls back to `store.terminal?.opened` for new
+          // sessions — this makes that fallback reflect the current user intent.
+          setTerminalOpened(bottom?.opened === true && bottom?.activeView === "terminal")
           if (!current) {
             setStore("sessionView", session, { scroll: {}, bottomPanel: bottom, rightPanelMode: next.rightPanelMode })
             return
diff --git a/packages/app/src/context/tabs.test.ts b/packages/app/src/context/tabs.test.ts
new file mode 100644
index 00000000..de8e7de2
--- /dev/null
+++ b/packages/app/src/context/tabs.test.ts
@@ -0,0 +1,38 @@
+import { describe, expect, test } from "bun:test"
+import { ServerConnection } from "./server"
+import { startupTab, tabKey, type Tab } from "./tabs"
+
+const primary = ServerConnection.Key.make("http://primary")
+const secondary = ServerConnection.Key.make("http://secondary")
+
+const servers = [
+  { type: "http" as const, http: { url: "http://primary" } },
+  { type: "http" as const, http: { url: "http://secondary" } },
+]
+
+const first: Tab = {
+  type: "session",
+  server: primary,
+  dirBase64: "L3ByaW1hcnk=",
+  sessionId: "first",
+}
+const lastActive: Tab = {
+  type: "session",
+  server: secondary,
+  dirBase64: "L3NlY29uZGFyeQ==",
+  sessionId: "last-active",
+}
+
+describe("startup tab recovery", () => {
+  test("restores the persisted active tab rather than the first tab", () => {
+    expect(startupTab([first, lastActive], tabKey(lastActive), servers)).toBe(lastActive)
+  })
+
+  test("falls back to the first stored tab when the active key is stale", () => {
+    expect(startupTab([first, lastActive], "stale", servers)).toBe(first)
+  })
+
+  test("does not restore a target whose server is unavailable", () => {
+    expect(startupTab([first, lastActive], tabKey(lastActive), [servers[0]])).toBeUndefined()
+  })
+})
diff --git a/packages/app/src/context/tabs.tsx b/packages/app/src/context/tabs.tsx
index 517d4057..64ac3e46 100644
--- a/packages/app/src/context/tabs.tsx
+++ b/packages/app/src/context/tabs.tsx
@@ -20,6 +20,16 @@ export type Tab = SessionTab
 export const tabHref = (tab: Tab) => `/${tab.dirBase64}/session/${tab.sessionId}`
 export const tabKey = (tab: Tab) => `${tab.server}\n${tabHref(tab)}`
 
+export function activeTab(tabs: Tab[], key?: string) {
+  return tabs.find((tab) => tabKey(tab) === key) ?? tabs[0]
+}
+
+export function startupTab(tabs: Tab[], activeKey: string | undefined, servers: ServerConnection.Any[]) {
+  const tab = activeTab(tabs, activeKey)
+  if (!tab) return
+  return servers.some((server) => ServerConnection.key(server) === tab.server) ? tab : undefined
+}
+
 export function sessionHasOpenTab(tabs: Tab[], server: ServerConnection.Key, session: Session) {
   const dirBase64 = base64Encode(session.directory)
   return tabs.some(
@@ -47,6 +57,10 @@ export const { use: useTabs, provider: TabsProvider } = createSimpleContext({
       },
       createStore<Tab[]>([]),
     )
+    const [active, setActive, _activePersist, activeReady] = persisted(
+      Persist.global("tabs.active"),
+      createStore({ key: undefined as string | undefined }),
+    )
 
     const params = useParams()
     const navigate = useNavigate()
@@ -61,6 +75,7 @@ export const { use: useTabs, provider: TabsProvider } = createSimpleContext({
     })
 
     const navigateTab = (tab: Tab) => {
+      setActive("key", tabKey(tab))
       const href = tabHref(tab)
       if (tab.server === server.key) {
         navigate(href)
@@ -96,13 +111,20 @@ export const { use: useTabs, provider: TabsProvider } = createSimpleContext({
             }),
           )
           if (nextTab) navigateTab(nextTab)
-          else navigate("/")
+          else {
+            setActive("key", undefined)
+            navigate("/")
+          }
         }).finally(() => closing.delete(key))
       },
       removeServer(key: ServerConnection.Key) {
         setStore((tabs) => tabs.filter((tab) => tab.server !== key))
+        if (active.key?.startsWith(`${key}\n`)) setActive("key", undefined)
         if (server.key === key) navigate("/")
       },
+      setActive(tab: Tab) {
+        navigateTab(tab)
+      },
       removeSessions: (input: SessionTabsRemovedDetail) => {
         void startTransition(() => {
           setStore(
@@ -145,6 +167,6 @@ export const { use: useTabs, provider: TabsProvider } = createSimpleContext({
       },
     }
 
-    return { ...actions, store, ready }
+    return { ...actions, active, ready: () => ready() && activeReady(), store }
   },
 })
diff --git a/packages/app/src/context/terminal.tsx b/packages/app/src/context/terminal.tsx
index 65c76500..eecaf9a0 100644
--- a/packages/app/src/context/terminal.tsx
+++ b/packages/app/src/context/terminal.tsx
@@ -52,6 +52,8 @@ export type LocalPTY = {
   titleNumber: number
   status: TerminalStatus
   error?: TerminalFailure
+  /** True when restored from cross-project navigation cache; cleared on first ready. */
+  restored?: boolean
 }
 
 export type TerminalStore = {
@@ -313,6 +315,14 @@ function removeTerminalPersistence(
 type TerminalSession = ReturnType<typeof createWorkspaceTerminalSession>
 const sessions = new Set<{ dir: string; scope: ServerScopeValue; value: TerminalSession }>()
 
+/** Per-directory PTY snapshot preserved across project switches so PTYs survive navigation. */
+interface TerminalPtySnapshot {
+  ptys: Array<Pick<LocalPTY, "id" | "ptyId" | "title" | "titleNumber">>
+  root: PaneNode
+  focusedPaneId: string
+}
+const directoryTerminalCache = new Map<string, { bottom: TerminalPtySnapshot | null; side: TerminalPtySnapshot | null }>()
+
 export function clearWorkspaceTerminals(
   dir: string,
   sessionIDs?: string[],
@@ -535,6 +545,37 @@ function createWorkspaceTerminalSession(
     resetRuntime() {
       reset()
     },
+    snapshot(): TerminalPtySnapshot | null {
+      if (store.all.length === 0) return null
+      return {
+        ptys: store.all.map((pty) => ({
+          id: pty.id,
+          ptyId: pty.ptyId,
+          title: pty.title,
+          titleNumber: pty.titleNumber,
+        })),
+        root: root(),
+        focusedPaneId: focusedPaneId(),
+      }
+    },
+    restore(snapshot: TerminalPtySnapshot) {
+      batch(() => {
+        setStore(
+          "all",
+          snapshot.ptys.map((p) => ({
+            id: p.id,
+            ptyId: p.ptyId,
+            title: p.title,
+            titleNumber: p.titleNumber,
+            status: "connecting" as TerminalStatus,
+            error: undefined,
+            restored: true,
+          })),
+        )
+        setRootSignal(clonePaneTree(snapshot.root))
+        setFocusedPaneId(snapshot.focusedPaneId)
+      })
+    },
     clear() {
       const ptyIds = store.all.map((pty) => pty.ptyId)
       reset()
@@ -633,7 +674,7 @@ function createWorkspaceTerminalSession(
     setStatus(id: string, ptyId: string, status: TerminalStatus, error?: TerminalFailure) {
       const index = store.all.findIndex((pty) => pty.id === id && pty.ptyId === ptyId)
       if (index === -1) return
-      setStore("all", index, { status, error: status === "ready" ? undefined : error })
+      setStore("all", index, { status, error: status === "ready" ? undefined : error, ...(status === "ready" ? { restored: false } : {}) })
     },
     update(input: Partial<LocalPTY> & { id: string }) {
       if (input.title === undefined) return
@@ -772,6 +813,16 @@ const { use: useTerminalDual, provider: TerminalProvider } = createSimpleContext
     bottomSession = createWorkspaceTerminalSession(sdk, runtime)
     sideSession = createWorkspaceTerminalSession(sdk, runtime)
 
+    // Restore previously-saved PTYs when returning to a directory.
+    // The cached snapshot was written by onCleanup — PTYs stayed alive on the
+    // server and just need to be re-registered in local reactive state.
+    const cached = directoryTerminalCache.get(sdk.directory)
+    if (cached) {
+      directoryTerminalCache.delete(sdk.directory)
+      if (cached.bottom) bottomSession.restore(cached.bottom)
+      if (cached.side) sideSession.restore(cached.side)
+    }
+
     const bottomReg = { dir: sdk.directory, scope, value: bottomSession }
     const sideReg = { dir: sdk.directory, scope, value: sideSession }
     sessions.add(bottomReg)
@@ -786,10 +837,14 @@ const { use: useTerminalDual, provider: TerminalProvider } = createSimpleContext
       onCleanup(() => clearInterval(timer))
     })
     onCleanup(() => {
+      // Save PTY state so terminals survive project navigation. Do NOT call
+      // clear() here — that would DELETE PTYs on the server. The PTYs keep
+      // running and are reconnected when the user returns to this directory.
+      const bottomSnap = bottomSession?.snapshot() ?? null
+      const sideSnap = sideSession?.snapshot() ?? null
+      directoryTerminalCache.set(sdk.directory, { bottom: bottomSnap, side: sideSnap })
       sessions.delete(bottomReg)
       sessions.delete(sideReg)
-      bottomSession?.clear()
-      sideSession?.clear()
     })
 
     return { bottom: bottomSession, side: sideSession }
diff --git a/packages/app/src/pages/session/session-side-panel.tsx b/packages/app/src/pages/session/session-side-panel.tsx
index ce5bc65d..1262d115 100644
--- a/packages/app/src/pages/session/session-side-panel.tsx
+++ b/packages/app/src/pages/session/session-side-panel.tsx
@@ -23,6 +23,7 @@ import { IdeFileEditor } from "@/pages/session/ide-file-editor"
 import { setSessionHandoff } from "@/pages/session/handoff"
 import { useSessionLayout } from "@/pages/session/session-layout"
 import { SidePanelSubagents } from "@/pages/session/side-panel-subagents"
+import { isSubagentInterrupted } from "@/pages/session/subagent-state"
 import { SidePanelBrowser } from "@/pages/session/side-panel-browser"
 import { SidePanelWorktree } from "@/pages/session/side-panel-worktree"
 import { SidePanelDebug } from "@/pages/session/side-panel-debug"
@@ -135,11 +136,7 @@ export function SessionSidePanel(props: {
     () => subagentChildren().filter((s) => sync.data.session_working(s.id)).length,
   )
   const interruptedSubagentCount = createMemo(
-    () =>
-      subagentChildren().filter((s) => {
-        const sub = (s.metadata?.["deepagent"] as { subagent?: { interrupted?: boolean } } | undefined)?.subagent
-        return sub?.interrupted === true
-      }).length,
+    () => subagentChildren().filter(isSubagentInterrupted).length,
   )
   const subagentAttentionCount = createMemo(
     () => runningSubagentCount() + interruptedSubagentCount(),
diff --git a/packages/app/src/pages/session/side-panel-subagents.tsx b/packages/app/src/pages/session/side-panel-subagents.tsx
index 801e13ad..91a616fd 100644
--- a/packages/app/src/pages/session/side-panel-subagents.tsx
+++ b/packages/app/src/pages/session/side-panel-subagents.tsx
@@ -6,6 +6,7 @@ import { useNavigate, useParams } from "@solidjs/router"
 import { useSDK } from "@/context/sdk"
 import { fetchCapabilities } from "@/components/deepagent/panel-goal.api"
 import { OversightDashboard } from "@/components/deepagent/oversight-dashboard"
+import { isSubagentInterrupted } from "./subagent-state"
 
 // Phase 2 (§3): SidePanelSubagents is now the single "子Agent监督" entry for the right rail.
 // It holds a `selectedSessionID` to track which subagent is being inspected; that selection
@@ -23,6 +24,7 @@ import { OversightDashboard } from "@/components/deepagent/oversight-dashboard"
 // (scope+route SessionStateKey): that never equals a child's parentID, so the list was always
 // empty. Resolving it internally here (like SidePanelIM / SidePanelDebug do) keeps the contract
 // simple and immune to that mismatch.
+
 export const SidePanelSubagents: Component<{ onClose: () => void }> = (props) => {
   const sync = useSync()
   const language = useLanguage()
@@ -56,10 +58,7 @@ export const SidePanelSubagents: Component<{ onClose: () => void }> = (props) =>
     const sub = (child.metadata?.["deepagent"] as { subagent?: { finished?: boolean } } | undefined)?.subagent
     return sub?.finished === true
   }
-  const isInterrupted = (child: { metadata?: Record<string, unknown> }): boolean => {
-    const sub = (child.metadata?.["deepagent"] as { subagent?: { interrupted?: boolean } } | undefined)?.subagent
-    return sub?.interrupted === true
-  }
+  const isInterrupted = isSubagentInterrupted
   const statusOf = (
     child: { id: string; metadata?: Record<string, unknown> },
   ): "running" | "finished" | "interrupted" | "idle" => {
diff --git a/packages/app/src/pages/session/subagent-state.test.ts b/packages/app/src/pages/session/subagent-state.test.ts
new file mode 100644
index 00000000..e5ff23d2
--- /dev/null
+++ b/packages/app/src/pages/session/subagent-state.test.ts
@@ -0,0 +1,10 @@
+import { describe, expect, test } from "bun:test"
+import { isSubagentInterrupted } from "./subagent-state"
+
+describe("isSubagentInterrupted", () => {
+  test("recognizes durable state and legacy boolean interruption markers", () => {
+    expect(isSubagentInterrupted({ deepagent: { subagent: { state: "interrupted" } } })).toBe(true)
+    expect(isSubagentInterrupted({ deepagent: { subagent: { interrupted: true } } })).toBe(true)
+    expect(isSubagentInterrupted({ deepagent: { subagent: { state: "finished" } } })).toBe(false)
+  })
+})
diff --git a/packages/app/src/pages/session/subagent-state.ts b/packages/app/src/pages/session/subagent-state.ts
new file mode 100644
index 00000000..188ceafd
--- /dev/null
+++ b/packages/app/src/pages/session/subagent-state.ts
@@ -0,0 +1,10 @@
+type SubagentMetadata = {
+  state?: string
+  interrupted?: boolean
+}
+
+/** Supports durable state markers and legacy boolean interruption markers. */
+export const isSubagentInterrupted = (metadata?: Record<string, unknown>) => {
+  const subagent = (metadata?.["deepagent"] as { subagent?: SubagentMetadata } | undefined)?.subagent
+  return subagent?.state === "interrupted" || subagent?.interrupted === true
+}
diff --git a/packages/app/src/pages/session/terminal-view.tsx b/packages/app/src/pages/session/terminal-view.tsx
index 51f95c3e..3ff8df0e 100644
--- a/packages/app/src/pages/session/terminal-view.tsx
+++ b/packages/app/src/pages/session/terminal-view.tsx
@@ -309,6 +309,7 @@ function TerminalSessionView(props: { pty: LocalPTY; focused: boolean }) {
               autoFocus={props.focused}
               runtimeId={terminal.runtimeId()}
               onStatusChange={(next, error) => terminal.setStatus(props.pty.id, ptyId, next, error)}
+              optimisticReady={props.pty.restored}
             />
           )}
         </Show>
diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts
index 715ec449..b8ff85ff 100644
--- a/packages/core/src/database/migration.gen.ts
+++ b/packages/core/src/database/migration.gen.ts
@@ -53,5 +53,6 @@ export const migrations = (
     import("./migration/20260712040000_deepagent_event_drop_distinct"),
     import("./migration/20260712050000_session_steer_queue"),
     import("./migration/20260719000000_deepagent_consumer_group"),
+    import("./migration/20260722000000_session_steer_correlation"),
   ])
 ).map((module) => module.default) satisfies DatabaseMigration.Migration[]
diff --git a/packages/core/src/database/migration/20260722000000_session_steer_correlation.ts b/packages/core/src/database/migration/20260722000000_session_steer_correlation.ts
new file mode 100644
index 00000000..472ff4d4
--- /dev/null
+++ b/packages/core/src/database/migration/20260722000000_session_steer_correlation.ts
@@ -0,0 +1,16 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260722000000_session_steer_correlation",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE session_steer ADD COLUMN correlation_id TEXT`)
+      yield* tx.run(`
+        CREATE UNIQUE INDEX IF NOT EXISTS session_steer_session_correlation_idx
+        ON session_steer (session_id, correlation_id)
+        WHERE correlation_id IS NOT NULL
+      `)
+    })
+  },
+} satisfies DatabaseMigration.Migration
diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts
index 1b678fa9..c509d4a0 100644
--- a/packages/core/src/session/sql.ts
+++ b/packages/core/src/session/sql.ts
@@ -185,11 +185,15 @@ export const SessionSteerTable = sqliteTable(
   "session_steer",
   {
     seq: integer().primaryKey({ autoIncrement: true }),
+    // Canonical durable/V1 identity — always server-minted. Never reuse a client optimistic id.
     id: text().$type<SessionMessage.ID>().notNull().unique(),
     session_id: text()
       .$type<SessionSchema.ID>()
       .notNull()
       .references(() => SessionTable.id, { onDelete: "cascade" }),
+    // Optional client retry key isolated from the canonical durable message id. Identical retries
+    // return the stored row; different payload for the same key is a CorrelationConflict.
+    correlation_id: text(),
     prompt: text({ mode: "json" }).notNull().$type<Prompt>(),
     delivery: text().$type<SessionInput.Delivery>().notNull(),
     consumed_seq: integer(),
@@ -199,6 +203,7 @@ export const SessionSteerTable = sqliteTable(
   },
   (table) => [
     index("session_steer_session_pending_seq_idx").on(table.session_id, table.consumed_seq, table.seq),
+    uniqueIndex("session_steer_session_correlation_idx").on(table.session_id, table.correlation_id),
   ],
 )
 
diff --git a/packages/deepagent-code/src/session/prompt.ts b/packages/deepagent-code/src/session/prompt.ts
index 5403b577..9b7f201e 100644
--- a/packages/deepagent-code/src/session/prompt.ts
+++ b/packages/deepagent-code/src/session/prompt.ts
@@ -61,7 +61,7 @@ import { Truncate } from "@/tool/truncate"
 import { Image } from "@/image/image"
 import { decodeDataUrl } from "@/util/data-url"
 import { Process } from "@/util/process"
-import { Cause, Effect, Exit, Latch, Layer, Option, Scope, Context, Schema, Types } from "effect"
+import { Cause, Data, Effect, Exit, Latch, Layer, Option, Scope, Context, Schema, Types } from "effect"
 import * as EffectLogger from "@deepagent-code/core/effect/logger"
 import { InstanceState } from "@/effect/instance-state"
 import { TaskTool, type TaskPromptOps } from "@/tool/task"
@@ -144,14 +144,48 @@ function isOrphanedInterruptedTool(part: SessionV1.ToolPart) {
 // isTerminalGoalPhase (kept as a local const to avoid a circular import: goal-manager imports this file).
 const TERMINAL_GOAL_PHASES: ReadonlySet<string> = new Set(["done", "needs_human", "rolled_back", "stopped"])
 
-// §S1.2 — extract the plain text a steer should carry from a PromptInput's parts (text parts only; file/
-// agent/subtask attachments are not re-encoded into a steer's text — a steered turn is a user message).
-const promptInputText = (parts: PromptInput["parts"]): string =>
-  parts
+class InvalidInput extends Data.TaggedError("SessionPrompt.InvalidInput")<{ readonly message: string }> {}
+
+// §S1.2 — convert PromptInput parts to the durable Prompt model used by the steer buffer.
+// All part types that have a Prompt equivalent are preserved; subtask parts are explicitly rejected
+// so they never produce a silent empty steer. The steer caller should surface this as a client error.
+const promptInputToPrompt = (
+  parts: PromptInput["parts"],
+): Effect.Effect<Prompt, InvalidInput> => {
+  if (parts.some((p) => p.type === "subtask"))
+    return Effect.fail(
+      new InvalidInput({ message: "Subtask prompt parts cannot be steered while a session is busy" }),
+    )
+  const text = parts
     .filter((p): p is Extract<PromptInput["parts"][number], { type: "text" }> => p.type === "text")
     .map((p) => p.text)
     .join("\n")
     .trim()
+  const files = parts
+    .filter((p): p is Extract<PromptInput["parts"][number], { type: "file" }> => p.type === "file")
+    .map(
+      (p) =>
+        new FileAttachment({
+          uri: p.url,
+          mime: p.mime,
+          ...(p.filename !== undefined ? { name: p.filename } : {}),
+        }),
+    )
+  const agents = parts
+    .filter((p): p is Extract<PromptInput["parts"][number], { type: "agent" }> => p.type === "agent")
+    .map((p) => new AgentAttachment({ name: p.name }))
+  if (text.length === 0 && files.length === 0 && agents.length === 0)
+    return Effect.fail(
+      new InvalidInput({ message: "Steer prompt must contain at least one supported part" }),
+    )
+  return Effect.succeed(
+    Prompt.fromUserMessage({
+      text,
+      ...(files.length === 0 ? {} : { files }),
+      ...(agents.length === 0 ? {} : { agents }),
+    }),
+  )
+}
 
 export interface Interface {
   readonly cancel: (sessionID: SessionID) => Effect.Effect<void>
@@ -159,7 +193,12 @@ export interface Interface {
   // V4.1 §S1.1: buffer a mid-turn user message into the durable steer queue for absorption at the next
   // model-request boundary of the live turn loop. This is the admit() API; S1.2 wires the busy-session
   // ingress that decides WHEN to route a message here vs. the normal prompt() path. Idempotent on `id`.
-  readonly steer: (input: SteerInput) => Effect.Effect<SessionSteer.Admitted>
+  readonly steer: (input: {
+    sessionID: SessionID
+    prompt: Prompt
+    delivery?: SessionSteer.Delivery
+    messageID?: SessionMessage.ID
+  }) => Effect.Effect<SessionSteer.Admitted>
   // V4.1 §S1.2: the busy-session ingress decision. If the session is IDLE (no live turn) → run a normal
   // turn (prompt). If it is BUSY (mid-turn) and steering is enabled → buffer the message as a steer so
   // the running turn absorbs it at its next boundary (delivery="goal_steer" when a non-terminal goal is
@@ -1913,10 +1952,8 @@ export const layer = Layer.effect(
     // turn. The steer id is an ascending SessionMessage.ID minted at admit time, so tail-sorting (Check 3)
     // is preserved. This replaces the earlier stamp-then-persist ordering, whose crash window between the
     // consume stamp and the message write could lose a steer permanently.
-    const steerPartID = (messageID: MessageID) =>
-      // Deterministic, valid PartID (prt_<message-suffix>) so a replayed persist targets the SAME part
-      // row and the projector's onConflictDoUpdate makes it a no-op instead of appending a duplicate.
-      PartID.make("prt_" + messageID.slice("msg_".length))
+    const steerPartID = (messageID: MessageID, suffix?: string) =>
+      PartID.make("prt_" + messageID.slice("msg_".length) + (suffix ?? ""))
     const drainSteers = Effect.fn("SessionPrompt.drainSteers")(function* (sessionID: SessionID) {
       if (!flags.v4Steering) return 0
       const pending = yield* steerBuffer.pending(sessionID)
@@ -1952,17 +1989,35 @@ export const layer = Layer.effect(
             ...(variant ? { variant } : {}),
           },
         }
-        // PERSIST-FIRST: materialize the history message + its text part (both keyed by the steer id, so
-        // a replay after a mid-drain crash is an idempotent upsert, not a duplicate) BEFORE stamping the
-        // steer consumed below.
+        // PERSIST-FIRST: materialize the history message and all durable parts before stamping consumed.
+        // Part IDs are derived from the steer id so post-crash replays are idempotent upserts.
         yield* sessions.updateMessage(info)
-        yield* sessions.updatePart({
-          id: steerPartID(info.id),
-          messageID: info.id,
-          sessionID,
-          type: "text",
-          text: admitted.prompt.text,
-        })
+        if (admitted.prompt.text.length > 0)
+          yield* sessions.updatePart({
+            id: steerPartID(info.id),
+            messageID: info.id,
+            sessionID,
+            type: "text",
+            text: admitted.prompt.text,
+          })
+        for (const [i, file] of (admitted.prompt.files ?? []).entries())
+          yield* sessions.updatePart({
+            id: steerPartID(info.id, `_f${i}`),
+            messageID: info.id,
+            sessionID,
+            type: "file",
+            url: file.uri,
+            mime: file.mime,
+            filename: file.name ?? file.uri,
+          })
+        for (const [i, agent] of (admitted.prompt.agents ?? []).entries())
+          yield* sessions.updatePart({
+            id: steerPartID(info.id, `_a${i}`),
+            messageID: info.id,
+            sessionID,
+            type: "agent",
+            name: agent.name,
+          })
         persisted.push(admitted.id)
         yield* elog.info("steer absorbed at boundary", { sessionID, messageID: info.id, seq: admitted.seq })
       }
@@ -2599,39 +2654,38 @@ export const layer = Layer.effect(
       },
     )
 
-    // V4.1 §S1.1: admit a mid-turn user message into the durable steer buffer. Pure buffering — it does
-    // NOT interrupt the in-flight run; the runLoop absorbs it at its next model-request boundary (see
-    // drainSteers below). With the kill-switch OFF this is a no-op guard: the caller (S1.2 ingress)
-    // must not route here when steering is disabled, but we defensively refuse to buffer so an
-    // orphaned steer can never accumulate undrained. Idempotent on `id`.
-    const steer: (input: SteerInput) => Effect.Effect<SessionSteer.Admitted> = Effect.fn("SessionPrompt.steer")(
-      function* (input: SteerInput) {
-        if (!flags.v4Steering)
-          return yield* Effect.die(new NamedError.Unknown({ message: "Steering is disabled (v4Steering=false)" }))
-        const prompt = Prompt.fromUserMessage({
-          text: input.text,
-          ...(input.files === undefined ? {} : { files: input.files }),
-          ...(input.agents === undefined ? {} : { agents: input.agents }),
-          ...(input.references === undefined ? {} : { references: input.references }),
-        })
-        // §S1.3 delivery channel: "goal_steer" is drained by the goal driver between ticks; "steer"
-        // (default) by the session's own runLoop. The two never contend on the same buffer rows.
-        const delivery = input.delivery ?? "steer"
-        const admitted = yield* steerBuffer.admit({
-          id: input.messageID ?? SessionMessage.ID.create(),
-          sessionID: input.sessionID,
-          prompt,
-          delivery,
-        })
-        yield* elog.info("steer admitted", {
-          sessionID: input.sessionID,
-          messageID: admitted.id,
-          seq: admitted.seq,
-          delivery,
-        })
-        return admitted
-      },
-    )
+    // V4.1 §S1.1: admit a mid-turn user message into the durable steer buffer.
+    // The canonical durable ID is always server-minted by admit(); the caller's messageID is used
+    // only as an optional correlationID for idempotent retries.
+    const steer: (input: {
+      sessionID: SessionID
+      prompt: Prompt
+      delivery?: SessionSteer.Delivery
+      messageID?: SessionMessage.ID
+    }) => Effect.Effect<SessionSteer.Admitted> = Effect.fn(
+      "SessionPrompt.steer",
+    )(function* (input) {
+      if (!flags.v4Steering)
+        return yield* Effect.die(new NamedError.Unknown({ message: "Steering is disabled (v4Steering=false)" }))
+      const delivery = input.delivery ?? "steer"
+      const admitted = yield* steerBuffer.admit({
+        sessionID: input.sessionID,
+        prompt: input.prompt,
+        delivery,
+        correlationID: input.messageID,
+      }).pipe(
+        Effect.catchTag("SessionSteer.CorrelationConflict", () =>
+          Effect.die(new NamedError.Unknown({ message: "Steer correlation conflict: duplicate follow-up" })),
+        ),
+      )
+      yield* elog.info("steer admitted", {
+        sessionID: input.sessionID,
+        messageID: admitted.id,
+        seq: admitted.seq,
+        delivery,
+      })
+      return admitted
+    })
 
     // V4.1 §S1.2 — the ingress decision. Both the HTTP prompt route and the IM agent executor call THIS
     // instead of prompt() directly, so the steer-vs-turn choice lives in exactly one place.
@@ -2660,23 +2714,22 @@ export const layer = Layer.effect(
       const goal = AgentGateway.DeepAgentSessionState.getActiveGoal(input.sessionID)
       const goalActive = goal != null && !TERMINAL_GOAL_PHASES.has(goal.phase)
       if (goalActive) {
-        const text = promptInputText(input.parts)
+        const steerPrompt = yield* promptInputToPrompt(input.parts).pipe(
+          Effect.catchTag("SessionPrompt.InvalidInput", (e) =>
+            Effect.die(e),
+          ),
+        )
         const admitted = yield* steer({
           sessionID: input.sessionID,
-          text,
+          prompt: steerPrompt,
           delivery: "goal_steer",
-          // §S1.2 ID bridge: forward the client-supplied messageID so the steer row carries the same
-          // id as the frontend's optimistic message — prevents a duplicate entry when drainSteers
-          // materialises the steer into V1 history. PromptInput.messageID is a different Schema brand
-          // (MessageID vs SessionMessage.ID) but both are ascending-string ids at runtime; the cast is
-          // safe because sendFollowupDraft always mints via Identifier.ascending().
           messageID: input.messageID as unknown as SessionMessage.ID | undefined,
         })
         // V4.1 governance audit — this is the REAL user goal-steer path (the ingress every busy-goal
         // steer flows through). Record the human intervention into the goal's Document Graph alongside
         // the per-tick worklog trail. Length only (not free-text) to keep the body bounded + PII-light;
         // best-effort (never blocks the steer). goal!.goalId is safe here: goalActive ⇒ goal != null.
-        writeGovernanceAudit(input.sessionID, goal!.goalId, "steer", { textChars: text.trim().length })
+        writeGovernanceAudit(input.sessionID, goal!.goalId, "steer", { textChars: steerPrompt.text.trim().length })
         return { kind: "steer" as const, delivery: "goal_steer" as const, admitted }
       }
       // (3) No active goal → a parent chat turn in flight becomes a chat steer.
@@ -2686,15 +2739,14 @@ export const layer = Layer.effect(
         const message = yield* prompt(input)
         return { kind: "turn" as const, message }
       }
-      // §S1.2 ID bridge: forward the client-supplied messageID so the steer row carries the same id as
-      // the frontend's optimistic message — prevents a duplicate entry when drainSteers materialises
-      // the steer into V1 history. PromptInput.messageID is a different Schema brand (MessageID vs
-      // SessionMessage.ID) but both are ascending-string ids at runtime; the cast is safe because
-      // sendFollowupDraft always mints via Identifier.ascending(). steer() will still generate a fresh
-      // ascending id when messageID is omitted (e.g. non-async callers that don't supply one).
+      const steerPrompt = yield* promptInputToPrompt(input.parts).pipe(
+        Effect.catchTag("SessionPrompt.InvalidInput", (e) =>
+          Effect.die(e),
+        ),
+      )
       const admitted = yield* steer({
         sessionID: input.sessionID,
-        text: promptInputText(input.parts),
+        prompt: steerPrompt,
         delivery: "steer",
         messageID: input.messageID as unknown as SessionMessage.ID | undefined,
       })
diff --git a/packages/deepagent-code/src/session/steer.ts b/packages/deepagent-code/src/session/steer.ts
index 68c5b4cf..6a3a653a 100644
--- a/packages/deepagent-code/src/session/steer.ts
+++ b/packages/deepagent-code/src/session/steer.ts
@@ -1,5 +1,5 @@
 import { and, asc, eq, inArray, isNull } from "drizzle-orm"
-import { Context, DateTime, Effect, Layer, Schema } from "effect"
+import { Context, Data, DateTime, Effect, Layer, Schema } from "effect"
 import { Database } from "@deepagent-code/core/database/database"
 import { SessionInput } from "@deepagent-code/core/session/input"
 import { SessionMessage } from "@deepagent-code/core/session/message"
@@ -37,6 +37,14 @@ import { SessionID } from "./schema"
 
 export type Delivery = SessionInput.Delivery
 
+// Raised when the same correlationID is reused with a different payload, which
+// would silently overwrite or ignore the earlier steer. Callers should surface
+// this as a 409-style client error.
+export class CorrelationConflict extends Data.TaggedError("SessionSteer.CorrelationConflict")<{
+  readonly sessionID: SessionID
+  readonly correlationID: string
+}> {}
+
 export class Admitted extends Schema.Class<Admitted>("SessionSteer.Admitted")({
   seq: Schema.Int,
   id: SessionMessage.ID,
@@ -60,14 +68,15 @@ const fromRow = (row: typeof SessionSteerTable.$inferSelect): Admitted =>
   })
 
 export interface Interface {
-  // Buffer a user message for later absorption. Idempotent on `id` (a duplicate admit is a no-op and
-  // returns the already-stored record) so an at-least-once ingress (S1.2) never double-buffers.
+  // Buffer a user message for later absorption. `id` is always server-minted.
+  // `correlationID` is an optional client retry key: identical payload retries return the stored row;
+  // different payload for the same key returns a CorrelationConflict (never silently drops).
   readonly admit: (input: {
-    readonly id: SessionMessage.ID
     readonly sessionID: SessionID
     readonly prompt: Prompt
     readonly delivery?: Delivery
-  }) => Effect.Effect<Admitted>
+    readonly correlationID?: string
+  }) => Effect.Effect<Admitted, CorrelationConflict>
   // NON-consuming read of pending steers for the session, in send-order (ascending `seq`). Persist-first
   // step 1: the runLoop reads these, materializes each as a V1 history message keyed by the steer id
   // (idempotent), THEN calls markConsumed. Reading does NOT mark anything — a crash before markConsumed
@@ -98,11 +107,13 @@ export const layer = Layer.effect(
   Effect.gen(function* () {
     const { db } = yield* Database.Service
 
-    const find = (id: SessionMessage.ID) =>
+    const findByCorrelation = (sessionID: SessionID, correlationID: string) =>
       db
         .select()
         .from(SessionSteerTable)
-        .where(eq(SessionSteerTable.id, id))
+        .where(
+          and(eq(SessionSteerTable.session_id, sessionID), eq(SessionSteerTable.correlation_id, correlationID)),
+        )
         .get()
         .pipe(
           Effect.orDie,
@@ -112,25 +123,33 @@ export const layer = Layer.effect(
     const admit: Interface["admit"] = Effect.fn("SessionSteer.admit")(function* (input) {
       const delivery = input.delivery ?? "steer"
       const timeCreated = DateTime.toEpochMillis(yield* DateTime.now)
+      // Always server-minted: the canonical durable/V1 message ID is never client-supplied.
+      const id = SessionMessage.ID.create()
       const inserted = yield* db
         .insert(SessionSteerTable)
         .values({
-          id: input.id,
+          id,
           session_id: input.sessionID,
+          correlation_id: input.correlationID,
           prompt: encodePrompt(input.prompt),
           delivery,
           time_created: timeCreated,
         })
-        .onConflictDoNothing({ target: SessionSteerTable.id })
+        .onConflictDoNothing()
         .returning()
         .get()
         .pipe(Effect.orDie)
       if (inserted) return fromRow(inserted)
-      // Lost the insert race (id already admitted) — return the stored record. Consume-once means the
-      // caller must never re-buffer, so surfacing the existing row is the correct idempotent answer.
-      const existing = yield* find(input.id)
-      if (existing) return existing
-      return yield* Effect.die("SessionSteer.admit: conflicting row vanished")
+      // Correlation conflict path: another row with the same (session, correlationID) already exists.
+      if (input.correlationID === undefined)
+        return yield* Effect.die("SessionSteer.admit: server-generated id conflicted (impossible)")
+      const existing = yield* findByCorrelation(input.sessionID, input.correlationID)
+      if (!existing) return yield* Effect.die("SessionSteer.admit: conflicting correlation row vanished")
+      // Identical payload = idempotent retry; different payload = explicit conflict.
+      if (existing.delivery === delivery && Prompt.equivalence(existing.prompt, input.prompt)) return existing
+      return yield* Effect.fail(
+        new CorrelationConflict({ sessionID: input.sessionID, correlationID: input.correlationID }),
+      )
     })
 
     const pending: Interface["pending"] = Effect.fn("SessionSteer.pending")(function* (sessionID, delivery = "steer") {
diff --git a/packages/deepagent-code/src/session/v4-event-runtime.ts b/packages/deepagent-code/src/session/v4-event-runtime.ts
index 6f4281ac..256c67f0 100644
--- a/packages/deepagent-code/src/session/v4-event-runtime.ts
+++ b/packages/deepagent-code/src/session/v4-event-runtime.ts
@@ -461,16 +461,50 @@ const anyV4DaemonEnabled = (flags: RuntimeFlags.Info): boolean =>
 // Register the durable delivery policy before any V4 publisher can emit. `Layer.mergeAll` starts
 // components concurrently, so relying on each consumer's live subscribe stream loses first-start events.
 // Individual consumers repeat this idempotently for standalone use; this layer establishes the runtime
-// producer-after-registration boundary. Disabled groups are deliberately absent and accrue no backlog.
-const consumerRegistrationLayer = Layer.effectDiscard(
+// producer-after-registration boundary. The policy lists ONLY groups owned by this runtime, so reconciling
+// disabled flags and releasing this scope cannot remove a group owned by another feature or instance.
+type RuntimeConsumerGroup = {
+  readonly id: string
+  readonly typeFilter?: string
+  readonly enabled: (flags: RuntimeFlags.Info) => boolean
+}
+
+const runtimeConsumerGroups: ReadonlyArray<RuntimeConsumerGroup> = [
+  { id: DISPATCH_GROUP, enabled: anyV4DaemonEnabled },
+  {
+    id: TICK_GROUP,
+    typeFilter: LMNEvents.GOAL_TICK_REQUESTED,
+    enabled: (flags) => flags.v4MultiAgentRuntime,
+  },
+  { id: CONVENE_GROUP, enabled: (flags) => flags.v4PanelAutoConvene },
+  { id: ARCHIVE_GROUP, enabled: (flags) => flags.v4EventDrivenArchive || flags.v4MultiAgentRuntime },
+  { id: NOTIFY_GROUP, enabled: (flags) => flags.v4AgentPushEnabled },
+]
+
+// Exported for direct lifecycle testing. On startup this reconciles historical V4 groups left by an older
+// runtime: enabled groups are registered and disabled groups are removed. Scope release removes exactly
+// the enabled groups this runtime owned, stopping future publishes from accruing offline deliveries.
+export const consumerRegistrationLayer = Layer.effectDiscard(
   Effect.gen(function* () {
     const flags = yield* RuntimeFlags.Service
     const bus = yield* DeepAgentEventBus.Service
-    if (anyV4DaemonEnabled(flags)) yield* bus.registerConsumerGroup(DISPATCH_GROUP)
-    if (flags.v4MultiAgentRuntime) yield* bus.registerConsumerGroup(TICK_GROUP, LMNEvents.GOAL_TICK_REQUESTED)
-    if (flags.v4PanelAutoConvene) yield* bus.registerConsumerGroup(CONVENE_GROUP)
-    if (flags.v4EventDrivenArchive || flags.v4MultiAgentRuntime) yield* bus.registerConsumerGroup(ARCHIVE_GROUP)
-    if (flags.v4AgentPushEnabled) yield* bus.registerConsumerGroup(NOTIFY_GROUP)
+    const enabled = runtimeConsumerGroups.filter((group) => group.enabled(flags))
+    const disabled = runtimeConsumerGroups.filter((group) => !group.enabled(flags))
+
+    yield* Effect.forEach(enabled, (group) => bus.registerConsumerGroup(group.id, group.typeFilter), {
+      concurrency: "unbounded",
+      discard: true,
+    })
+    yield* Effect.forEach(disabled, (group) => bus.unregisterConsumerGroup(group.id), {
+      concurrency: "unbounded",
+      discard: true,
+    })
+    yield* Effect.addFinalizer(() =>
+      Effect.forEach(enabled, (group) => bus.unregisterConsumerGroup(group.id), {
+        concurrency: "unbounded",
+        discard: true,
+      }),
+    )
   }),
 )
 
diff --git a/packages/deepagent-code/src/tool/task_read.ts b/packages/deepagent-code/src/tool/task_read.ts
index 58f2c4f3..02d3c646 100644
--- a/packages/deepagent-code/src/tool/task_read.ts
+++ b/packages/deepagent-code/src/tool/task_read.ts
@@ -1,4 +1,5 @@
 import * as Tool from "./tool"
+import { MessageV2 } from "@/session/message-v2"
 import { Session } from "@/session/session"
 import { SessionV1 } from "@deepagent-code/core/v1/session"
 import { Effect, Schema } from "effect"
@@ -92,21 +93,18 @@ export const TaskReadTool = Tool.define(
         )
       }
 
-      // Read messages — Session.messages returns them oldest-first, we reverse for newest-first cursor.
-      const allMessages = yield* sessions
-        .messages({ sessionID: childSessionID, limit: MAX_LIMIT + 1 })
-        .pipe(Effect.catchCause(() => Effect.succeed([] as SessionV1.WithParts[])))
-
-      // Apply `before` cursor (message ID boundary for pagination).
-      const beforeID = params.before
-      const filteredMessages = beforeID
-        ? allMessages.filter((m) => m.info.id < beforeID)
-        : allMessages
-
-      // Take newest `limit` messages.
-      const page = filteredMessages.slice(-limit)
-      const hasMore = filteredMessages.length > limit
-      const nextCursor = page[0]?.info.id
+      // MessageV2.page applies the opaque cursor in storage and returns chronological items.
+      const result = yield* MessageV2.page({
+        sessionID: childSessionID,
+        limit,
+        before: params.before,
+      }).pipe(Effect.catchCause(() => Effect.succeed({ items: [] as SessionV1.WithParts[], more: false, cursor: undefined as string | undefined })))
+      const page = result.items
+      const nextCursor: string | undefined = result.cursor
+      // A cursor is the only valid continuation token. Never advertise another page when a
+      // storage implementation reports `more` without one: callers would resend `undefined`
+      // and restart from the newest messages.
+      const hasMore = result.more && nextCursor !== undefined
 
       // Read durable state from metadata.
       const deepagent = child.metadata?.["deepagent"] as Record<string, unknown> | undefined
@@ -167,7 +165,13 @@ export const TaskReadTool = Tool.define(
 
       return {
         title: `Task transcript: ${child.title ?? childSessionID}`,
-        metadata: { sessionID: childSessionID, state: durableState, messageCount: page.length, hasMore },
+        metadata: {
+          sessionID: childSessionID,
+          state: durableState,
+          messageCount: page.length,
+          hasMore,
+          ...(nextCursor !== undefined ? { before: nextCursor } : {}),
+        },
         output: transcript + paginationHint,
       }
     })
@@ -176,7 +180,7 @@ export const TaskReadTool = Tool.define(
       description: DESCRIPTION,
       parameters: Parameters,
       execute: (params: Schema.Schema.Type<typeof Parameters>, ctx: Tool.Context) =>
-        run(params, ctx).pipe(Effect.catchCause((cause) => Effect.die(cause))),
+        run(params, ctx).pipe(Effect.catchCause((cause) => Effect.die(cause))) as unknown as Effect.Effect<Tool.ExecuteResult>,
     }
   }),
 )
diff --git a/packages/deepagent-code/test/session/steer.test.ts b/packages/deepagent-code/test/session/steer.test.ts
index 5c32d47f..dbda5744 100644
--- a/packages/deepagent-code/test/session/steer.test.ts
+++ b/packages/deepagent-code/test/session/steer.test.ts
@@ -321,9 +321,9 @@ off.instance("admit buffers steers, pending returns them in send-order, markCons
     const sessions = yield* Session.Service
     const chat = yield* sessions.create({ title: "Steer unit" })
 
-    yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("first") })
-    yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("second") })
-    yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("third") })
+    yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("first") })
+    yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("second") })
+    yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("third") })
 
     expect(yield* steer.hasPending(chat.id)).toBe(true)
 
@@ -358,9 +358,8 @@ off.instance(
 
       // On the SAME session id: one parent-chat steer (delivery="steer") and one goal-directed steer
       // (delivery="goal_steer"). This models a goal running in a session whose parent runLoop is also live.
-      yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("chat steer") })
+      yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("chat steer") })
       yield* steer.admit({
-        id: SessionMessage.ID.create(),
         sessionID: chat.id,
         prompt: mkPrompt("goal guidance"),
         delivery: "goal_steer",
@@ -396,10 +395,10 @@ off.instance("admit is idempotent on message id (no double-buffer)", () =>
     const steer = yield* SessionSteer.Service
     const sessions = yield* Session.Service
     const chat = yield* sessions.create({ title: "Steer idempotent" })
-    const id = SessionMessage.ID.create()
+    const correlationID = SessionMessage.ID.create()
 
-    const a = yield* steer.admit({ id, sessionID: chat.id, prompt: mkPrompt("once") })
-    const b = yield* steer.admit({ id, sessionID: chat.id, prompt: mkPrompt("once") })
+    const a = yield* steer.admit({ correlationID, sessionID: chat.id, prompt: mkPrompt("once") })
+    const b = yield* steer.admit({ correlationID, sessionID: chat.id, prompt: mkPrompt("once") })
     expect(a.seq).toBe(b.seq)
 
     const drained = yield* steer.pending(chat.id)
@@ -418,7 +417,6 @@ off.instance("consume-once survives a fresh drain cycle (durable, no double-appl
     const chat = yield* sessions.create({ title: "Steer durable" })
 
     const admitted = yield* steer.admit({
-      id: SessionMessage.ID.create(),
       sessionID: chat.id,
       prompt: mkPrompt("persisted"),
     })
@@ -479,7 +477,6 @@ off.instance(
       const chat = yield* sessions.create({ title: "Steer crash window" })
 
       const admitted = yield* steer.admit({
-        id: SessionMessage.ID.create(),
         sessionID: chat.id,
         prompt: mkPrompt("DONT-LOSE-ME"),
       })
@@ -548,7 +545,6 @@ on.instance(
 
       // Admit the steer while the first model request is in flight.
       const admitted = yield* steer.admit({
-        id: SessionMessage.ID.create(),
         sessionID: chat.id,
         prompt: mkPrompt("STEERED-MESSAGE"),
       })
@@ -613,7 +609,7 @@ on.instance(
         .prompt({ sessionID: chat.id, agent: "build", model: ref, parts: [{ type: "text", text: "initial" }] })
         .pipe(Effect.forkChild)
       yield* llm.wait(1)
-      yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("STEER") })
+      yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("STEER") })
       yield* Deferred.succeed(gate, void 0)
       yield* Fiber.await(fiber)
 
@@ -660,7 +656,7 @@ off.instance(
       })
 
       // Pre-buffer a steer directly (bypassing ingress) so we can prove the loop ignores it when OFF.
-      yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: chat.id, prompt: mkPrompt("IGNORED-STEER") })
+      yield* steer.admit({ sessionID: chat.id, prompt: mkPrompt("IGNORED-STEER") })
 
       yield* llm.text("done")
       const result = yield* prompt.prompt({
@@ -832,7 +828,6 @@ on.instance(
 
       // A steer lands in the isBusy→admit race window (buffered while the session is idle).
       const admitted = yield* steer.admit({
-        id: SessionMessage.ID.create(),
         sessionID: drained.id,
         prompt: mkPrompt("STEP0-STEER"),
       })
@@ -864,7 +859,7 @@ on.instance(
       yield* llm.text("first-answer-2")
       yield* prompt.prompt({ sessionID: normal.id, agent: "build", model: ref, parts: [{ type: "text", text: "initial" }] })
       const before = yield* llm.calls
-      yield* steer.admit({ id: SessionMessage.ID.create(), sessionID: normal.id, prompt: mkPrompt("NOT-DRAINED") })
+      yield* steer.admit({ sessionID: normal.id, prompt: mkPrompt("NOT-DRAINED") })
 
       // Default loop() → drainFirst=false. Step 0 does NOT drain; the loop breaks at the finish check.
       yield* prompt.loop({ sessionID: normal.id })
diff --git a/packages/deepagent-code/test/session/v4-event-runtime.test.ts b/packages/deepagent-code/test/session/v4-event-runtime.test.ts
index 3e4fe3dc..31cf452e 100644
--- a/packages/deepagent-code/test/session/v4-event-runtime.test.ts
+++ b/packages/deepagent-code/test/session/v4-event-runtime.test.ts
@@ -1,5 +1,6 @@
 import { describe, expect } from "bun:test"
-import { Deferred, Effect, Exit, Fiber, Layer } from "effect"
+import { Context, Deferred, Effect, Exit, Fiber, Layer } from "effect"
+import * as Scope from "effect/Scope"
 import { V4EventRuntime } from "../../src/session/v4-event-runtime"
 import { DeepAgentEventBus } from "@deepagent-code/core/deepagent/deepagent-event-bus"
 import { DeepAgentEvent } from "@deepagent-code/core/deepagent/deepagent-event"
@@ -59,8 +60,85 @@ describe("V4EventRuntime.layer", () => {
   )
 })
 
-// P1.6 — the production schedule bootstrap. Proves the tick loop now has real rows: registration is
-// flag-gated, idempotent across restarts, and the "3× CI failure → repair" condition fires when seeded.
+// Durable group lifecycle: registration must be reconciled when flags change, and scope release must
+// unregister only the groups this runtime owns so later publishes cannot leave an offline backlog.
+describe("V4EventRuntime durable consumer-group lifecycle", () => {
+  const staleRuntimeGroups = ["event-dispatcher", "goal-tick-consumer", "panel-convener", "wiki-archiver", "supervisor-notifier"]
+  const externalGroup = "other-feature-consumer"
+
+  const registration = (flags: Partial<RuntimeFlags.Info>) =>
+    V4EventRuntime.consumerRegistrationLayer.pipe(Layer.provide(RuntimeFlags.layer(flags)))
+
+  const fullRuntimeFlagsOff: Partial<RuntimeFlags.Info> = {
+    v4MultiAgentRuntime: false,
+    v4EventDrivenIm: false,
+    v4PanelAutoConvene: false,
+    v4EventDrivenArchive: false,
+    v4AgentPushEnabled: false,
+  }
+
+  const publish = (key: string): DeepAgentEvent.PublishInput => ({
+    type: "monitor.alert",
+    source: "monitor",
+    workspaceID: "wrk_1",
+    idempotencyKey: key,
+    priority: "normal",
+    payload: {},
+  })
+
+  baseIt.effect("enabled runtime registers its groups; a subsequent disabled startup removes only those historical groups", () =>
+    Effect.gen(function* () {
+      const busScope = yield* Scope.make()
+      const busContext = yield* Layer.build(
+        DeepAgentEventBus.layer.pipe(Layer.provideMerge(Database.layerFromPath(":memory:"))),
+      ).pipe(Scope.provide(busScope))
+      const bus = Context.get(busContext, DeepAgentEventBus.Service)
+
+      const enabledScope = yield* Scope.make()
+      yield* Layer.build(registration({ v4MultiAgentRuntime: true })).pipe(
+        Scope.provide(enabledScope),
+        Effect.provide(busContext),
+      )
+      yield* bus.registerConsumerGroup(externalGroup)
+      yield* Scope.close(enabledScope, Exit.void)
+
+      const disabledScope = yield* Scope.make()
+      yield* Layer.build(registration(fullRuntimeFlagsOff)).pipe(
+        Scope.provide(disabledScope),
+        Effect.provide(busContext),
+      )
+      const event = yield* bus.publish(publish("flags-off"))
+      const due = yield* bus.dueRetries(Number.MAX_SAFE_INTEGER)
+
+      expect(due.some((delivery) => delivery.eventID === event.id && staleRuntimeGroups.includes(delivery.subscriptionGroup))).toBe(false)
+      expect(due.some((delivery) => delivery.eventID === event.id && delivery.subscriptionGroup === externalGroup)).toBe(true)
+      yield* Scope.close(disabledScope, Exit.void)
+      yield* Scope.close(busScope, Exit.void)
+    }),
+  )
+
+  baseIt.effect("scope release unregisters enabled groups, so later publishes create no V4 delivery", () =>
+    Effect.gen(function* () {
+      const busScope = yield* Scope.make()
+      const busContext = yield* Layer.build(
+        DeepAgentEventBus.layer.pipe(Layer.provideMerge(Database.layerFromPath(":memory:"))),
+      ).pipe(Scope.provide(busScope))
+      const bus = Context.get(busContext, DeepAgentEventBus.Service)
+      const registrationScope = yield* Scope.make()
+      yield* Layer.build(registration({ v4MultiAgentRuntime: true })).pipe(
+        Scope.provide(registrationScope),
+        Effect.provide(busContext),
+      )
+      yield* Scope.close(registrationScope, Exit.void)
+
+      const event = yield* bus.publish(publish("after-release"))
+      const due = yield* bus.dueRetries(Number.MAX_SAFE_INTEGER)
+      expect(due.some((delivery) => delivery.eventID === event.id && staleRuntimeGroups.includes(delivery.subscriptionGroup))).toBe(false)
+      yield* Scope.close(busScope, Exit.void)
+    }),
+  )
+})
+
 describe("V4EventRuntime schedule bootstrap", () => {
   const database = Database.layerFromPath(":memory:")
   const it = testEffect(Scheduler.defaultLayer.pipe(Layer.provideMerge(database)))
diff --git a/packages/deepagent-code/test/tool/task-read.test.ts b/packages/deepagent-code/test/tool/task-read.test.ts
new file mode 100644
index 00000000..8e40c4cb
--- /dev/null
+++ b/packages/deepagent-code/test/tool/task-read.test.ts
@@ -0,0 +1,107 @@
+import { afterEach, describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { Agent } from "../../src/agent/agent"
+import { BackgroundJob } from "@/background/job"
+import { EventV2Bridge } from "@/event-v2-bridge"
+import { Config } from "@/config/config"
+import { CrossSpawnSpawner } from "@deepagent-code/core/cross-spawn-spawner"
+import { Session } from "@/session/session"
+import { SessionRunState } from "@/session/run-state"
+import { SessionStatus } from "@/session/status"
+import { Database } from "@deepagent-code/core/database/database"
+import { MessageID, PartID, SessionID } from "../../src/session/schema"
+import { ModelV2 } from "@deepagent-code/core/model"
+import { ProviderV2 } from "@deepagent-code/core/provider"
+import { TaskReadTool } from "../../src/tool/task_read"
+import { Truncate } from "@/tool/truncate"
+import { ToolRegistry } from "@/tool/registry"
+import { RuntimeFlags } from "@/effect/runtime-flags"
+import { disposeAllInstances } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+afterEach(async () => {
+  await disposeAllInstances()
+})
+
+const it = testEffect(
+  Layer.mergeAll(
+    Agent.defaultLayer,
+    BackgroundJob.defaultLayer,
+    EventV2Bridge.defaultLayer,
+    Config.defaultLayer,
+    CrossSpawnSpawner.defaultLayer,
+    Session.defaultLayer,
+    SessionRunState.defaultLayer,
+    SessionStatus.defaultLayer,
+    Truncate.defaultLayer,
+    ToolRegistry.defaultLayer,
+    Database.defaultLayer,
+    RuntimeFlags.layer(),
+  ),
+)
+
+const execCtx = (sessionID: SessionID) => ({
+  sessionID,
+  messageID: MessageID.ascending(),
+  agent: "build",
+  abort: new AbortController().signal,
+  extra: {},
+  messages: [],
+  metadata: () => Effect.void,
+  ask: () => Effect.void,
+})
+
+const addTextMessage = (sessionID: SessionID, text: string, created = Date.now()) =>
+  Effect.gen(function* () {
+    const sessions = yield* Session.Service
+    const message = yield* sessions.updateMessage({
+      id: MessageID.ascending(),
+      role: "user",
+      sessionID,
+      agent: "build",
+      model: { providerID: ProviderV2.ID.make("test"), modelID: ModelV2.ID.make("test") },
+      time: { created },
+    })
+    yield* sessions.updatePart({
+      id: PartID.ascending(),
+      sessionID,
+      messageID: message.id,
+      type: "text",
+      text,
+    })
+  })
+
+const readTexts = (output: string) => [...output.matchAll(/<message[^>]*>\s*([^<]+?)\s*<\/message>/g)].map((match) => match[1])
+
+describe("tool.task_read", () => {
+  it.instance("pages 203 child messages through storage cursors without duplicates", () =>
+    Effect.gen(function* () {
+      const sessions = yield* Session.Service
+      const parent = yield* sessions.create({ title: "Parent" })
+      const child = yield* sessions.create({ parentID: parent.id, agent: "general", title: "Long task" })
+      const expected = Array.from({ length: 203 }, (_, index) => `message-${String(index + 1).padStart(3, "0")}`)
+
+      for (const [index, text] of expected.entries()) yield* addTextMessage(child.id, text, 1_700_000_000_000 + index)
+
+      const tool = yield* TaskReadTool
+      const def = yield* tool.init()
+      const first = yield* def.execute({ task_id: child.id, limit: 100 }, execCtx(parent.id))
+      const second = yield* def.execute(
+        { task_id: child.id, limit: 100, before: first.metadata.before },
+        execCtx(parent.id),
+      )
+      const third = yield* def.execute(
+        { task_id: child.id, limit: 100, before: second.metadata.before },
+        execCtx(parent.id),
+      )
+
+      expect(first.metadata.hasMore).toBe(true)
+      expect(second.metadata.hasMore).toBe(true)
+      expect(third.metadata.hasMore).toBe(false)
+      expect(readTexts(first.output)).toEqual(expected.slice(103))
+      expect(readTexts(second.output)).toEqual(expected.slice(3, 103))
+      expect(readTexts(third.output)).toEqual(expected.slice(0, 3))
+      expect([...readTexts(third.output), ...readTexts(second.output), ...readTexts(first.output)]).toEqual(expected)
+    }),
+  )
+})