Skip to content

Commit a342e11

Browse files
edgeheroclaude
andcommitted
feat(pause): per-folder/repo scheduled pause windows (quiet hours), defer + auto-resume
Add scoped, time-bounded pauses: hold a folder's (local) or repo's (github) runs "between certain times" and resume automatically after — recurring daily, with optional weekday and date-range bounds and a per-window IANA timezone (default UTC, DST-correct via built-in Intl, no new dependency). Enforcement — defer, don't drop (DES-SCOPED-PAUSE-VIA-MOVE-TO-DELAYED) - worker/src/pause-windows.mjs: pure, fs-injectable validator (parsePauseWindows), loader (loadPauseWindows), and the timezone-aware predicate (pauseUntilMs + windowEndAt). Fully unit-testable with injected `now`. - The processor gate is FIRST in the wrapper, before the kill timer and the budget reservation: if the job's scope is in an active window, job.moveToDelayed(end, token) + throw DelayedError. The job keeps its dedup identity, survives restart, reserves NO budget slot, and auto-resumes when BullMQ re-picks it. Consistent with CONST-BUDGET-BEFORE-TOKENS (a deferral is not a job start). - pause-windows.json (PI_PAUSE_WINDOWS_FILE) is boot-loaded fail-loud and live-reloaded via the same dir-watch machinery as triggers.json; a bad edit keeps the last-good windows. Admin control (mirrors the confirm-gated triggers CRUD) - read/writePauseWindows (shared validator, atomic, fail-closed); resolvePaths gains pauseWindowsPath. Tools: dispatch_pauses (read) + dispatch_pause_add/_delete (confirm-gated, fail-closed without an operator). Operator `w` key -> add/delete dialogs. A PAUSES panel section marks each window ● paused (with a resume countdown) or ○ open. The operate-pi-dispatch skill documents the gate. Specs: REQ-SCOPED-PAUSE-WINDOWS, DES-SCOPED-PAUSE-VIA-MOVE-TO-DELAYED, INT-PAUSE-WINDOWS-FILE-CONTRACT, a constitution changelog note (before-budget placement), and a job-budget rule pause-gate-before-budget. Tests: worker 432 pass (pause-windows predicate matrix incl. overnight/days/dates/ non-UTC tz; pause-gate defers-before-budget + identity preservation), admin 167 pass (gate tests + panel section), receiver unaffected. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01D3npQvQXLWsW6qsrRMAmNL
1 parent a9c302e commit a342e11

20 files changed

Lines changed: 939 additions & 7 deletions

admin/skills/operate-pi-dispatch/SKILL.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,9 @@ dialog before it takes effect**:
3030
- `dispatch_set` — change a limit/setting (e.g. `dailyCap`, `weeklyCap`, `maxTurns`, `model`). Omit `value`
3131
to unset.
3232
- `dispatch_trigger_add` / `dispatch_trigger_edit` / `dispatch_trigger_delete` — manage triggers.
33+
- `dispatch_pause_add` / `dispatch_pause_delete` — manage scheduled pause windows (per folder/repo "quiet
34+
hours": runs for a scope are deferred between certain times and auto-resume after; `dispatch_pauses` lists
35+
them). Deferring never drops a job and costs no budget.
3336

3437
Use them like this:
3538

admin/src/dashboard.ts

Lines changed: 38 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,8 @@ import { dayKey, weekKey, monthKey, tokenDayKey, windowState } from "@pi-dispatc
2222
import { parseConnection, makeRedisClient } from "@pi-dispatch/worker/connection";
2323
import { makeQueue } from "@pi-dispatch/worker/queue";
2424
import { STALL_KEY } from "@pi-dispatch/worker/scheduler-stall-guard";
25-
import { listRuns, readSettingsView, mapSchedulers, readTriggers } from "./read-model.mjs";
25+
import { windowEndAt } from "@pi-dispatch/worker/pause-windows";
26+
import { listRuns, readSettingsView, mapSchedulers, readTriggers, readPauseWindows } from "./read-model.mjs";
2627
import { renderStatus, renderBudget, renderTriggers, renderSettingsView } from "./render.mjs";
2728
import { matchesKey } from "./keys.mjs";
2829
import { box, meter, clip } from "./panel.mjs";
@@ -85,6 +86,7 @@ export function createDashboardDeps(paths: any) {
8586
runs: listRuns({ logsDir: paths.logsDir, limit: RUNS_ON_DASHBOARD }),
8687
settings: readSettingsView({ settingsFile: paths.settingsFile }),
8788
triggers: readTriggers({ triggersPath: paths.triggersPath }),
89+
pauseWindows: readPauseWindows({ pauseWindowsPath: paths.pauseWindowsPath }),
8890
// ONLY the id off the active Job -- a Job's `.data` holds issue title/body/username (PII), so it
8991
// never enters the snapshot (no-pii-in-logs, INT-RUN-HISTORY-FILE-CONTRACT).
9092
activeJobId: activeList?.[0]?.id ?? null,
@@ -308,6 +310,10 @@ export function makeDashboard({
308310
void dispose().finally(() => done({ action: "editSettings" }));
309311
return;
310312
}
313+
if (data === "w" || data === "W") {
314+
void dispose().finally(() => done({ action: "managePauses" }));
315+
return;
316+
}
311317
if (matchesKey(data, "up")) {
312318
selected = Math.max(0, selected - 1);
313319
tui?.requestRender?.();
@@ -465,6 +471,11 @@ function buildListLines(snapshot: any, selected: number, inner: number, styler:
465471
for (const l of trg.lines) lines.push(l);
466472
lines.push(RULE);
467473

474+
const pw = pauseLines(snapshot.pauseWindows, inner, styler);
475+
lines.push(styler.divider("pause windows", `${pw.count} · w manage`, inner));
476+
for (const l of pw.lines) lines.push(l);
477+
lines.push(RULE);
478+
468479
// Active + run rows follow the triggers in buildRows, so offset the selection index by the trigger count.
469480
const runRows = buildRows(snapshot).slice(trg.count);
470481
const runCount = Array.isArray(snapshot.runs) ? snapshot.runs.length : 0;
@@ -612,6 +623,31 @@ function targetColored(t: any, styler: any): string {
612623
return `${arrow} ${styler.fg("accent", "github")} ${flow}`;
613624
}
614625

626+
/** The scheduled pause windows as colored rows, each marked `●` (paused now, with a resume countdown) or `○`. */
627+
function pauseLines(pauseWindows: any, inner: number, styler: any): { count: number; lines: string[] } {
628+
const lines: string[] = [];
629+
if (pauseWindows && pauseWindows.missing) { lines.push(styler.cell("(no pause windows · w to manage)", inner, { color: "dim" })); return { count: 0, lines }; }
630+
if (pauseWindows && pauseWindows.invalid) { lines.push(styler.cell(`(pause-windows file invalid: ${pauseWindows.invalid})`, inner, { color: "error" })); return { count: 0, lines }; }
631+
const list = (pauseWindows && pauseWindows.windows) ?? [];
632+
if (list.length === 0) { lines.push(styler.cell("(no pause windows · w to manage)", inner, { color: "dim" })); return { count: 0, lines }; }
633+
const now = Date.now();
634+
for (const w of list) lines.push(pauseRow(w, now, inner, styler));
635+
return { count: list.length, lines };
636+
}
637+
638+
function pauseRow(w: any, now: number, inner: number, styler: any): string {
639+
const until = windowEndAt(w, now); // ms when this window resumes, or null when not active now
640+
const dot = until ? styler.fg("warning", "●") : styler.fg("dim", "○");
641+
const bits = [
642+
`${dot} ${styler.fg("accent", w.scope ?? "-")}`,
643+
styler.fg("text", `${w.from ?? "-"}${w.to ?? "-"}`) + " " + styler.fg("dim", w.tz ?? "UTC"),
644+
];
645+
if (w.days) bits.push(styler.fg("muted", `[${w.days.join(",")}]`));
646+
if (w.dateFrom || w.dateTo) bits.push(styler.fg("dim", `${w.dateFrom ?? "…"}${w.dateTo ?? "…"}`));
647+
if (until) bits.push(styler.fg("warning", `resumes in ${humanizeMs(until - now) || "<1m"}`));
648+
return fitLine(bits.join(styler.fg("dim", " ")), inner, styler);
649+
}
650+
615651
/** The interactive RUNS list, colored: cursor, id, target, flow, outcome (✔/⚠/✘), turns, tokens. */
616652
function runLines(rows: any[], selected: number, inner: number, styler: any): string[] {
617653
if (!Array.isArray(rows) || rows.length === 0) return [styler.cell("(no runs)", inner, { color: "dim" })];
@@ -760,7 +796,7 @@ function triggerDetailHints(inner: number, styler: any): string {
760796
/** The colored key-hint footer. */
761797
function keyHints(inner: number, styler: any): string {
762798
const k = (key: string, label: string) => styler.fg("accent", key) + " " + styler.fg("dim", label);
763-
const hints = [k("↑↓", "select"), k("↵", "open"), k("a", "add"), k("l", "logs"), k("p", "pause"), k("r", "resume"), k("q", "quit")].join(styler.fg("dim", " · "));
799+
const hints = [k("↑↓", "select"), k("↵", "open"), k("a", "add"), k("w", "pauses"), k("l", "logs"), k("p", "pause"), k("r", "resume"), k("q", "quit")].join(styler.fg("dim", " · "));
764800
return fitLine(hints, inner, styler);
765801
}
766802

admin/src/index.ts

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,10 +47,12 @@ import {
4747
readLogTail,
4848
readSettingsView,
4949
readTriggers,
50+
readPauseWindows,
5051
listRunIds,
5152
setQueuePaused,
5253
writeSettings,
5354
writeTriggers,
55+
writePauseWindows,
5456
enqueueDispatchRun,
5557
KNOWN_KEYS,
5658
} from "./read-model.mjs";
@@ -366,6 +368,83 @@ function registerTools(pi: ExtensionAPI): void {
366368
},
367369
});
368370

371+
pi.registerTool({
372+
name: "dispatch_pauses",
373+
label: "pi-dispatch pause windows",
374+
description:
375+
"Read-only. Lists the scheduled pause windows (per folder/repo quiet hours) with their array index. " +
376+
"Use the index for dispatch_pause_delete.",
377+
parameters: Type.Object({}),
378+
async execute() {
379+
const paths = resolvePaths(process.env);
380+
const p = readPauseWindows({ pauseWindowsPath: paths.pauseWindowsPath });
381+
const data = Array.isArray(p?.windows) ? p.windows.map((w: any, index: number) => ({ index, ...w })) : p;
382+
return toolText(JSON.stringify(data));
383+
},
384+
});
385+
386+
pi.registerTool({
387+
name: "dispatch_pause_add",
388+
label: "pi-dispatch add pause window",
389+
description:
390+
"Adds a scheduled pause window and applies it live: runs for `scope` (a repo \"owner/name\", a local " +
391+
"folder path, or \"*\" for all) are DEFERRED between `from` and `to` (\"HH:MM\" 24h; from>to = overnight) " +
392+
"and resume automatically after — nothing is dropped, and deferring costs no budget. Optional `tz` (IANA, " +
393+
"default UTC), `days` (mon..sun), `dateFrom`/`dateTo` (\"YYYY-MM-DD\"). The operator MUST approve a confirm " +
394+
"dialog showing the window; refused with no interactive operator.",
395+
executionMode: "sequential",
396+
parameters: Type.Object({
397+
scope: Type.String(),
398+
from: Type.String(),
399+
to: Type.String(),
400+
tz: Type.Optional(Type.String()),
401+
days: Type.Optional(Type.Array(Type.String())),
402+
dateFrom: Type.Optional(Type.String()),
403+
dateTo: Type.Optional(Type.String()),
404+
}),
405+
async execute(_id, params, _signal, _onUpdate, ctx) {
406+
const paths = resolvePaths(process.env);
407+
const w = buildPauseWindow(params);
408+
const result = await confirmedWrite(
409+
ctx,
410+
{ title: "Add pause window", message: `Add to pause-windows.json:\n${JSON.stringify(w)}` },
411+
() => {
412+
const res = writePauseWindows({ pauseWindowsPath: paths.pauseWindowsPath, mutate: (list: any[]) => [...list, w] });
413+
if (res.invalid) throw new Error(`rejected: ${res.invalid}`);
414+
return { applied: true, added: w };
415+
},
416+
);
417+
return toolText(JSON.stringify(result));
418+
},
419+
});
420+
421+
pi.registerTool({
422+
name: "dispatch_pause_delete",
423+
label: "pi-dispatch delete pause window",
424+
description:
425+
"Removes a scheduled pause window (by array index from dispatch_pauses) and applies it live. The operator " +
426+
"MUST approve a confirm dialog showing the window; refused with no interactive operator.",
427+
executionMode: "sequential",
428+
parameters: Type.Object({ index: Type.Integer({ minimum: 0 }) }),
429+
async execute(_id, params, _signal, _onUpdate, ctx) {
430+
const paths = resolvePaths(process.env);
431+
const p = readPauseWindows({ pauseWindowsPath: paths.pauseWindowsPath });
432+
const list = Array.isArray(p?.windows) ? p.windows : [];
433+
const cur = list[params.index];
434+
if (!cur) throw new Error(`no pause window at index ${params.index} (have ${list.length})`);
435+
const result = await confirmedWrite(
436+
ctx,
437+
{ title: `Delete pause window #${params.index + 1}`, message: `Remove pause window #${params.index + 1}: ${cur.scope} ${cur.from}-${cur.to} ${cur.tz}` },
438+
() => {
439+
const res = writePauseWindows({ pauseWindowsPath: paths.pauseWindowsPath, mutate: (l: any[]) => l.filter((_, i) => i !== params.index) });
440+
if (res.invalid) throw new Error(`rejected: ${res.invalid}`);
441+
return { applied: true, deletedIndex: params.index };
442+
},
443+
);
444+
return toolText(JSON.stringify(result));
445+
},
446+
});
447+
369448
pi.registerTool({
370449
name: "dispatch_trigger_delete",
371450
label: "pi-dispatch delete trigger",
@@ -455,6 +534,26 @@ function buildTriggerEntry(kind: string, f: any): any {
455534
return null;
456535
}
457536

537+
/**
538+
* Build one pause-window entry from a kind-less field bag (shared by the `dispatch_pause_add` tool and the
539+
* operator dialog). Required scope/from/to; optional tz/days/dateFrom/dateTo included only when non-blank so
540+
* an omitted field drops out of the JSON. `days` accepts an array (tool) or a space-separated string (dialog).
541+
* All value validation (time format, IANA tz, weekday names, date format) lives in the shared
542+
* `parsePauseWindows`, which the write goes through — a bad value is rejected there, never written.
543+
*/
544+
function buildPauseWindow(f: any): any {
545+
const w: any = { scope: String(f.scope ?? "").trim(), from: String(f.from ?? "").trim(), to: String(f.to ?? "").trim() };
546+
const tz = optStr(f.tz);
547+
const days = asWords(f.days);
548+
const dateFrom = optStr(f.dateFrom);
549+
const dateTo = optStr(f.dateTo);
550+
if (tz) w.tz = tz;
551+
if (days.length > 0) w.days = days;
552+
if (dateFrom) w.dateFrom = dateFrom;
553+
if (dateTo) w.dateTo = dateTo;
554+
return w;
555+
}
556+
458557
/** Normalise a labels/action field to a trimmed non-empty string list, accepting an array or a string. */
459558
function asWords(x: any): string[] {
460559
if (Array.isArray(x)) return x.map((s) => String(s).trim()).filter(Boolean);
@@ -638,9 +737,56 @@ export async function handleDashboardAction(result: any, paths: any, ctx: any):
638737
return deleteTriggerEntry(paths, ui, notify, result.index);
639738
case "editSettings":
640739
return editSettingsViaDialogs(paths, ui, notify);
740+
case "managePauses":
741+
return managePausesViaDialogs(paths, ui, notify);
641742
}
642743
}
643744

745+
/** Pause-window management: pick add or delete, then run the matching dialog. Keeps one overlay key (`w`). */
746+
async function managePausesViaDialogs(paths: any, ui: any, notify: Notify): Promise<void> {
747+
const action = await ui.select("Pause windows", ["Add a pause window", "Delete a pause window"]);
748+
if (!action) return;
749+
if (action.startsWith("Add")) return addPauseWindowViaDialogs(paths, ui, notify);
750+
return deletePauseWindowViaDialogs(paths, ui, notify);
751+
}
752+
753+
/** Add a pause window: scope + from/to, then the optional tz/days/date bounds (blank = omit). Validated + live. */
754+
async function addPauseWindowViaDialogs(paths: any, ui: any, notify: Notify): Promise<void> {
755+
const scope = await ui.input("scope — a repo \"owner/name\", a local folder path, or \"*\" for all", "");
756+
if (scope === undefined || scope.trim() === "") return;
757+
const from = await ui.input("from — pause start \"HH:MM\" 24h (from > to = overnight)", "22:00");
758+
if (from === undefined) return;
759+
const to = await ui.input("to — resume time \"HH:MM\" 24h", "06:00");
760+
if (to === undefined) return;
761+
const tz = await ui.input("tz — IANA timezone (blank = UTC), e.g. Europe/Amsterdam", "");
762+
if (tz === undefined) return;
763+
const days = await ui.input("days — space-separated weekdays to restrict to (blank = every day), e.g. mon tue", "");
764+
if (days === undefined) return;
765+
const dateFrom = await ui.input("dateFrom — only on/after \"YYYY-MM-DD\" (blank = no bound)", "");
766+
if (dateFrom === undefined) return;
767+
const dateTo = await ui.input("dateTo — only on/before \"YYYY-MM-DD\" (blank = no bound)", "");
768+
if (dateTo === undefined) return;
769+
const w = buildPauseWindow({ scope, from, to, tz, days, dateFrom, dateTo });
770+
const res = writePauseWindows({ pauseWindowsPath: paths.pauseWindowsPath, mutate: (list: any[]) => [...list, w] });
771+
notify?.(res.ok ? `pause window added (live) — ${w.scope} ${w.from}-${w.to}` : `add rejected: ${res.invalid}`, res.ok ? "info" : "error");
772+
}
773+
774+
/** Delete a pause window: select which (by a scope/time label), confirm, remove. */
775+
async function deletePauseWindowViaDialogs(paths: any, ui: any, notify: Notify): Promise<void> {
776+
const p = readPauseWindows({ pauseWindowsPath: paths.pauseWindowsPath });
777+
const list: any[] = Array.isArray(p?.windows) ? p.windows : [];
778+
if (list.length === 0) { notify?.("no pause windows to delete", "info"); return; }
779+
const labels = list.map((w, i) => `#${i + 1} ${w.scope} ${w.from}-${w.to} ${w.tz}${w.days ? ` [${w.days.join(",")}]` : ""}`);
780+
const picked = await ui.select("Delete which pause window", labels);
781+
if (!picked) return;
782+
const index = labels.indexOf(picked);
783+
if (index < 0) return;
784+
const ok = await ui.confirm("Delete pause window", `Remove ${labels[index]}?`);
785+
if (!ok) return;
786+
const res = writePauseWindows({ pauseWindowsPath: paths.pauseWindowsPath, mutate: (l: any[]) => l.filter((_, i) => i !== index) });
787+
notify?.(res.ok ? `pause window #${index + 1} deleted (live)` : `delete rejected: ${res.invalid}`, res.ok ? "info" : "error");
788+
}
789+
644790
/** Add a trigger: kind-first (the on x run diagonal is locked by construction), then the per-kind fields. */
645791
async function addTriggerViaDialogs(paths: any, ui: any, notify: Notify): Promise<void> {
646792
const kind = await ui.select("Add trigger — kind", ["cron", "label", "comment", "pull_request"]);

admin/src/read-model.mjs

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import { settingsFilePath, readOverlay, writeOverlay, KNOWN_KEYS } from "@pi-dis
2323
import { sanitizeJobId } from "@pi-dispatch/worker/run-history";
2424
import { dayKey, weekKey, monthKey } from "@pi-dispatch/worker/budget";
2525
import { parseTriggers } from "@pi-dispatch/worker/triggers";
26+
import { parsePauseWindows } from "@pi-dispatch/worker/pause-windows";
2627
import { parseConnection, makeRedisClient } from "@pi-dispatch/worker/connection";
2728
import { makeQueue, enqueueLocalJob } from "@pi-dispatch/worker/queue";
2829
import { readFlowGate } from "@pi-dispatch/worker/flow-gate";
@@ -44,6 +45,7 @@ export function resolvePaths(env = process.env) {
4445
logsDir: env.PI_LOGS_DIR || defaultLogsDir(),
4546
settingsFile: settingsFilePath(env),
4647
triggersPath: env.PI_TRIGGERS_FILE ?? "deploy/triggers.json",
48+
pauseWindowsPath: env.PI_PAUSE_WINDOWS_FILE ?? "deploy/pause-windows.json",
4749
captureJobLogs: env.PI_CAPTURE_JOB_LOGS === "1",
4850
// The two dispatch_run bounds the extension enforces producer-side, read DIRECTLY from env (never
4951
// loadConfig, which throws on unrelated GitHub-auth problems). `delimitedList`/`nonNegativeInt` are
@@ -307,6 +309,54 @@ export function writeTriggers({ triggersPath, mutate, fs = nodeFs }) {
307309
return { ok: true };
308310
}
309311

312+
/**
313+
* Read + validate the pause-windows file for display (REQ-SCOPED-PAUSE-WINDOWS). Returns `{ windows }` of
314+
* normalized entries (with `fromMin`/`toMin`), or `{ missing }` / `{ invalid }` so the viewer degrades rather
315+
* than throwing. Uses the SHARED `parsePauseWindows`, so the admin and the worker cannot drift on the schema.
316+
*/
317+
export function readPauseWindows({ pauseWindowsPath, fs = nodeFs }) {
318+
let text;
319+
try {
320+
text = fs.readFileSync(pauseWindowsPath, "utf8");
321+
} catch {
322+
return { missing: true };
323+
}
324+
try {
325+
return { windows: parsePauseWindows(text, pauseWindowsPath) };
326+
} catch (e) {
327+
return { invalid: e?.message ?? String(e) };
328+
}
329+
}
330+
331+
/**
332+
* Read-modify-write the pause-windows file (mirrors `writeTriggers`): `mutate(windows)` receives a copy of the
333+
* current raw `windows` array and returns the new array; the result is re-serialized, VALIDATED through the
334+
* SHARED `parsePauseWindows` (fail-closed — a rejected result is NEVER written, so the worker loader can always
335+
* parse it), and written ATOMICALLY (tmp + rename) so the live-reload watcher never sees a half-written file.
336+
* Reached from operator-typed `/dispatch pause …` handlers AND the confirm-gated `dispatch_pause_*` tools; the
337+
* tools route through `confirmedWrite` (an operator approves before this runs). Returns `{ ok }` or `{ invalid }`.
338+
*/
339+
export function writePauseWindows({ pauseWindowsPath, mutate, fs = nodeFs }) {
340+
let current = [];
341+
try {
342+
const raw = JSON.parse(fs.readFileSync(pauseWindowsPath, "utf8"));
343+
if (Array.isArray(raw?.windows)) current = raw.windows;
344+
} catch {
345+
// Missing/invalid file: start from empty; the validated atomic write below repairs it.
346+
}
347+
const next = mutate(current.map((w) => ({ ...w })));
348+
const text = `${JSON.stringify({ windows: next }, null, 2)}\n`;
349+
try {
350+
parsePauseWindows(text, pauseWindowsPath); // the loader's own validator -- never write a file it would reject
351+
} catch (e) {
352+
return { invalid: e?.message ?? String(e) };
353+
}
354+
const tmp = `${pauseWindowsPath}.tmp`;
355+
fs.writeFileSync(tmp, text, { mode: 0o644 });
356+
fs.renameSync(tmp, pauseWindowsPath);
357+
return { ok: true };
358+
}
359+
310360
async function readWorkerCount(queue) {
311361
try {
312362
const list = await queue.getWorkers();

0 commit comments

Comments
 (0)