fix(session): preserve recovery after empty stops and surface request-phase retries

packages/cli/src/cli/cmd/import.ts

@@ -5,7 +5,7 @@ import { Session } from "../../session"
 import { MessageV2 } from "../../session/message-v2"
 import { cmd } from "./cmd"
 import { bootstrap } from "../bootstrap"
-import { Database } from "../../storage"
+import { Database, eq } from "../../storage"
 import { SessionTable, MessageTable, PartTable } from "../../session/session.sql"
 import { Instance } from "../../project/instance"
 import { ShareNext } from "../../share"
@@ -95,7 +95,15 @@ export function storeImportedSession(
       const ids = []
       for (const part of msg.parts) {
         const partInfo = MessageV2.Part.parse(part)
+        if (partInfo.type === "tool" && partInfo.execution === "not-started") delete partInfo.execution
         const { id: partId, sessionID: _s, messageID, ...partData } = partInfo
+        const current = tx.select().from(PartTable).where(eq(PartTable.id, partId)).get()
+        if (current && (current.data.type === "tool" || partInfo.type === "tool")) {
+          if (current.session_id !== row.id || current.message_id !== messageID ||
+            current.data.type !== "tool" || partInfo.type !== "tool") throw new Error("Imported tool identity mismatch")
+          const previous = MessageV2.ToolPart.parse({ ...current.data, id: current.id, sessionID: current.session_id, messageID: current.message_id })
+          if (previous.callID !== partInfo.callID) throw new Error("Imported tool call identity mismatch")
+        }
         tx.insert(PartTable)
           .values({ id: partId, message_id: messageID, session_id: row.id, data: partData })
           .onConflictDoNothing()

packages/cli/src/effect/instance-state.ts

@@ -1,4 +1,4 @@
-import { Effect, Fiber, ScopedCache, Scope, Context } from "effect"
+import { Deferred, Effect, Fiber, ScopedCache, Scope, Context } from "effect"
 import * as EffectLogger from "./logger"
 import { Instance, type InstanceContext } from "@/project/instance"
 import { LocalContext } from "@/util"
@@ -10,7 +10,7 @@ const TypeId = "~opencode/InstanceState"
 
 export interface InstanceState<A, E = never, R = never> {
   readonly [TypeId]: typeof TypeId
-  readonly cache: ScopedCache.ScopedCache<string, A, E, R>
+  readonly cache: ScopedCache.ScopedCache<string, Deferred.Deferred<A, E>, never, R>
 }
 
 export const bind = <F extends (...args: any[]) => any>(fn: F): F => {
@@ -41,12 +41,27 @@ export const make = <A, E = never, R = never>(
   opts?: { phase?: "normal" | "late" },
 ): Effect.Effect<InstanceState<A, E, Exclude<R, Scope.Scope>>, never, R | Scope.Scope> =>
   Effect.gen(function* () {
-    const cache = yield* ScopedCache.make<string, A, E, R>({
+    const cache = yield* ScopedCache.make<string, Deferred.Deferred<A, E>, never, R>({
       capacity: Number.POSITIVE_INFINITY,
       lookup: () =>
         Effect.gen(function* () {
-          return yield* init(yield* context)
-        }),
+          const ctx = yield* context
+          const workspace = yield* workspaceID
+          const entry = yield* Effect.scope
+          // Register resources first so entry closure stops their users before releasing them.
+          const resources = yield* Scope.fork(entry)
+          const result = yield* Deferred.make<A, E>()
+          yield* Effect.suspend(() => init(ctx)).pipe(
+            Scope.provide(resources),
+            Effect.provideService(InstanceRef, ctx),
+            Effect.provideService(WorkspaceRef, workspace),
+            Effect.interruptible,
+            Deferred.into(result),
+            // Enter Deferred.into even if invalidation precedes the fiber's first scheduled step.
+            Effect.forkIn(entry, { uninterruptible: true }),
+          )
+          return result
+        }).pipe(Effect.uninterruptible),
     })
 
     const off = registerDisposer(
@@ -63,7 +78,8 @@ export const make = <A, E = never, R = never>(
 
 export const get = <A, E, R>(self: InstanceState<A, E, R>) =>
   Effect.gen(function* () {
-    return yield* ScopedCache.get(self.cache, yield* directory)
+    const result = yield* ScopedCache.get(self.cache, yield* directory).pipe(Effect.uninterruptible)
+    return yield* Deferred.await(result)
   })
 
 export const use = <A, E, R, B>(self: InstanceState<A, E, R>, select: (value: A) => B) => Effect.map(get(self), select)

packages/cli/src/effect/runner.ts

@@ -13,7 +13,7 @@ export interface Runner<A, E = never, B = never> {
    */
   readonly startOwned: (
     work: Effect.Effect<A, E>,
-  ) => Effect.Effect<{ readonly runId: number; readonly interruptOwned: Effect.Effect<void> }, B>
+  ) => Effect.Effect<{ readonly runId: number; readonly interruptOwned: Effect.Effect<void>; readonly awaitExit: Effect.Effect<void> }, B>
   readonly startShell: (work: Effect.Effect<A, E>) => Effect.Effect<A, E | B>
   readonly cancel: Effect.Effect<void>
 }
@@ -98,6 +98,7 @@ export const make = <A, E = never, B = never>(
     Effect.gen(function* () {
       const id = next()
       const release = opts?.onRunStart ? yield* opts.onRunStart : () => {}
+      yield* busy
       const fiber = yield* work.pipe(
         Effect.onExit((exit) => finishRun(id, done, exit)),
         Effect.forkIn(scope),
@@ -344,18 +345,19 @@ export const make = <A, E = never, B = never>(
 
   const startOwned = (
     work: Effect.Effect<A, E>,
-  ): Effect.Effect<{ readonly runId: number; readonly interruptOwned: Effect.Effect<void> }, B> =>
+  ): Effect.Effect<{ readonly runId: number; readonly interruptOwned: Effect.Effect<void>; readonly awaitExit: Effect.Effect<void> }, B> =>
     SynchronizedRef.modifyEffect(
       ref,
       Effect.fnUntraced(function* (st) {
         if (st._tag !== "Idle") {
-          return [busyFailure<{ runId: number; interruptOwned: Effect.Effect<void> }>(), st] as const
+          return [busyFailure<{ runId: number; interruptOwned: Effect.Effect<void>; awaitExit: Effect.Effect<void> }>(), st] as const
         }
         const done = yield* Deferred.make<A, E | Cancelled>()
         const run = yield* startRun(work, done)
         const owned = {
           runId: run.id,
           interruptOwned: interruptOwned(run.id),
+          awaitExit: Fiber.await(run.fiber).pipe(Effect.asVoid),
         }
         return [Effect.succeed(owned), { _tag: "Running", run } as const] as const
       }),

packages/cli/src/plugin/index.ts

@@ -30,8 +30,9 @@ import { PoeAuthPlugin } from "opencode-poe-auth"
 import { CloudflareAIGatewayAuthPlugin, CloudflareWorkersAuthPlugin } from "./cloudflare"
 import { CheckpointSplitoverPlugin } from "./checkpoint-splitover"
 import { SubagentProgressCheckerPlugin } from "./subagent-progress-checker"
-import { Effect, Layer, Context, Stream } from "effect"
+import { Effect, Exit, Layer, Context, Stream } from "effect"
 import { EffectBridge } from "@/effect"
+import * as ToolSettlement from "@/tool/settlement"
 import { InstanceState } from "@/effect"
 import { errorMessage } from "@/util/error"
 import { PluginLoader } from "./loader"
@@ -631,11 +632,16 @@ export const layer = Layer.effect(
       if (!name) return output
       const s = yield* InstanceState.get(state)
       const fh = yield* freshFileHooks
+      const toolHook = name === "tool.execute.before" || name === "tool.execute.after"
 
       for (const entry of s.hooksWithMeta ?? []) {
         const fn = entry?.hook?.[name] as ((input: Input, output: Output) => unknown) | undefined
         if (typeof fn !== "function") continue
-        yield* Effect.promise(async () => fn(input, output))
+        yield* toolHook
+          ? ToolSettlement.promise(async () => fn(input, output), (exit) =>
+              Exit.isSuccess(exit) ? Effect.succeed(exit.value) : Effect.failCause(exit.cause),
+            ).pipe(Effect.orDie)
+          : Effect.promise(async () => fn(input, output))
       }
 
       for (const entry of fh.meta ?? []) {
@@ -649,17 +655,7 @@ export const layer = Layer.effect(
         }
 
         const snapshot = structuredClone(output)
-        const failed = yield* Effect.tryPromise({
-          try: async () => {
-            await Promise.race([
-              Promise.resolve(fn(input, output)),
-              new Promise<never>((_, reject) =>
-                setTimeout(() => reject(new Error(`hook timed out after ${HOOK_TIMEOUT_MS}ms`)), HOOK_TIMEOUT_MS),
-              ),
-            ])
-          },
-          catch: (err) => err,
-        }).pipe(
+        const finish = (effect: Effect.Effect<unknown, unknown>) => effect.pipe(
           Effect.map(() => false),
           Effect.catch((err) => {
             Object.assign(output as any, snapshot)
@@ -675,6 +671,33 @@ export const layer = Layer.effect(
             return Effect.succeed(true)
           }),
         )
+        const failed = yield* toolHook
+          ? ToolSettlement.promise(() => {
+              const pending = Promise.resolve().then(() => fn(input, output))
+              let timer: ReturnType<typeof setTimeout>
+              const timed = Promise.race([
+                pending,
+                new Promise<never>((_, reject) => {
+                  timer = setTimeout(() => reject(new Error(`hook timed out after ${HOOK_TIMEOUT_MS}ms`)), HOOK_TIMEOUT_MS)
+                }),
+              ])
+              return timed.catch((error) => pending.then(
+                () => { throw error },
+                () => { throw error },
+              )).finally(() => clearTimeout(timer))
+            }, (exit) => finish(Exit.isSuccess(exit) ? Effect.succeed(exit.value) : Effect.failCause(exit.cause)))
+              .pipe(Effect.orDie)
+          : finish(Effect.tryPromise({
+              try: async () => {
+                await Promise.race([
+                  Promise.resolve(fn(input, output)),
+                  new Promise<never>((_, reject) =>
+                    setTimeout(() => reject(new Error(`hook timed out after ${HOOK_TIMEOUT_MS}ms`)), HOOK_TIMEOUT_MS),
+                  ),
+                ])
+              },
+              catch: (err) => err,
+            }))
         if (!failed) hookFailures.delete(hookID)
       }
       return output

packages/cli/src/provider/host-transport.ts

@@ -1,8 +1,12 @@
+import { Context } from "effect"
+
 /** Optional in-process embedding port. No application policy lives in the engine. */
 export interface ModelCallScope {
   sessionID: string
   userMessageID: string
   assistantMessageID?: string
+  executionAttempt?: ExecutionAttempt
+  directory?: string
   providerID: string
   modelID: string
   sdk: string
@@ -11,7 +15,24 @@ export interface ModelCallScope {
   format?: string
 }
 
+export interface ExecutionAttempt { requestId: string; attemptId: string }
+export interface ExecutionAdmission {
+  attempt: ExecutionAttempt
+  directory: string
+  sessionID: string
+  userMessageID: string
+  target: { userMessageID?: string; assistantMessageID?: string }
+}
+
+export const ExecutionScope = Context.Reference<Omit<ExecutionAdmission, "target"> | undefined>("~opencode/HostExecutionScope", {
+  defaultValue: () => undefined,
+})
+
+export interface ExecutionActivation { beforeWork: () => Promise<void> }
 export interface HostTransport {
+  prepareExecution?(input: ExecutionAdmission): void | Promise<void>
+  admitExecution?(input: ExecutionAdmission): ExecutionActivation | void
+  finishExecution?(attempt: ExecutionAttempt): void
   userMessage(input: { sessionID: string; userMessageID: string; parts: readonly unknown[] }): void
   modelCall<T>(scope: ModelCallScope, call: () => Promise<T>): Promise<T>
   request?(input: unknown, init: unknown, forward: () => Promise<Response>): Promise<Response>
@@ -20,6 +41,15 @@ export interface HostTransport {
 let host: HostTransport | undefined
 export function set(value: HostTransport | undefined) { host = value }
 export function userMessage(input: Parameters<HostTransport["userMessage"]>[0]) { host?.userMessage(input) }
+export function prepareExecution(input: ExecutionAdmission) {
+  if (!host?.admitExecution) throw new Error("Execution host is unavailable")
+  return host.prepareExecution?.(input)
+}
+export function admitExecution(input: ExecutionAdmission) {
+  if (!host?.admitExecution) throw new Error("Execution host is unavailable")
+  return host.admitExecution(input)
+}
+export function finishExecution(attempt: ExecutionAttempt) { host?.finishExecution?.(attempt) }
 export function modelCall<T>(scope: ModelCallScope, call: () => Promise<T>): Promise<T> {
   return host ? host.modelCall(scope, call) : call()
 }

packages/cli/src/question/index.ts

@@ -131,6 +131,7 @@ export interface Interface {
     sessionID: SessionID
     questions: ReadonlyArray<Info>
     tool?: Tool
+    abortSignal?: AbortSignal
   }) => Effect.Effect<ReadonlyArray<Answer>, RejectedError>
   readonly reply: (input: { requestID: QuestionID; answers: ReadonlyArray<Answer> }) => Effect.Effect<void>
   readonly reject: (requestID: QuestionID) => Effect.Effect<void>
@@ -169,8 +170,10 @@ export const layer = Layer.effect(
       sessionID: SessionID
       questions: ReadonlyArray<Info>
       tool?: Tool
+      abortSignal?: AbortSignal
     }) {
       const pending = (yield* InstanceState.get(state)).pending
+      if (input.abortSignal?.aborted) return yield* Effect.fail(new RejectedError())
       const id = QuestionID.ascending()
       log.info("asking", { id, questions: input.questions.length })
 
@@ -182,8 +185,24 @@ export const layer = Layer.effect(
         tool: input.tool,
       })
       pending.set(id, { info, deferred })
+      const signal = input.abortSignal
+      const awaiting = signal
+        ? Effect.raceFirst(
+            Deferred.await(deferred),
+            Effect.callback<ReadonlyArray<Answer>, RejectedError>((resume) => {
+              // Reuse request ownership: a reply already in flight must win over a late abort.
+              const onAbort = () => resume(reject(id).pipe(Effect.andThen(Deferred.await(deferred))))
+              if (signal.aborted) {
+                onAbort()
+                return Effect.void
+              }
+              signal.addEventListener("abort", onAbort, { once: true })
+              return Effect.sync(() => signal.removeEventListener("abort", onAbort))
+            }),
+          )
+        : Deferred.await(deferred)
       return yield* Effect.ensuring(
-        bus.publish(Event.Asked, info).pipe(Effect.andThen(Deferred.await(deferred))),
+        bus.publish(Event.Asked, info).pipe(Effect.andThen(awaiting)),
         Effect.gen(function* () {
           // A caller interrupt (e.g. MCP cancellation) must also withdraw the UI.
           if (!pending.delete(id)) return

packages/cli/src/server/routes/instance/session.ts

@@ -1137,8 +1137,8 @@ export const SessionRoutes = lazy(() =>
     .post(
       "/:sessionID/resume",
       describeRoute({
-        summary: "Resume from a trailing user",
-        description: "Start the next turn from a trailing user message without creating another user message.",
+        summary: "Resume an exact recovery target",
+        description: "Admit a user or assistant recovery target without creating another user message. Omit the target to select the latest candidate.",
         operationId: "session.resumeUser",
         responses: {
           202: { description: "Resume accepted" },
@@ -1158,9 +1158,12 @@ export const SessionRoutes = lazy(() =>
         modelID: z.string().optional(),
       })),
       // Body is optional: empty/{} = resume latest recovery candidate.
-      // userMessageID still accepted for explicit trailing-user targeting (TUI / cascade).
       validator("json", z.object({
         userMessageID: MessageID.zod.optional(),
+        assistantMessageID: MessageID.zod.optional(),
+        attempt: z.object({ requestId: z.string().min(1), attemptId: z.string().min(1) }).optional(),
+      }).refine((body) => !(body.userMessageID && body.assistantMessageID), {
+        message: "userMessageID and assistantMessageID are mutually exclusive",
       }).optional()),
       async (c) => {
         const params = c.req.valid("param")
@@ -1174,29 +1177,6 @@ export const SessionRoutes = lazy(() =>
           c,
           SessionRunState.Service.use((svc) => svc.assertNotBusy(params.sessionID, query.agentID)),
         )
-        // Explicit userMessageID → validate it is still the trailing user.
-        // Omitted → the engine resolves the latest recovery candidate (404 if none).
-        if (body?.userMessageID) {
-          await runRequest(
-            "SessionRoutes.resumeUser.validate",
-            c,
-            SessionPrompt.Service.use((svc) =>
-              svc.recovery({ sessionID: params.sessionID, agentID: query.agentID }).pipe(
-                Effect.flatMap((candidates) =>
-                  candidates.some(
-                    (candidate) => candidate.kind === "parent-user" && candidate.userMessageID === body.userMessageID,
-                  )
-                    ? Effect.void
-                    : Effect.fail(
-                        new NotFoundError({
-                          message: "No resumable trailing user found for message " + body.userMessageID,
-                        }),
-                      ),
-                ),
-              ),
-            ),
-          )
-        }
         // [TP-SR-R21-10] 202 = admission complete (plan + exclusive start), not fire-and-forget.
         // Main agent also cascades subagent recovery (same as /turn/:id/resume).
         const admitted = await runRequest(
@@ -1206,15 +1186,17 @@ export const SessionRoutes = lazy(() =>
             (query.agentID === undefined || query.agentID === "main")
               ? svc.resumeMainCascading({
                   sessionID: params.sessionID,
-                  ...(body?.userMessageID ? { userMessageID: body.userMessageID } : {}),
+                  ...body,
+                  waitForAdmission: true,
                   agentID: query.agentID,
                   task_id: query.task_id,
                   titleLocale: query.titleLocale,
                   ...(query.modelProviderID && query.modelID ? { model: { providerID: query.modelProviderID, modelID: query.modelID } } : {}),
                 })
               : svc.resumeBackground({
                   sessionID: params.sessionID,
-                  ...(body?.userMessageID ? { userMessageID: body.userMessageID } : {}),
+                  ...body,
+                  waitForAdmission: true,
                   agentID: query.agentID,
                   task_id: query.task_id,
                   titleLocale: query.titleLocale,