03beba2 · b90643e · f64132f
@@ -16,6 +16,7 @@
import { getRepoById } from "../core/repos.ts";
import { callTool } from "./feature_tools.ts";
import { SCHEDULE_EVERY, type ScheduleEntry } from "./feature.ts";
import { appsRuntime } from "./runtime.ts";
import { during } from "../perf/jobs.ts";
/** The entries whose interval has passed since they last fired (never fired: due). `lastFired` is ms by tool. */
export function due(schedule: readonly ScheduleEntry[], lastFired: ReadonlyMap<string, number>, now: number): ScheduleEntry[] {
@@ -148,7 +149,7 @@
export function startFeatureSchedule(app: App, tickMs: number = TICK_MS): { stop
  const tick = setInterval(() => {
    if (running) return;
    running = true;
    void fireDue(app).finally(() => (running = false));
    void during("feature-schedule", () => fireDue(app)).finally(() => (running = false));
  }, tickMs);
  tick.unref();
  return { stop: () => clearInterval(tick) };
@@ -28,6 +28,7 @@
import { readFileAt } from "./checkout.ts";
import { identityHeaders, type GrantWho } from "./proxy.ts";
import { parsePullAnswer, PULL_BODY_MAX, recordPull, type PullAnswer } from "./pull.ts";
import { appsRuntime } from "./runtime.ts";
import { during } from "../perf/jobs.ts";
/** How long the forge waits for one `/pull` answer, as for a scheduled tool. */
export const PULL_WAIT_MS = 10_000;
@@ -246,7 +247,7 @@
export function startPullSchedule(app: App, tickMs: number = TICK_MS): { stop():
  const tick = setInterval(() => {
    if (running) return;
    running = true;
    void pullDue(app).finally(() => (running = false));
    void during("pull-schedule", () => pullDue(app)).finally(() => (running = false));
  }, tickMs);
  tick.unref();
  return { stop: () => clearInterval(tick) };
@@ -9,6 +9,7 @@
import type { Database } from "bun:sqlite";
import { heapStats } from "bun:jsc";
import type { App } from "../app.ts";
import { quietly } from "../core/write_clock.ts";
import { during } from "../perf/jobs.ts";
export const MEMORY_EVERY_MS = 60_000;
export const HOUR_MS = 3_600_000;
@@ -68,8 +69,8 @@
export function startMemorySampler(app: App, everyMs = MEMORY_EVERY_MS): MemoryS
      app.log.warn("memory sample failed", { error: String(error) });
    }
  };
  tick();
  const timer = setInterval(tick, everyMs);
  during("memory-sampler", () => tick());
  const timer = setInterval(() => during("memory-sampler", () => tick()), everyMs);
  timer.unref?.();
  return { tick, stop: () => clearInterval(timer) };
}
@@ -28,6 +28,7 @@
import { cellCounters, type CellCounter } from "../apps/contain.ts";
import { startLoopLag, type LoopLag } from "../kernel/loop_lag.ts";
import { FINE_MS, record, type Point } from "./ring.ts";
import { readHwmon, readTopology, temperatures } from "./thermal.ts";
import { during } from "../perf/jobs.ts";
export type Paths = { cgroup: string; proc: string; sys: string };
export const SYSTEM: Paths = { cgroup: "/sys/fs/cgroup", proc: "/proc", sys: "/sys" };
@@ -240,8 +241,8 @@
export function startSampler(
      app.log.warn("machine sample failed", { error: String(error) });
    }
  };
  tick();
  const timer = setInterval(tick, options.everyMs ?? FINE_MS);
  during("sampler", tick);
  const timer = setInterval(() => during("sampler", tick), options.everyMs ?? FINE_MS);
  timer.unref?.();
  return {
    tick,
@@ -127,6 +127,7 @@
import { checkStage, type GateStage } from "./gate_reason.ts";
import { extraGates, GATE_DEADLINE_MS, gateQueued, gateRunId, gatesInFlight, watchGates } from "./queue.ts";
import type { JobOptions } from "../cluster/jobs.ts";
import type { CheckDetail } from "../core/runs.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("checks");
@@ -1660,7 +1661,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
  }
  const timer = setInterval(() => {
    if (!stopped) schedule();
    if (!stopped) during("lane", schedule);
  }, TICK_MS);
  // Main is verified after landing on a schedule, when the forge says how often.
  const sweeper =
@@ -1668,12 +1669,14 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
      ? undefined
      : setInterval(() => {
          if (!stopped)
            askVerifications().catch((error) => log.error("verification not asked", { error: String(error) }));
            during("verify-sweep", askVerifications).catch((error) =>
              log.error("verification not asked", { error: String(error) }),
            );
        }, Math.min(VERIFY_SWEEP_MS, options.verifyEveryMs));
  // A gate's slot frees: the pull check held for it starts now, not at the tick.
  const unwatch = watchGates(schedule);
  schedule();
  void catchUp();
  during("lane", schedule);
  void during("lane-catch-up", catchUp);
  /** One gate's run on the lane: the full tier on `commitOid`, traced, cut off by `controller`. */
  async function gateOnce(
@@ -22,6 +22,7 @@
import { id as newId } from "../kernel/id.ts";
import { createLog } from "../kernel/log.ts";
import { audit } from "./audit.ts";
import { probe, type Knock, type Probed } from "./probe.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("probes");
@@ -229,7 +230,7 @@
export function startProbes(
    if (!record) return;
    const timer = setTimeout(() => {
      timers.delete(timer);
      if (!stopped) void one(record);
      if (!stopped) void during("probes", () => one(record));
    }, AFTER_DEPLOY_MS);
    timers.add(timer);
  });
@@ -238,7 +239,7 @@
export function startProbes(
  const scheduleNightly = (): void => {
    if (options.nightly === false) return;
    nightly = setTimeout(() => {
      void track(round()).finally(scheduleNightly);
      void track(during("probes", round)).finally(scheduleNightly);
    }, untilNightly());
  };
  scheduleNightly();
@@ -132,6 +132,7 @@
import {
} from "./runner.ts";
import { STALE_AFTER_MS } from "../runs/lease.ts";
import { abortAgentRun } from "./running.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("crew");
@@ -275,18 +276,20 @@
export async function startCrew(
    // A tick that throws is logged and the next one tries again: thrown
    // from a timer it is an uncaught error that takes the process down.
    try {
      requeueInterrupted(app);
      recheckForDisk(app);
      if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
        lastAutonomy = Date.now();
        track(settleAll(app).catch((error) =>
          log.error("converge sweep failed", { error: String(error) }),
        ));
        track(autonomyTick(app).catch((error) =>
          log.error("autonomy tick failed", { error: String(error) }),
        ));
      }
      pump(app, ready, inFlight);
      during("crew", () => {
        requeueInterrupted(app);
        recheckForDisk(app);
        if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
          lastAutonomy = Date.now();
          track(settleAll(app).catch((error) =>
            log.error("converge sweep failed", { error: String(error) }),
          ));
          track(autonomyTick(app).catch((error) =>
            log.error("autonomy tick failed", { error: String(error) }),
          ));
        }
        pump(app, ready, inFlight);
      });
    } catch (error) {
      log.error("crew tick failed", { error: String(error) });
    }
@@ -301,11 +304,13 @@
export async function startCrew(
    const pruned = pruneRunFiles(app.config.dataDir);
    if (pruned > 0) log.info("removed old run logs", { count: pruned });
  };
  sweepClones();
  const cloneTimer = setInterval(() => !stopped && sweepClones(), 60 * 60_000);
  requeueInterrupted(app);
  sweepAssigned(app);
  pump(app, ready, inFlight);
  during("clone-sweep", sweepClones);
  const cloneTimer = setInterval(() => !stopped && during("clone-sweep", sweepClones), 60 * 60_000);
  during("crew", () => {
    requeueInterrupted(app);
    sweepAssigned(app);
    pump(app, ready, inFlight);
  });
  const handle: CrewHandle = {
    ready: detected.ok,
@@ -48,6 +48,7 @@
import { loadClog } from "../flow/clog_tick.ts";
import { redactSecrets } from "../kernel/log.ts";
import { forgeCeiling } from "../flow/ceiling.ts";
import { askLinesFor } from "../web/studio/asks.ts";
import { during } from "../perf/jobs.ts";
const KEY = "notify_url";
export const MAX_URL = 200;
@@ -585,13 +586,13 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
  // Eight days of silence is no event: the clock wakes the notifier too.
  const clock = setInterval(() => {
    queue = queue.then(async () => {
    queue = queue.then(() => during("notify", async () => {
      try {
        for (const id of people()) await tell(id, false);
      } catch (error) {
        app.log.error("notify failed", { error: String(error) });
      }
    });
    }));
  }, options.clockMs ?? CLOCK_MS);
  clock.unref();
@@ -607,7 +608,10 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
    });
    return queue;
  };
  const pager = setInterval(() => void pageAll(), Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS));
  const pager = setInterval(
    () => void during("notify-page", pageAll),
    Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS),
  );
  pager.unref();
  return {
@@ -35,6 +35,7 @@
import {
  type InstagramClient,
  type Reel,
} from "./instagram.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("prompts");
@@ -101,7 +102,7 @@
export function startPromptRoutine(
  const timer = setInterval(() => {
    if (stopped) return;
    runDue().catch((error) =>
    during("prompts", runDue).catch((error) =>
      log.error("prompt sweep failed", { error: String(error) }),
    );
  }, options.everyMs ?? TICK_MS);
@@ -17,6 +17,7 @@
import { clogStep, type ClogEvent, type ClogState } from "./clog_watch.ts";
import { readOldestApproved } from "./oldest_approved.ts";
import { readStranded } from "./stranded.ts";
import { readVitals } from "./vitals.ts";
import { during } from "../perf/jobs.ts";
const HOUR_MS = 3_600_000;
const log = createLog("clog");
@@ -90,7 +91,7 @@
export function startClogTick(
): () => void {
  const timer = setInterval(() => {
    try {
      tickClog(db, bus);
      during("clog", () => tickClog(db, bus));
    } catch (error) {
      onError(error);
    }
@@ -18,6 +18,7 @@
import { ownRepoPath } from "../kernel/own_repo.ts";
import { KEEPER_CAUSES, pickCause, readKeeperFacts, type KeeperCause, type KeeperMove } from "./keeper_cause.ts";
import { isLow, landedBetween, readRate, type Rate } from "./keeper_rate.ts";
import { oldestWaiting } from "./pile.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("keeper");
const HOUR_MS = 3_600_000;
@@ -177,7 +178,7 @@
export function tickKeeper(app: App, now = new Date(), onLift: () => void = () =
export function startKeeper(app: App, onLift: () => void, everyMs = 60_000, now: () => number = Date.now): () => void {
  const timer = setInterval(() => {
    try {
      tickKeeper(app, new Date(now()), onLift);
      during("keeper", () => tickKeeper(app, new Date(now()), onLift));
    } catch (error) {
      log.warn("keeper tick failed", { error: String(error) });
    }
@@ -495,6 +495,7 @@
import {
  type RouteTiming,
} from "../core/route_timing.ts";
import { captureStore, loopWatch, perfRecord, readLedger } from "../perf/live.ts";
import { during } from "../perf/jobs.ts";
import { classOf, countryOf, repoOf, signatureOf } from "../perf/traffic.ts";
import { createTrafficWriter } from "../perf/traffic_store.ts";
import { startFeatureSchedule } from "../apps/feature_schedule.ts";
@@ -2349,8 +2350,10 @@
export async function serve(app: App): Promise<Servers> {
  // suspension or sign-out another process wrote reaches it too (#846).
  const sweep = setInterval(
    () =>
      sweepStanding(app.db, (error) =>
        app.log.warn("standing sweep did not finish", { error: String(error) }),
      during("standing-sweep", () =>
        sweepStanding(app.db, (error) =>
          app.log.warn("standing sweep did not finish", { error: String(error) }),
        ),
      ),
    app.config.standingSweepMs ?? 5_000,
  );
@@ -71,6 +71,7 @@
import { addCommitLook } from "../checks/looks.ts";
import type { LookVitals } from "../checks/summary.ts";
import { keepAsMain } from "./main_looks.ts";
import { findChromium, isNoBrowser, takeLook, type Capture } from "./capture.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("looks");
@@ -225,8 +226,8 @@
export function startLooks(app: App, options: LooksOptions): LooksHandle {
  // Production missing a picture at its head — one never taken, or a
  // thumb from before thumbs were the page — gets one while it is up.
  // A woken app says nothing on the bus, so this is how it is pictured.
  const firstSweep = setTimeout(sweep, FIRST_SWEEP_MS);
  const sweeping = setInterval(sweep, SWEEP_MS);
  const firstSweep = setTimeout(() => during("looks", sweep), FIRST_SWEEP_MS);
  const sweeping = setInterval(() => during("looks", sweep), SWEEP_MS);
  firstSweep.unref?.();
  sweeping.unref?.();
@@ -101,6 +101,7 @@
import { crewRoot } from "./crew/workdir.ts";
import { createLog } from "./kernel/log.ts";
import { watchStalls } from "./kernel/loop_lag.ts";
import { startPerfWatcher, TICK_MS } from "./perf/duty.ts";
import { during } from "./perf/jobs.ts";
import { stallLine } from "./perf/loop.ts";
import { bootStall, gitRecord, loopWatch, perfRecord } from "./perf/live.ts";
import { timeStatements } from "./kernel/slow_sql.ts";
@@ -350,7 +351,10 @@
const bootSignup = signupFromEnv(process.env.FORGE_SIGNUP);
if (bootSignup.ok && bootSignup.value !== null)
  setSignup(app.db, bootSignup.value, BOOT_ACTOR);
startPerfWatcher(app, perfRecord, Date.now, TICK_MS, gitRecord);
const flagTimer = setInterval(() => flushFlagUses(app.db), 10_000);
const flagTimer = setInterval(
  () => during("flag-uses", () => flushFlagUses(app.db)),
  10_000,
);
await reconcile(app);
// Issues linked under a pillar's umbrella before pillars had labels.
const labelled = syncAllPillars(app);
@@ -589,14 +593,17 @@
const notifier = startNotifier(app, { forgeUrl });
// loses it. runDelivery never throws into the timer.
// Expired sessions and run tokens ride the same tick: cheap DELETEs, never
// worth a worker.
const deliveryTimer = setInterval(() => {
const deliveryTimer = setInterval(() => during("delivery", () => {
  runDelivery(app.db, app.log, app.bus);
  pruneSessions(app.db);
  pruneTokens(app.db);
}, 5_000);
runDelivery(app.db, app.log, app.bus); // sweep anything queued while we were down
pruneSessions(app.db);
pruneTokens(app.db);
}), 5_000);
// Sweep anything queued while we were down.
during("delivery", () => {
  runDelivery(app.db, app.log, app.bus);
  pruneSessions(app.db);
  pruneTokens(app.db);
});
function positive(value: string | undefined, fallback: number): number {
  const parsed = Number(value);
@@ -612,12 +619,13 @@
function positive(value: string | undefined, fallback: number): number {
// the sweep. The operator guard's sweep of stored origins rides it too: one
// origin a tick, the never-checked first (so on boot), every other daily.
const syncTimer = setInterval(() => {
  runSync(app);
  void runAheadReads(app);
  void runGuardSweep(app);
  during("sync", () => runSync(app));
  void during("ahead", () => runAheadReads(app));
  void during("guard-sweep", () => runGuardSweep(app));
}, 60_000);
runSync(app); // one pass on boot: a forge that was down has fetching to do
void runGuardSweep(app);
// One pass on boot: a forge that was down has fetching to do.
during("sync", () => runSync(app));
void during("guard-sweep", () => runGuardSweep(app));
// Which green pulls no longer merge as they stand, so the asks can leave
// them out (`crew/stale_asks.ts`): git answers, off the request path, and an
@@ -658,9 +666,12 @@
const tickAsks = async (): Promise<void> => {
};
// Followed upstreams are read on a slow tick (`actions/upstream_follow.ts`):
// the tip and how far behind land on the follow row; nothing is opened.
const followTimer = setInterval(() => void checkDue(app), 60_000);
const staleTimer = setInterval(() => void tickAsks(), 120_000);
void tickAsks();
const followTimer = setInterval(
  () => void during("follow", () => checkDue(app)),
  60_000,
);
const staleTimer = setInterval(() => void during("asks", tickAsks), 120_000);
void during("asks", tickAsks);
// Every start is counted, and one that follows a process that never said
// goodbye is a crash the log could not record (kernel/boots.ts, #3708).
@@ -701,23 +712,23 @@
void resumeAtBoot(app, undefined, { forgeUrl, runner: checks?.lane }).catch((err
  app.log.warn("resume at boot failed", { error: String(error) }),
);
// An agent's hold that has stood 24 h is released each minute (#6918).
const holdTimer = setInterval(() => {
const holdTimer = setInterval(() => during("holds", () => {
  try {
    expireCrewHolds(app);
  } catch (error) {
    app.log.warn("hold expiry failed", { error: String(error) });
  }
}, 60_000);
}), 60_000);
// The weekly model review (#7815): settles last week's changes, then proposes this week's; it runs once a week whatever the tick.
const modelReviewTimer = setInterval(() => {
const modelReviewTimer = setInterval(() => during("model-review", () => {
  try {
    runModelReview(app);
  } catch (error) {
    app.log.warn("model review failed", { error: String(error) });
  }
}, 3_600_000);
}), 3_600_000);
// The week is the sprint (#8704): at each weekly reset the forge writes its retrospective, unasked.
const weekTimer = setInterval(() => {
const weekTimer = setInterval(() => during("week", () => {
  try {
    const repos = plannedRepoIds(app.db).flatMap((id) => getRepoById(app.db, id) ?? []);
    for (const repo of tickWeeks(app.db, repos, new Date()))
@@ -725,18 +736,18 @@
const weekTimer = setInterval(() => {
  } catch (error) {
    app.log.warn("week retrospective failed", { error: String(error) });
  }
}, 3_600_000);
}), 3_600_000);
// Code scanning's nightly pass (#9593): each hour, the repositories whose last pass is a day old get secrets, advisories and static analysis over their default branch.
const nightlyScanTimer = setInterval(() => {
  void (async () => {
  void during("nightly-scan", async () => {
    for (const id of nightlyDue(app.db, new Date(Date.now() - 24 * 3_600_000).toISOString(), 25)) {
      const repo = getRepoById(app.db, id);
      if (repo !== undefined) await scanNightly(app, repo);
    }
  })().catch((error) => app.log.warn("nightly scan failed", { error: String(error) }));
  }).catch((error) => app.log.warn("nightly scan failed", { error: String(error) }));
}, 3_600_000);
const processTimer = setInterval(processTick, 60_000);
processTick();
const processTimer = setInterval(() => during("processes", processTick), 60_000);
during("processes", processTick);
// A new version proves itself against the public URL; a failed one again, 5 min after and backing off to hourly (release/prove.ts).
proveRunning({
  dataDir: app.config.dataDir,
@@ -760,11 +771,14 @@
if (app.config.forgeRepo) {
    read: ownTagsReader(app, app.config.forgeRepo),
  });
  setBehind(watch);
  void watch.tick();
  setInterval(() => void watch.tick(), CHECK_EVERY_MS);
  void during("release-behind", () => watch.tick());
  setInterval(
    () => void during("release-behind", () => watch.tick()),
    CHECK_EVERY_MS,
  );
  // The no-release alarm reads what this keeps of the running commit (#7638).
  void readLandings(app);
  setInterval(() => void readLandings(app), 60_000);
  void during("landings", () => readLandings(app));
  setInterval(() => void during("landings", () => readLandings(app)), 60_000);
}
// The release keeper (#7557): cut on cadence, promote, verify under page load, roll back to the named
// known-good image and hold. Off unless FORGE_RELEASE_KEEPER=on in a cluster (keeper/wire.ts).
@@ -786,9 +800,12 @@
if (keeperOn.on) {
          said = line;
        })
        .catch((error) => app.log.warn("release keeper step failed", { error: String(error) }));
    step();
    setInterval(step, 60_000);
    setInterval(() => void rollProbe(ports).catch(() => {}), 1000);
    during("release-keeper", step);
    setInterval(() => during("release-keeper", step), 60_000);
    setInterval(
      () => void during("roll-probe", () => rollProbe(ports)).catch(() => {}),
      1000,
    );
  }
} else app.log.info("the release keeper is off", { why: keeperOn.why });
// Only an agent is on a crew (the `repo_crew_agents_only` trigger refuses anyone else). A person
@@ -825,7 +842,7 @@
if (ssh)
// failed; checks are queued again while their commit heads an open pull;
// agent runs are the crew's to queue again, unless there is no crew here to
// do it.
const leaseTimer = setInterval(() => {
const leaseTimer = setInterval(() => during("lease", () => {
  failAbandonedRuns(
    app.db,
    STALE_AFTER_MS,
@@ -834,7 +851,7 @@
const leaseTimer = setInterval(() => {
  requeueAbandonedChecks(app, STALE_AFTER_MS);
  // A grant whose run is no longer running is a crash's leftover: close it.
  releaseOrphans(app.db);
}, 30_000);
}), 30_000);
// A release that read healthy beside this process takes over (#1663),
// and this one lets go of the port before it drains (#2040).
@@ -21,6 +21,7 @@
import { proposeQuarantine } from "./propose.ts";
import { type Ran, runJob } from "./run.ts";
import { lastSlot, owed, untilNext } from "./slots.ts";
import { readLast, writeLast } from "./stamps.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("nightly");
@@ -132,7 +133,7 @@
export function startNightly(
  const waitRadar = (): void => {
    const timer = setTimeout(() => {
      timers.delete(timer);
      void runRadarNight().finally(() => !abort.signal.aborted && waitRadar());
      void during("nightly", runRadarNight).finally(() => !abort.signal.aborted && waitRadar());
    }, untilNext(RADAR_SLOT, now()) + 1_000);
    timers.add(timer);
  };
@@ -142,7 +143,7 @@
export function startNightly(
  const wait = (job: Job): void => {
    const timer = setTimeout(() => {
      timers.delete(timer);
      void run(job.name).finally(() => !abort.signal.aborted && wait(job));
      void during("nightly", () => run(job.name)).finally(() => !abort.signal.aborted && wait(job));
    // A second past the slot: a timer that fires early would read yesterday's.
    }, untilNext(job.slot, now()) + 1_000);
    timers.add(timer);
@@ -35,6 +35,7 @@
import { vettedEnv } from "./git-mirror.ts";
import { hookSecret, hookUrl } from "./hook-secret.ts";
import { verifySignature } from "../core/webhooks.ts";
import type { Provider, RemoteIssue } from "./types.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("mirror-pulls");
@@ -335,7 +336,7 @@
export function startMirrorPulls(app: App): () => void {
      busy = false;
    });
  };
  const timer = setInterval(tick, MIRROR_PULLS_EVERY_MS);
  tick();
  const timer = setInterval(() => during("mirror-pulls", tick), MIRROR_PULLS_EVERY_MS);
  during("mirror-pulls", tick);
  return () => clearInterval(timer);
}
@@ -16,6 +16,7 @@
import { addLabels, kindLabels } from "../hunt/label.ts";
import { createGitRecord } from "./git.ts";
import type { PerfRecord } from "./record.ts";
import { evaluate, evaluateGit, fresh } from "./watch.ts";
import { during } from "./jobs.ts";
export const TICK_MS = 60_000;
@@ -46,7 +47,7 @@
export function startPerfWatcher(
      app.log.warn("performance watcher failed", { error: String(error) });
    }
  };
  const timer = setInterval(tick, everyMs);
  const timer = setInterval(() => during("perf-watcher", tick), everyMs);
  timer.unref();
  return () => clearInterval(timer);
}
@@ -0,0 +1,87 @@
/**
 * Background jobs by name (#10834): a stall no request ran across is counted
 * against the timer, scheduler tick or sweep that was running (`loop.ts`).
 *
 * `during(name, fn)` marks `name` as blocking from the call until `fn`
 * returns: that synchronous slice is the only time it is sure to hold the
 * loop. When `fn` returns a promise, the job is then marked awaiting until
 * the promise settles; awaiting holds nothing, so an awaiting job is named
 * for a stall only when no job's synchronous slice ran across it — then one
 * of its continuations is the likeliest blocker, and the name is a lead, not
 * a verdict. Each call costs a map write and a ring slot or two. The clock
 * is `performance.now`, the stall watcher's.
 */
/** Finished slices kept to match a stall against: a tick a second fills it in about two minutes. */
const KEPT = 256;
export interface JobSpan {
  readonly name: string;
  readonly start: number;
  readonly end: number;
  /** True for the time after `fn` returned a promise and before it settled. */
  readonly awaiting: boolean;
}
export interface JobWatch {
  /** Run `fn` marked as job `name`; returns what it returns, and throws what it throws. */
  during<T>(name: string, fn: () => T): T;
  /**
   * The jobs whose synchronous slice ran across `[from, to]`, longest first,
   * each once; when there are none, the jobs awaiting across it.
   */
  across(from: number, to: number): string[];
}
export function createJobWatch(now: () => number = () => performance.now()): JobWatch {
  const awaiting = new Map<number, { name: string; start: number }>();
  const ring: (JobSpan | undefined)[] = new Array(KEPT);
  let next = 0;
  let id = 0;
  const keep = (span: JobSpan): void => {
    ring[next] = span;
    next = (next + 1) % KEPT;
  };
  const settle = (key: number): void => {
    const job = awaiting.get(key);
    if (job === undefined) return;
    awaiting.delete(key);
      
        sentinelcrew
        1d ago
        outdated
      
      A promise counts as running until it settles, so a job that is only awaiting I/O is blamed for stalls it did not cause. Narrow it to the synchronous slices, or add a test that pins and documents the limit.
    
    keep({ name: job.name, start: job.start, end: now(), awaiting: true });
  };
  const names = (spans: JobSpan[]): string[] => [
    ...new Set(spans.sort((a, b) => b.end - b.start - (a.end - a.start)).map((s) => s.name)),
  ];
  return {
    during(name, fn) {
      const start = now();
      let result: ReturnType<typeof fn>;
      try {
        result = fn();
      } finally {
        keep({ name, start, end: now(), awaiting: false });
      }
      if (result instanceof Promise) {
        const key = ++id;
        awaiting.set(key, { name, start: now() });
        result.then(() => settle(key), () => settle(key));
      }
      return result;
    },
    across(from, to) {
      const overlapping = ring.filter(
        (s): s is JobSpan => s !== undefined && s.end >= from && s.start <= to,
      );
      const blocking = overlapping.filter((s) => !s.awaiting);
      if (blocking.length > 0) return names(blocking);
      const t = now();
      for (const r of awaiting.values())
        if (r.start <= to) overlapping.push({ name: r.name, start: r.start, end: t, awaiting: true });
      return names(overlapping);
    },
  };
}
/** The process's one job watch, read by the loop watch in `live.ts`. */
export const jobWatch = createJobWatch();
/** Run `fn` as background job `name` on the process's job watch. */
export const during = <T>(name: string, fn: () => T): T => jobWatch.during(name, fn);
@@ -2,13 +2,15 @@
import { createBootStall } from "./boot_stall.ts";
import { createGitRecord } from "./git.ts";
import { jobWatch } from "./jobs.ts";
import { createLoopWatch } from "./loop.ts";
import { createCaptureStore } from "./capture.ts";
import { createReadLedger } from "./reads.ts";
import { createPerfRecord } from "./record.ts";
export const perfRecord = createPerfRecord();
export const loopWatch = createLoopWatch();
/** A stall no request ran across names the background jobs that did (#10834). */
export const loopWatch = createLoopWatch(undefined, jobWatch.across);
export const gitRecord = createGitRecord();
/** The worst stall in this process's first five minutes, read by the release keeper (#8848). */
export const bootStall = createBootStall();
@@ -10,6 +10,10 @@
 * full against every route that ran across it — the same reading as the log
 * lines, so the sum over routes can exceed the wall time — and against
 * `NO_REQUEST` when none did.
 *
 * A stall no request ran across is counted against the background jobs that
 * did (#10834, `jobs.ts`), each as `job:<name>`; `NO_REQUEST` is left for a
 * stall nothing named ran across.
 */
export const LIMIT_MS = 250;
@@ -20,6 +24,10 @@
export const HOUR_MS = 3_600_000;
const HOUR_KEPT = 14_400;
/** The route a stall is counted against when no request ran across it. */
export const NO_REQUEST = "none";
/** The prefix a background job's name is tallied under, apart from routes. */
export const JOB_PREFIX = "job:";
/** The jobs that ran across `[from, to]` (`JobWatch.across`). */
export type JobsAcross = (from: number, to: number) => readonly string[];
export interface RouteStall {
  readonly route: string;
@@ -38,6 +46,8 @@
export interface Stall {
  readonly at: number;
  readonly lateMs: number;
  readonly routes: readonly string[];
  /** The background jobs running across it; read only when no request was. */
  readonly jobs: readonly string[];
}
export interface LoopWatch {
@@ -49,10 +59,14 @@
export interface LoopWatch {
  tally(now: number): StallTally;
}
/** What a stall is counted against: its routes, else its jobs, else `NO_REQUEST`. */
const blamed = (s: Stall): readonly string[] =>
  s.routes.length > 0 ? s.routes : s.jobs.length > 0 ? s.jobs.map((j) => JOB_PREFIX + j) : [NO_REQUEST];
function byRoute(stalls: Iterable<Stall>): RouteStall[] {
  const sums = new Map<string, { ms: number; count: number }>();
  for (const s of stalls)
    for (const route of s.routes.length === 0 ? [NO_REQUEST] : s.routes) {
    for (const route of blamed(s)) {
      const t = sums.get(route) ?? { ms: 0, count: 0 };
      t.ms += s.lateMs;
      t.count += 1;
@@ -61,7 +75,7 @@
function byRoute(stalls: Iterable<Stall>): RouteStall[] {
  return [...sums].map(([route, t]) => ({ route, ms: t.ms, count: t.count }));
}
export function createLoopWatch(limitMs = LIMIT_MS): LoopWatch {
export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = () => []): LoopWatch {
  const recent: { pattern: string; start: number; end: number }[] = [];
  const totals = new Map<string, { ms: number; count: number }>();
  const hour: Stall[] = [];
@@ -81,7 +95,8 @@
export function createLoopWatch(limitMs = LIMIT_MS): LoopWatch {
            .map((r) => r.pattern),
        ),
      ];
      const stall = { at, lateMs, routes };
      const jobs = routes.length === 0 ? jobsAcross(from, at) : [];
      const stall = { at, lateMs, routes, jobs };
      for (const r of byRoute([stall])) {
        const t = totals.get(r.route) ?? { ms: 0, count: 0 };
        t.ms += r.ms;
@@ -104,4 +119,6 @@
export function createLoopWatch(limitMs = LIMIT_MS): LoopWatch {
/** The stall as a finding line. */
export const stallLine = (s: Stall): string =>
  `event loop blocked ${Math.round(s.lateMs)} ms; running then: ${s.routes.length === 0 ? "no request" : s.routes.join(", ")}`;
  `event loop blocked ${Math.round(s.lateMs)} ms; running then: ${
    s.routes.length > 0 ? s.routes.join(", ") : s.jobs.length > 0 ? `no request; jobs ${s.jobs.join(", ")}` : "no request"
  }`;
@@ -35,6 +35,7 @@
import { lastMirror, mirrorPass, mirrorTarget, MIRROR_EVERY_MS, recordMirror, ty
import { isOffBox } from "./off_box.ts";
import { noteOffBox, recordSnapshotOutcome } from "./outcome.ts";
import { s3FromEnv, type Store } from "./store.ts";
import { during } from "../perf/jobs.ts";
const log = createLog("snapshots");
@@ -379,7 +380,7 @@
export function startSnapshots(
    if (timer !== undefined) clearTimeout(timer);
    timer = undefined;
    if (options.nightly === false || stopped) return;
    timer = setTimeout(() => void take().catch(() => {}), ms);
    timer = setTimeout(() => void during("snapshot", take).catch(() => {}), ms);
  }
  arm(untilNightly());
@@ -398,7 +399,7 @@
export function startSnapshots(
  const scheduleDrill = (): void => {
    if (options.nightly === false || stopped) return;
    drillTimer = setTimeout(() => {
      runIfDue().finally(scheduleDrill);
      during("drill", () => runIfDue()).finally(scheduleDrill);
    }, untilDrill());
  };
  // While the newest drill is red, look again when it is owed again.
@@ -409,7 +410,7 @@
export function startSnapshots(
    const newest = lastDrill(app.db);
    if (newest?.verdict !== "red") return;
    const wait = Math.max(catchUpMs, Date.parse(newest.at) + redRetryMs - Date.now());
    redRetry = setTimeout(() => void runIfDue().finally(armRedRetry), wait);
    redRetry = setTimeout(() => void during("drill", () => runIfDue()).finally(armRedRetry), wait);
  };
  // Owed and not yet kept: run it once. A restart after a green drill finds it
  // kept; one after a red drill runs it again, once it is past the boot floor.
@@ -423,7 +424,9 @@
export function startSnapshots(
  // shortly after boot; a red one is looked at again until one passes.
  if (options.nightly !== false) {
    catchUp = setTimeout(
      () => void runIfDue(options.bootRedFloorMs ?? BOOT_RED_FLOOR_MS).finally(armRedRetry),
      () =>
        void during("drill", () => runIfDue(options.bootRedFloorMs ?? BOOT_RED_FLOOR_MS))
          .finally(armRedRetry),
      catchUpMs,
    );
  }
@@ -434,7 +437,7 @@
export function startSnapshots(
  const scheduleMirror = (first: boolean): void => {
    if (options.nightly === false || stopped) return;
    mirrorTimer = setTimeout(() => {
      mirrorNow()
      during("snapshot-mirror", mirrorNow)
        .catch(() => {})
        .finally(() => scheduleMirror(false));
    }, first ? Math.min(30_000, everyMs) : everyMs);
@@ -447,7 +450,7 @@
export function startSnapshots(
  const scheduleCatchUp = (ms: number): void => {
    if (options.nightly === false || stopped) return;
    catchUpSnapshot = setTimeout(() => {
      takeIfDue()
      during("snapshot", takeIfDue)
        .catch(() => {})
        .finally(() => {
          bootLookAt = undefined;
@@ -26,6 +26,7 @@
import { upstreamState } from "./upstream.ts";
import { progress } from "../core/next-steps.ts";
import { running } from "../kernel/version.ts";
import { forgeCeiling } from "../flow/ceiling.ts";
import { during } from "../perf/jobs.ts";
export const SIGNATURE_HEADER = "x-forge-signature";
/** A child reports this often; one silent for three beats reads as quiet. */
@@ -323,7 +324,7 @@
export async function sendBeat(
/** Report now and then every BEAT_MS; returns the timer to clear. */
export function startHeartbeat(app: App): ReturnType<typeof setInterval> {
  void sendBeat(app);
  return setInterval(() => void sendBeat(app), BEAT_MS);
  return setInterval(() => void during("heartbeat", () => sendBeat(app)), BEAT_MS);
}
export type Standing = "reporting" | "quiet" | "unheard";
@@ -25,6 +25,7 @@
import type { App } from "../app.ts";
import { isSlug, slugFor } from "../core/apps.ts";
import type { FiredAlarm } from "../flow/alarms.ts";
import { childPublicUrl, parentOrigin } from "./placement.ts";
import { during } from "../perf/jobs.ts";
export const APP_HEADER = "x-forge-app";
@@ -283,8 +284,8 @@
export function startReachWatch(
  everyMs = REACH_EVERY_MS,
  asleep: (name: string) => boolean = () => false,
): ReturnType<typeof setInterval> {
  const first = setTimeout(() => void sweepReach(app, asleep), 30_000);
  const timer = setInterval(() => void sweepReach(app, asleep), everyMs);
  const first = setTimeout(() => void during("reach", () => sweepReach(app, asleep)), 30_000);
  const timer = setInterval(() => void during("reach", () => sweepReach(app, asleep)), everyMs);
  app.closing.signal.addEventListener("abort", () => clearTimeout(first), { once: true });
  return timer;
}
@@ -23,6 +23,7 @@
import { basicHeader, gitAuthEnv, redact, secretsIn } from "../origins/credentia
import { forgeLinkOf, originOf } from "./chain.ts";
import { parentCredential } from "./parent_credential.ts";
import { newestCommits, takeFromParent, type Taken } from "./take.ts";
import { during } from "../perf/jobs.ts";
export const CHECK_MS = 60 * 60_000;
export const FETCH_TIMEOUT_MS = 60_000;
@@ -211,7 +212,7 @@
function tellOwner(app: App, repo: Repo, fork: string, behind: number, commits =
/** Check at boot and every hour; returns the timer to clear. */
export function startUpstreamWatch(app: App): ReturnType<typeof setInterval> {
  void checkUpstream(app);
  return setInterval(() => void checkUpstream(app), CHECK_MS);
  return setInterval(() => void during("upstream", () => checkUpstream(app)), CHECK_MS);
}
/** The state in one plain sentence, escaped by whoever renders it. */
@@ -0,0 +1,103 @@
/** A stall no request ran across names the background job that was running (#10834). */
import { expect, test } from "bun:test";
import { createJobWatch, during } from "../src/perf/jobs.ts";
import { loopWatch } from "../src/perf/live.ts";
import { createLoopWatch, NO_REQUEST, stallLine } from "../src/perf/loop.ts";
const block = (ms: number): void => {
  const until = performance.now() + ms;
  while (performance.now() < until) {}
};
test("a job that blocks the loop 300 ms is named in the tally and the stall line", () => {
  const jobs = createJobWatch();
  const w = createLoopWatch(250, jobs.across);
  jobs.during("sweep", () => block(300));
  const at = performance.now();
  const stall = w.stalled(310, at)!;
  expect(stall.routes).toEqual([]);
  expect(stall.jobs).toEqual(["sweep"]);
  expect(stallLine(stall)).toContain("jobs sweep");
  const tally = w.tally(at);
  expect(tally.total).toEqual([{ route: "job:sweep", ms: 310, count: 1 }]);
  expect(tally.hour.map((r) => r.route)).toEqual(["job:sweep"]);
});
test("an async job is named while it awaits only when no job blocked across the stall", async () => {
  let t = 0;
  const jobs = createJobWatch(() => t);
  let done!: () => void;
  const body = jobs.during("lane", () => {
    t = 10; // the synchronous slice, up to the first await
    return new Promise<void>((r) => (done = r));
  });
  t = 1_000;
  expect(jobs.across(0, 5)).toEqual(["lane"]);
  // Nothing blocked across [500, 1000]: the awaiting lane is the lead.
  expect(jobs.across(500, 1_000)).toEqual(["lane"]);
  // A job that blocks while the lane awaits is the one named; the lane is not.
  jobs.during("sweep", () => { t = 1_400; });
  expect(jobs.across(1_100, 1_400)).toEqual(["sweep"]);
  t = 2_000;
  done();
  await body;
  t = 5_000;
  expect(jobs.across(1_500, 5_000)).toEqual(["lane"]);
  expect(jobs.across(2_001, 6_000)).toEqual([]);
});
test("a loop stall while a job awaits and another blocks is tallied to the blocker only", async () => {
  const jobs = createJobWatch();
  const w = createLoopWatch(250, jobs.across);
  let done!: () => void;
  const lane = jobs.during("lane", () => new Promise<void>((r) => (done = r)));
  const from = performance.now();
  jobs.during("sweep", () => block(300));
  const at = performance.now();
  const stall = w.stalled(at - from, at)!;
  expect(stall.jobs).toEqual(["sweep"]);
  expect(w.tally(at).total.map((r) => r.route)).toEqual(["job:sweep"]);
  done();
  await lane;
});
test("the process's loop watch reads the process's job watch", () => {
  const before = loopWatch.tally(performance.now()).total.find((r) => r.route === "job:wired-test")?.count ?? 0;
  during("wired-test", () => block(300));
  const at = performance.now();
  loopWatch.stalled(300, at);
  const after = loopWatch.tally(at).total.find((r) => r.route === "job:wired-test");
  expect(after?.count).toBe(before + 1);
});
test("a job that throws still ends, and the throw reaches the caller", () => {
  let t = 0;
  const jobs = createJobWatch(() => t);
  expect(() => jobs.during("bad", () => { t = 10; throw new Error("no"); })).toThrow("no");
  expect(jobs.across(20, 30)).toEqual([]);
  expect(jobs.across(5, 30)).toEqual(["bad"]);
});
test("a request across the stall is blamed, not the job", () => {
  let t = 0;
  const jobs = createJobWatch(() => t);
  const w = createLoopWatch(250, jobs.across);
  jobs.during("sweep", () => { t = 400; });
  w.finished("/boards", 0, 400);
  const stall = w.stalled(400, 400)!;
  expect(stall.routes).toEqual(["/boards"]);
  expect(w.tally(400).total.map((r) => r.route)).toEqual(["/boards"]);
});
test("an idle stall with no request and no job is still none", () => {
  let t = 0;
  const jobs = createJobWatch(() => t);
  const w = createLoopWatch(250, jobs.across);
  jobs.during("early", () => { t = 10; });
  t = 5_000;
  const stall = w.stalled(300, 5_000)!;
  expect(stall.jobs).toEqual([]);
  expect(stallLine(stall)).toBe("event loop blocked 300 ms; running then: no request");
  expect(w.tally(5_000).total).toEqual([{ route: NO_REQUEST, ms: 300, count: 1 }]);
});
		@@ -16,6 +16,7 @@ import { getRepoById } from "../core/repos.ts";
16	16	import { callTool } from "./feature_tools.ts";
17	17	import { SCHEDULE_EVERY, type ScheduleEntry } from "./feature.ts";
18	18	import { appsRuntime } from "./runtime.ts";
	19	import { during } from "../perf/jobs.ts";
19	20
20	21	/** The entries whose interval has passed since they last fired (never fired: due). `lastFired` is ms by tool. */
21	22	export function due(schedule: readonly ScheduleEntry[], lastFired: ReadonlyMap<string, number>, now: number): ScheduleEntry[] {
		@@ -148,7 +149,7 @@ export function startFeatureSchedule(app: App, tickMs: number = TICK_MS): { stop
148	149	const tick = setInterval(() => {
149	150	if (running) return;
150	151	running = true;
151		void fireDue(app).finally(() => (running = false));
	152	void during("feature-schedule", () => fireDue(app)).finally(() => (running = false));
152	153	}, tickMs);
153	154	tick.unref();
154	155	return { stop: () => clearInterval(tick) };
		@@ -28,6 +28,7 @@ import { readFileAt } from "./checkout.ts";
28	28	import { identityHeaders, type GrantWho } from "./proxy.ts";
29	29	import { parsePullAnswer, PULL_BODY_MAX, recordPull, type PullAnswer } from "./pull.ts";
30	30	import { appsRuntime } from "./runtime.ts";
	31	import { during } from "../perf/jobs.ts";
31	32
32	33	/** How long the forge waits for one `/pull` answer, as for a scheduled tool. */
33	34	export const PULL_WAIT_MS = 10_000;
		@@ -246,7 +247,7 @@ export function startPullSchedule(app: App, tickMs: number = TICK_MS): { stop():
246	247	const tick = setInterval(() => {
247	248	if (running) return;
248	249	running = true;
249		void pullDue(app).finally(() => (running = false));
	250	void during("pull-schedule", () => pullDue(app)).finally(() => (running = false));
250	251	}, tickMs);
251	252	tick.unref();
252	253	return { stop: () => clearInterval(tick) };
		@@ -9,6 +9,7 @@ import type { Database } from "bun:sqlite";
9	9	import { heapStats } from "bun:jsc";
10	10	import type { App } from "../app.ts";
11	11	import { quietly } from "../core/write_clock.ts";
	12	import { during } from "../perf/jobs.ts";
12	13
13	14	export const MEMORY_EVERY_MS = 60_000;
14	15	export const HOUR_MS = 3_600_000;
		@@ -68,8 +69,8 @@ export function startMemorySampler(app: App, everyMs = MEMORY_EVERY_MS): MemoryS
68	69	app.log.warn("memory sample failed", { error: String(error) });
69	70	}
70	71	};
71		tick();
72		const timer = setInterval(tick, everyMs);
	72	during("memory-sampler", () => tick());
	73	const timer = setInterval(() => during("memory-sampler", () => tick()), everyMs);
73	74	timer.unref?.();
74	75	return { tick, stop: () => clearInterval(timer) };
75	76	}
		@@ -28,6 +28,7 @@ import { cellCounters, type CellCounter } from "../apps/contain.ts";
28	28	import { startLoopLag, type LoopLag } from "../kernel/loop_lag.ts";
29	29	import { FINE_MS, record, type Point } from "./ring.ts";
30	30	import { readHwmon, readTopology, temperatures } from "./thermal.ts";
	31	import { during } from "../perf/jobs.ts";
31	32
32	33	export type Paths = { cgroup: string; proc: string; sys: string };
33	34	export const SYSTEM: Paths = { cgroup: "/sys/fs/cgroup", proc: "/proc", sys: "/sys" };
		@@ -240,8 +241,8 @@ export function startSampler(
240	241	app.log.warn("machine sample failed", { error: String(error) });
241	242	}
242	243	};
243		tick();
244		const timer = setInterval(tick, options.everyMs ?? FINE_MS);
	244	during("sampler", tick);
	245	const timer = setInterval(() => during("sampler", tick), options.everyMs ?? FINE_MS);
245	246	timer.unref?.();
246	247	return {
247	248	tick,
		@@ -127,6 +127,7 @@ import { checkStage, type GateStage } from "./gate_reason.ts";
127	127	import { extraGates, GATE_DEADLINE_MS, gateQueued, gateRunId, gatesInFlight, watchGates } from "./queue.ts";
128	128	import type { JobOptions } from "../cluster/jobs.ts";
129	129	import type { CheckDetail } from "../core/runs.ts";
	130	import { during } from "../perf/jobs.ts";
130	131
131	132	const log = createLog("checks");
132	133
		@@ -1660,7 +1661,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1660	1661	}
1661	1662
1662	1663	const timer = setInterval(() => {
1663		if (!stopped) schedule();
	1664	if (!stopped) during("lane", schedule);
1664	1665	}, TICK_MS);
1665	1666	// Main is verified after landing on a schedule, when the forge says how often.
1666	1667	const sweeper =
		@@ -1668,12 +1669,14 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1668	1669	? undefined
1669	1670	: setInterval(() => {
1670	1671	if (!stopped)
1671		askVerifications().catch((error) => log.error("verification not asked", { error: String(error) }));
	1672	during("verify-sweep", askVerifications).catch((error) =>
	1673	log.error("verification not asked", { error: String(error) }),
	1674	);
1672	1675	}, Math.min(VERIFY_SWEEP_MS, options.verifyEveryMs));
1673	1676	// A gate's slot frees: the pull check held for it starts now, not at the tick.
1674	1677	const unwatch = watchGates(schedule);
1675		schedule();
1676		void catchUp();
	1678	during("lane", schedule);
	1679	void during("lane-catch-up", catchUp);
1677	1680
1678	1681	/** One gate's run on the lane: the full tier on `commitOid`, traced, cut off by `controller`. */
1679	1682	async function gateOnce(
		@@ -22,6 +22,7 @@ import { id as newId } from "../kernel/id.ts";
22	22	import { createLog } from "../kernel/log.ts";
23	23	import { audit } from "./audit.ts";
24	24	import { probe, type Knock, type Probed } from "./probe.ts";
	25	import { during } from "../perf/jobs.ts";
25	26
26	27	const log = createLog("probes");
27	28
		@@ -229,7 +230,7 @@ export function startProbes(
229	230	if (!record) return;
230	231	const timer = setTimeout(() => {
231	232	timers.delete(timer);
232		if (!stopped) void one(record);
	233	if (!stopped) void during("probes", () => one(record));
233	234	}, AFTER_DEPLOY_MS);
234	235	timers.add(timer);
235	236	});
		@@ -238,7 +239,7 @@ export function startProbes(
238	239	const scheduleNightly = (): void => {
239	240	if (options.nightly === false) return;
240	241	nightly = setTimeout(() => {
241		void track(round()).finally(scheduleNightly);
	242	void track(during("probes", round)).finally(scheduleNightly);
242	243	}, untilNightly());
243	244	};
244	245	scheduleNightly();
		@@ -132,6 +132,7 @@ import {
132	132	} from "./runner.ts";
133	133	import { STALE_AFTER_MS } from "../runs/lease.ts";
134	134	import { abortAgentRun } from "./running.ts";
	135	import { during } from "../perf/jobs.ts";
135	136
136	137	const log = createLog("crew");
137	138
		@@ -275,18 +276,20 @@ export async function startCrew(
275	276	// A tick that throws is logged and the next one tries again: thrown
276	277	// from a timer it is an uncaught error that takes the process down.
277	278	try {
278		requeueInterrupted(app);
279		recheckForDisk(app);
280		if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
281		lastAutonomy = Date.now();
282		track(settleAll(app).catch((error) =>
283		log.error("converge sweep failed", { error: String(error) }),
284		));
285		track(autonomyTick(app).catch((error) =>
286		log.error("autonomy tick failed", { error: String(error) }),
287		));
288		}
289		pump(app, ready, inFlight);
	279	during("crew", () => {
	280	requeueInterrupted(app);
	281	recheckForDisk(app);
	282	if (Date.now() - lastAutonomy >= AUTONOMY_TICK_MS) {
	283	lastAutonomy = Date.now();
	284	track(settleAll(app).catch((error) =>
	285	log.error("converge sweep failed", { error: String(error) }),
	286	));
	287	track(autonomyTick(app).catch((error) =>
	288	log.error("autonomy tick failed", { error: String(error) }),
	289	));
	290	}
	291	pump(app, ready, inFlight);
	292	});
290	293	} catch (error) {
291	294	log.error("crew tick failed", { error: String(error) });
292	295	}
		@@ -301,11 +304,13 @@ export async function startCrew(
301	304	const pruned = pruneRunFiles(app.config.dataDir);
302	305	if (pruned > 0) log.info("removed old run logs", { count: pruned });
303	306	};
304		sweepClones();
305		const cloneTimer = setInterval(() => !stopped && sweepClones(), 60 * 60_000);
306		requeueInterrupted(app);
307		sweepAssigned(app);
308		pump(app, ready, inFlight);
	307	during("clone-sweep", sweepClones);
	308	const cloneTimer = setInterval(() => !stopped && during("clone-sweep", sweepClones), 60 * 60_000);
	309	during("crew", () => {
	310	requeueInterrupted(app);
	311	sweepAssigned(app);
	312	pump(app, ready, inFlight);
	313	});
309	314
310	315	const handle: CrewHandle = {
311	316	ready: detected.ok,
		@@ -48,6 +48,7 @@ import { loadClog } from "../flow/clog_tick.ts";
48	48	import { redactSecrets } from "../kernel/log.ts";
49	49	import { forgeCeiling } from "../flow/ceiling.ts";
50	50	import { askLinesFor } from "../web/studio/asks.ts";
	51	import { during } from "../perf/jobs.ts";
51	52
52	53	const KEY = "notify_url";
53	54	export const MAX_URL = 200;
		@@ -585,13 +586,13 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
585	586
586	587	// Eight days of silence is no event: the clock wakes the notifier too.
587	588	const clock = setInterval(() => {
588		queue = queue.then(async () => {
	589	queue = queue.then(() => during("notify", async () => {
589	590	try {
590	591	for (const id of people()) await tell(id, false);
591	592	} catch (error) {
592	593	app.log.error("notify failed", { error: String(error) });
593	594	}
594		});
	595	}));
595	596	}, options.clockMs ?? CLOCK_MS);
596	597	clock.unref();
597	598
		@@ -607,7 +608,10 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
607	608	});
608	609	return queue;
609	610	};
610		const pager = setInterval(() => void pageAll(), Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS));
	611	const pager = setInterval(
	612	() => void during("notify-page", pageAll),
	613	Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS),
	614	);
611	615	pager.unref();
612	616
613	617	return {
		@@ -35,6 +35,7 @@ import {
35	35	type InstagramClient,
36	36	type Reel,
37	37	} from "./instagram.ts";
	38	import { during } from "../perf/jobs.ts";
38	39
39	40	const log = createLog("prompts");
40	41
		@@ -101,7 +102,7 @@ export function startPromptRoutine(
101	102
102	103	const timer = setInterval(() => {
103	104	if (stopped) return;
104		runDue().catch((error) =>
	105	during("prompts", runDue).catch((error) =>
105	106	log.error("prompt sweep failed", { error: String(error) }),
106	107	);
107	108	}, options.everyMs ?? TICK_MS);
		@@ -17,6 +17,7 @@ import { clogStep, type ClogEvent, type ClogState } from "./clog_watch.ts";
17	17	import { readOldestApproved } from "./oldest_approved.ts";
18	18	import { readStranded } from "./stranded.ts";
19	19	import { readVitals } from "./vitals.ts";
	20	import { during } from "../perf/jobs.ts";
20	21
21	22	const HOUR_MS = 3_600_000;
22	23	const log = createLog("clog");
		@@ -90,7 +91,7 @@ export function startClogTick(
90	91	): () => void {
91	92	const timer = setInterval(() => {
92	93	try {
93		tickClog(db, bus);
	94	during("clog", () => tickClog(db, bus));
94	95	} catch (error) {
95	96	onError(error);
96	97	}