@lucamattiazzi/task-eta

Framework-agnostic run-duration prediction (ETA) for long-running work — AI workflows, extraction pipelines, batch jobs. It learns from your own history: each finished run contributes a (features → duration) sample, and future runs get a live estimate as their features become known.

• Non-parametric. k-nearest-neighbours over historical samples in scaled feature space — no functional form to fit or re-tune.
• Conservative fallback. Too few similar runs → a quantile baseline over all history for that workflow; no history at all → a prior you provide.
• A range, not a false promise. Every estimate is a [p50, p90] bracket (configurable), because duration is inherently uncertain.
• Two seams, zero lock-in. You provide a DurationStore (where samples live) and, optionally, an EmitEstimate (how you surface the live estimate). The engine itself is pure and dependency-free.
• Optional Mastra layer at @lucamattiazzi/task-eta/mastra.

Install

npm install @lucamattiazzi/task-eta

@mastra/core is an optional peer dependency — only needed if you import the /mastra subpath.

Quick start

import {
  createDurationTracker,
  createInMemoryStore,
} from '@lucamattiazzi/task-eta'

const store = createInMemoryStore() // swap for your own DurationStore

async function runJob(input: { pages: number }) {
  const tracker = createDurationTracker({
    workflow: 'pdf-extraction',
    store,
    prior: { p50Ms: 120_000, p90Ms: 300_000 }, // cold-start guess
    emit: async ({ estimate }) => {
      console.log(`ETA ~${Math.round(estimate.p50Ms / 1000)}s`)
    },
  })

  // Call whenever you learn something — features accumulate across calls.
  await tracker.emitPrediction({ pages: input.pages }, 'starting')

  await doTheWork(input)

  // Record the finished run so future estimates learn from it.
  await tracker.finish()
}

The one-shot estimator is also exported directly:

import { estimateFromSamples } from '@lucamattiazzi/task-eta'

const { p50Ms, p90Ms, tier } = estimateFromSamples(
  candidates,
  { pages: 42 },
  prior,
)

The seams

DurationStore

Where samples live. Implement it against your database. Candidate loads are scoped by workflow and an optional categorical partition (e.g. file type); recordSample persists a finished run.

interface DurationStore {
  loadCandidates(args: {
    workflow: string
    partition?: Record<string, string>
  }): Promise<DurationSample[]>
  recordSample(args: {
    workflow: string
    workflowVersion?: number
    durationS: number
    features: Record<string, number | null | undefined>
    partition?: Record<string, string>
  }): Promise<void>
}

createInMemoryStore() is a ready reference implementation for tests and prototyping. Keep samples minimal and non-identifying — duration, features, and maybe a categorical partition are all the engine uses.

EmitEstimate (optional)

How you surface the live estimate — a websocket broadcast, an SSE frame, a log line. Omit it entirely if you only want to record samples.

type EmitEstimate = (e: {
  estimate: Estimate
  phase: string
  startedAtMs: number
}) => Promise<void> | void

computeProgress({ startedAtMs, p50Ms, p90Ms, nowMs }) turns an estimate into a capped percent + remaining-time, if you want to drive a progress bar.

Mastra integration

Inside a Mastra workflow, stash the tracker in the run's requestContext so one step can start it and another can finish it:

import {
  setDurationTracker,
  getDurationTracker,
} from '@lucamattiazzi/task-eta/mastra'

// in the step that starts the active phase:
const tracker = createDurationTracker({
  workflow: 'my-workflow',
  store,
  prior,
  emit,
})
setDurationTracker(requestContext, tracker)
await tracker.emitPrediction(featuresKnownNow)

// in the completion step:
await getDurationTracker(requestContext)?.finish()

How the estimate is chosen

1. Tier-2 (kNN) — if enough candidates share the query's numeric features, take the k nearest (range-scaled Euclidean distance) and return their duration quantiles.
2. Tier-1 (baseline) — otherwise, quantiles over all candidates for the workflow (+partition).
3. Prior — if history is too thin, the cold-start prior you passed.

Tune via the config/quantiles options (k, thresholds, the bracket).

License

MIT © lucamattiazzi