Reaping & Retention (Advanced)

// sweep.ts
import {
  probeRunExit,
  reapDetachedRuns,
  sandboxReclaimer,
} from '@tanstack/ai-sandbox'
import type { RunRecord } from '@tanstack/ai'
import type { ReapResult, RunExitProbe } from '@tanstack/ai-sandbox'
import {
  durabilityFor,
  driveRun,
  instances,
  locks,
  runs,
  sandbox,
} from './run-setup'

// Did the agent exit? Read the journal tail in the run's sandbox.
// Anything this cannot answer must be 'unknown', never 'finished'.
async function hasFinished(record: RunRecord): Promise<RunExitProbe> {
  if (record.sandboxKey === undefined) return { state: 'unknown' }
  try {
    const instance = await instances.get(record.sandboxKey)
    if (instance === null) return { state: 'unknown' }
    const handle = await sandbox.provider.resume({
      id: instance.providerSandboxId,
    })
    if (handle === null) return { state: 'unknown' }
    return await probeRunExit({ handle, runId: record.runId })
  } catch (error) {
    return { state: 'unknown', error }
  }
}

export function sweepDetachedRuns(): Promise<ReapResult> {
  return reapDetachedRuns({
    runs,
    locks,
    // Must resolve the same log the chat routes wrote.
    durability: durabilityFor,
    hasFinished,
    // The same drive the takeover route uses.
    drive: driveRun,
    now: Date.now(),
    // The only place this time limit is set. In milliseconds.
    detachedRunTtlMs: 30 * 60 * 1000,
    maxRuns: 25,
    reclaim: sandboxReclaimer({ provider: sandbox.provider, instances }),
  })
}

import { sweepDetachedRuns } from './sweep'

let inFlight = false

async function tick(): Promise<void> {
  if (inFlight) return
  inFlight = true
  try {
    const result = await sweepDetachedRuns()
    console.log('reap', result.considered, result.outcomes)
    for (const run of result.runs) {
      if (
        run.outcome === 'failed' ||
        run.outcome === 'unknown' ||
        run.outcome === 'reclaim-failed'
      ) {
        console.warn('reap needs attention', run.runId, run.outcome, run.error)
      }
    }
  } finally {
    inFlight = false
  }
}

setInterval(() => void tick(), 60_000)

// app/api/cron/reap/route.ts
import { sweepDetachedRuns } from '../../../../sweep'

export async function GET(request: Request) {
  const secret = process.env.CRON_SECRET
  if (
    secret === undefined ||
    request.headers.get('authorization') !== `Bearer ${secret}`
  ) {
    return new Response('Unauthorized', { status: 401 })
  }
  const result = await sweepDetachedRuns()
  return Response.json({
    considered: result.considered,
    probed: result.probed,
    outcomes: result.outcomes,
  })
}

{
  "crons": [{ "path": "/api/cron/reap", "schedule": "*/5 * * * *" }]
}

import { sweepDetachedRuns } from './sweep'

// The part of DurableObjectState this needs. In a real Worker, pass
// `state.storage` from your Workers types.
interface AlarmStorage {
  setAlarm: (scheduledTime: number) => Promise<void>
}

export class RunReaper {
  constructor(private readonly storage: AlarmStorage) {}

  async alarm(): Promise<void> {
    try {
      const result = await sweepDetachedRuns()
      console.log('reap', result.considered, result.outcomes)
    } finally {
      // Re-arm in `finally`. An alarm that throws without a new alarm stops
      // reaping forever.
      await this.storage.setAlarm(Date.now() + 60_000)
    }
  }
}