T4 — Interruption-safe scope cleanup

{`
return yield* Effect.gen(function* () {
  yield* Effect.gen(function* () {
    // Reset state cho iteration mới
    ctx.currentText = undefined
    ctx.reasoningMap = {}

    // T2: Consume stream với demux
    const stream = llm.stream(streamInput)
    yield* stream.pipe(
      Stream.tap((event) => handleEvent(event)),
      Stream.takeUntil(() => ctx.needsCompaction),
      Stream.runDrain,
    )

  }).pipe(
    // Hook: khi bị interrupt → mark aborted
    Effect.onInterrupt(() => Effect.gen(function* () {
      aborted = true
      if (!ctx.assistantMessage.error) {
        yield* halt(new DOMException("Aborted", "AbortError"))
      }
    })),

    // Chỉ propagate non-interrupt errors
    Effect.catchCauseIf(
      (cause) => !Cause.hasInterruptsOnly(cause),
      (cause) => Effect.fail(Cause.squash(cause)),
    ),

    // Retry logic (bên trong ensuring)
    Effect.retry(retrySchedule),

    // Catch remaining errors → halt
    Effect.catch(halt),

    // ← CLEANUP LUÔN CHẠY — dù retry, interrupt, error, hay normal
    Effect.ensuring(cleanup()),
  )
})
`}

{`
function* cleanup() {
  // 1. Finalize partial text nếu có
  if (ctx.currentText) {
    yield* session.finalizeText(ctx.currentText)
  }

  // 2. Mark tất cả pending tools là aborted
  for (const [id, call] of Object.entries(ctx.toolcalls)) {
    yield* session.updatePart(call.partID, { status: "aborted" })
    // 3. Resolve Deferred (T3) để không block
    yield* Deferred.succeed(call.done, undefined)
  }

  // 4. Update session status
  yield* status.set(sessionID, { type: "idle" })
}
`}

{`
Effect.onInterrupt(() => Effect.gen(function* () {
  aborted = true  // flag cho code downstream biết
  // Chỉ halt nếu chưa có error khác
  if (!ctx.assistantMessage.error) {
    yield* halt(new DOMException("Aborted", "AbortError"))
  }
}))
// Sau onInterrupt → ensuring(cleanup()) vẫn chạy
`}

{`
// WRONG: không cleanup khi cancel
async function processStream(signal: AbortSignal) {
  for await (const event of stream) {
    if (signal.aborted) return  // exit ngay, không cleanup
    handleEvent(event)
  }
}
// Tool partID vẫn có status: "running" trong DB
// User reload page → vẫn thấy tool "running" từ session cũ
`}

{`
// Tool mở file để đọc
const file = await fs.open(path, "r")
try {
  const content = await file.readFile()
  return content
} catch (err) {
  // Nếu stream bị interrupt ở đây:
  // file.close() không được gọi → fd leak
  throw err
}
// CORRECT: dùng finally
} finally {
  await file.close()  // luôn đóng
}
`}

{`
// WRONG: cleanup chạy sau mỗi retry
pipeline.pipe(
  Effect.ensuring(cleanup()),  // ← chạy sau mỗi attempt
  Effect.retry(schedule),
)

// CORRECT: cleanup chạy sau retry exhausted
pipeline.pipe(
  Effect.retry(schedule),      // retry trong ensuring
  Effect.ensuring(cleanup()),  // ← chỉ chạy 1 lần sau cùng
)
`}

{`
interface StreamContext {
  pendingTools: Map<string, { resolve: () => void }>
  currentText: string
  aborted: boolean
}

async function streamWithCleanup(
  streamFn: (signal: AbortSignal) => AsyncIterable<StreamEvent>,
  onEvent: (event: StreamEvent, ctx: StreamContext) => Promise<void>,
  options = { retries: 2, toolTimeoutMs: 250 }
) {
  const ctx: StreamContext = {
    pendingTools: new Map(),
    currentText: "",
    aborted: false,
  }

  const controller = new AbortController()

  // Expose cancel để caller có thể trigger
  const cancel = () => {
    ctx.aborted = true
    controller.abort()
  }

  let attempt = 0

  while (attempt <= options.retries) {
    try {
      const stream = streamFn(controller.signal)

      for await (const event of stream) {
        if (controller.signal.aborted) break
        await onEvent(event, ctx)
      }

      // Success — exit retry loop
      break

    } catch (err) {
      if (controller.signal.aborted) {
        // User cancel — không retry
        break
      }
      attempt++
      if (attempt > options.retries) throw err
      // Wait before retry
      await new Promise((r) => setTimeout(r, Math.pow(2, attempt) * 1000))

    } finally {
      // CLEANUP — luôn chạy (cancel, error, success)
      await cleanup(ctx, options.toolTimeoutMs)
    }
  }

  return { aborted: ctx.aborted, text: ctx.currentText }
}

async function cleanup(ctx: StreamContext, timeoutMs: number) {
  // 1. Await all pending tools with timeout
  await Promise.all(
    [...ctx.pendingTools.values()].map((d) =>
      Promise.race([
        new Promise<void>((r) => {
          // Replace resolve to track when done
          const orig = d.resolve
          d.resolve = () => { orig(); r() }
        }),
        new Promise<void>((r) => setTimeout(r, timeoutMs)),
      ])
    )
  )

  // 2. Mark remaining tools aborted
  ctx.pendingTools.clear()

  console.log("[cleanup] done — text length:", ctx.currentText.length)
}
`}