src/ops-feed.ts

Functions

Imports

It imports src/agents.ts, src/auth.ts, src/auto-merge-policy.ts, src/auto-merge-tick.ts, src/canon.ts, src/env.ts, src/http.ts, src/inbox.ts, src/job-signing.ts, src/jobs-holder.ts, src/jobs-schema.ts, src/log.ts, src/merge-safety.ts, src/ops-hooks.ts, src/ops-packages.ts, src/ops-sites.ts, src/ops-snapshot.ts, src/ops-types/feed.ts, src/overnight.ts, src/portal-auth.ts, src/portal-host.ts, src/roster.ts, src/status.ts, src/store-guards.ts, src/task-runs.ts, src/time.ts, src/watcher-digest.ts, src/watcher-gather.ts, src/watcher.ts. Imported by src/portal-actions.ts, src/portal-claims.ts, src/portal-convergence.ts, src/portal-library.ts, src/portal-maintenance.ts, src/portal-overnight.ts, src/portal-packages.ts, src/portal-stale.ts, src/portal-trends.ts, src/push.ts, src/routes.ts.

Source

1import { gatherInbox } from "./inbox";
2import { adminAgentForEmail } from "./agents";
3import { getCookie } from "./auth";
4import { AWAITING_SEAT_KEY } from "./auto-merge-tick";
5import { loadMergePolicy } from "./auto-merge-policy";
6import { readSafetyReport } from "./merge-safety";
7import { PORTAL_CSRF_COOKIE, PORTAL_PREFIX, PORTAL_SESSION_TTL_SECONDS, type PortalUser, portalDisplay, portalGate, sourceAddress } from "./portal-auth";
8import { portalCookiePath } from "./portal-host";
9import type { Env } from "./env";
10import { agentSummaries, type AgentSummary } from "./status";
11import { APP_ROSTER } from "./roster";
12import { commandFromSummary, isQuestionSummary, resumeMarkerIn } from "./jobs-holder";
13import { checkJobText, type SignatureCheck } from "./job-signing";
14import { OPEN_JOB_STATUSES } from "./jobs-schema";
15import { readSiteConfig } from "./ops-sites";
16import { readPackageConfig } from "./ops-packages";
17import { readSnapshot } from "./ops-snapshot";
18import { INCIDENT_FAILURES, NEEDS_INPUT_INCIDENT_MS } from "./ops-hooks";
19import type { OpsAgent, OpsAwaitingSeat, OpsFeed, OpsJob, OpsJobStatus, OpsLive, OpsPr, OpsSession } from "./ops-types/feed";
20import { overnightState } from "./overnight";
21import { auditStatement } from "./store-guards";
22import { readTaskRuns, taskStates, TASKS } from "./task-runs";
23import { openQuietFindings } from "./watcher-digest";
24import { DEFAULT_CADENCE_MINUTES, watcherTick, WATCHER_ACTOR, type Gathered, type WatcherReport } from "./watcher";
25import { gatherFindings } from "./watcher-gather";
26import { logEvent } from "./log";
27import { pendingProposals } from "./canon";
28import { textResponse } from "./http";
29import { MINUTE_MS, DAY_MS } from "./time";
30
31// The Watch Floor's one read (capsid/research/design-ops-console.md): GET
32// /portal/api/ops returns OpsFeed (src/ops-types/feed.ts), the watcher's last pass from KV
33// plus what changes between passes, read live. POST /portal/api/ops/refresh runs one
34// watcher pass now and returns the new feed.
35//
36// Both answer to portalGate, as every Portal route does: the administrator's Access
37// session and the ADMIN_EMAIL or CO_OWNER_EMAILS check on every request, and a 403 for any Authorization header.
38// They are routes, not tools, so no grant is checked here
39// (one enforcement point, test/route-gates.test.ts); src/scope.ts lists them among the routes gated some other way.
40//
41// READS PER FEED REQUEST, stated because the dashboard polls this and a per-namespace
42// loop would multiply them. Asserted by test-integration/ops-feed.test.ts, which counts.
43//   D1, 13 statements plus one per TASKS entry (9 today, so 22):
44//     1  jobs: every open job and every job that ended in the last 24 hours
45//     3  agentSummaries: the inventory, then loadRecordRows' two grouped reads
46//     1  job_outcome_prs in the last 7 days
47//     1  ops_sites: the site configuration, every row
48//     1  ops_packages: the package configuration, every row
49//     1  agent_sessions: live sessions from the hook receiver, at most 50
50//     1  canon_proposals: the pending proposals joined to their documents, at most 50
51//     9  task_runs: each scheduled task's newest runs, one keyed read per task in
52//        one batch (src/task-runs.ts, TASKS)
53//     1  watcher_findings: the open findings that post no job (src/watcher-digest.ts)
54//     3  gatherInbox (src/inbox.ts): the blocked jobs, the applied migrations (a block
55//        on a migration already applied is settled) and the site configuration again,
56//        so Needs you counts exactly what GET /ops/inbox answers with
57//   KV, 10 gets: gatherInbox's own
58//     reads of the stale pull request cache (src/stale-jobs.ts), the manual steps cache
59//     (src/manual-steps.ts), ops:snapshot, the awaiting-seat set and the merge units' report
60//     (src/merge-units.ts), then ops:snapshot, the
61//     awaiting-seat set, the auto-merge safety report (src/merge-safety.ts), the refresh stamp,
62//     and the overnight switch's mode (its decision record is
63//     read only while the mode is subscription).
64// They run concurrently; the longest chain is agentSummaries' two steps.
65
66export const OPS_FEED_PATH = "/portal/api/ops";
67export const OPS_REFRESH_PATH = "/portal/api/ops/refresh";
68// Where a sign-in started from one of these routes lands afterwards: the app.
69export const OPS_RETURN_TO = PORTAL_PREFIX;
70
71// D1 statements: one per task, and 14 more, one of them the signed policy's read (liveMergePolicy).
72export const OPS_FEED_READS = { d1: 14 + TASKS.length, kv: 10 } as const;
73
74// The Portal's double-submit CSRF cookie (OpsFeed.csrf), named in src/portal-auth.ts.
75// Minted when absent or malformed and then left alone, never rotated per poll, so a
76// preview and its perform carry one value.
77export { PORTAL_CSRF_COOKIE };
78const CSRF_SHAPE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
79
80/** The request's Portal CSRF value, and the Set-Cookie that mints one when it has none. */
81export function portalCsrf(request: Request): { value: string; setCookie: string | null } {
82  const presented = getCookie(request, PORTAL_CSRF_COOKIE);
83  if (presented !== null && CSRF_SHAPE.test(presented)) return { value: presented, setCookie: null };
84  const value = crypto.randomUUID();
85  return {
86    value,
87    setCookie: `${PORTAL_CSRF_COOKIE}=${value}; HttpOnly; Secure; SameSite=Lax; Path=${portalCookiePath(new URL(request.url))}; Max-Age=${PORTAL_SESSION_TTL_SECONDS}`,
88  };
89}
90
91// The refresh's rate limit: one on-demand pass per two minutes, stamped in KV.
92export const OPS_REFRESH_KEY = "ops:refresh:last";
93const OPS_REFRESH_INTERVAL_MS = 2 * MINUTE_MS;
94// A cross-site form cannot set a custom header, and a cross-site fetch that sets one
95// is preflighted and refused, so its presence says the request came from this origin.
96export const OPS_REFRESH_HEADER = "X-Capsid-Ops";
97const OPS_REFRESH_HEADER_VALUE = "refresh";
98
99const WEEK_MS = 7 * DAY_MS;
100// The live-sessions read's bound.
101const SESSION_ROWS_LIMIT = 50;
102
103// D1's datetime('now') text form, for comparing against columns written by default.
104const sqliteTime = (at: Date): string => at.toISOString().slice(0, 19).replace("T", " ");
105
106/** An instant as ISO. Columns written by datetime('now') read "YYYY-MM-DD HH:MM:SS",
107 *  which is UTC but which a browser's Date.parse reads as local time, so the feed
108 *  hands every timestamp over in one unambiguous form. Anything else passes through. */
109export function isoTime(value: string): string;
110export function isoTime(value: string | null): string | null;
111export function isoTime(value: string | null): string | null {
112  if (value === null) return null;
113  return /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(value) ? `${value.replace(" ", "T")}Z` : value;
114}
115
116export interface JobFeedRow {
117  id: string;
118  namespace: string;
119  title: string;
120  status: string;
121  priority: number;
122  posted_by: string;
123  claimed_by: string | null;
124  created_at: string;
125  updated_at: string;
126  blocked_count: number;
127  resumed_count: number;
128  gate_required: number;
129  result_ref: string | null;
130  result_summary: string | null;
131  summary_sig?: string | null;
132  // The started label (migrations/0042). Absent in a row read before it.
133  started_by?: string | null;
134  started_at?: string | null;
135  // From the finding's watcher_findings row when this job is its current one
136  // (src/watcher-findings.ts). Null, or absent, otherwise.
137  finding_seen_count?: number | null;
138  finding_last_seen?: string | null;
139}
140
141// The watcher's title carries its fingerprint in brackets at the end. The same
142// pattern openWatcherFingerprints (src/watcher.ts) reads it with; test/ops-feed.test.ts
143// runs both over one set of titles so the two cannot disagree.
144const FINGERPRINT = /\[([^\]]+)\]\s*$/;
145
146/** A job as the feed shows it. A blocked job's summary is split into what it waits on
147 *  and the exact command (commandFromSummary), so the app can show the command to copy. */
148export function opsJobFrom(row: JobFeedRow, commandSignature: SignatureCheck | null = null): OpsJob {
149  const blocked = row.status === "blocked";
150  const summary = row.result_summary;
151  const marker = summary === null ? -1 : (resumeMarkerIn(summary)?.at ?? -1);
152  const waitsOn = !blocked || summary === null ? null : (marker === -1 ? summary : summary.slice(0, marker)).trim() || null;
153  const print = row.posted_by === WATCHER_ACTOR ? FINGERPRINT.exec(row.title) : null;
154  return {
155    id: row.id,
156    namespace: row.namespace,
157    title: row.title,
158    status: row.status as OpsJobStatus,
159    priority: row.priority,
160    posted_by: row.posted_by,
161    claimed_by: row.claimed_by,
162    created_at: isoTime(row.created_at),
163    updated_at: isoTime(row.updated_at),
164    blocked_count: row.blocked_count,
165    resumed_count: row.resumed_count,
166    gate_required: row.gate_required === 1,
167    waits_on: waitsOn,
168    // A command whose signature does not match is withheld, so it cannot be copied.
169    command: blocked && commandSignature !== "mismatch" ? commandFromSummary(summary) : null,
170    command_signature: blocked ? commandSignature : null,
171    question: blocked && commandSignature !== "mismatch" && isQuestionSummary(summary),
172    ...(row.status === "parked" && summary !== null ? { parked_reason: summary } : {}),
173    ...(row.started_by && row.started_at ? { started_by: row.started_by, started_at: isoTime(row.started_at) } : {}),
174    result_ref: row.result_ref,
175    finding: print ? { fingerprint: print[1], seen_count: row.finding_seen_count ?? null, last_seen: isoTime(row.finding_last_seen ?? null) } : null,
176  };
177}
178
179/** One credential as the feed shows it: the inventory capsid_run_improve action status serves, with its
180 *  record. Grants are left out; the contract carries flags only. */
181export function opsAgentFrom(agent: AgentSummary): OpsAgent {
182  const r = agent.record;
183  return {
184    name: agent.name,
185    kind: agent.kind,
186    namespaces: agent.namespaces,
187    flags: agent.flags,
188    last_seen: isoTime(agent.last_seen),
189    revoked_at: isoTime(agent.revoked_at),
190    archived_at: isoTime(agent.archived_at),
191    jobs_done: r.jobs_done,
192    jobs_failed: r.jobs_failed,
193    jobs_blocked: r.jobs_blocked,
194    prs_opened: r.prs_opened,
195    prs_merged: r.prs_merged,
196    pr_merge_rate: r.pr_merge_rate,
197    ci_green_rate: r.ci_green_rate,
198    median_duration_minutes: r.median_duration_minutes,
199  };
200}
201
202/** The auto-merge tick's awaiting-seat set. A value that does not parse is logged and
203 *  shown as empty, because the contract has no field to carry the reason; a KV read
204 *  that fails throws and fails the feed. */
205export function awaitingFrom(raw: string | null): OpsAwaitingSeat[] {
206  if (raw === null) return [];
207  try {
208    const parsed: unknown = JSON.parse(raw);
209    if (!Array.isArray(parsed)) throw new Error("not an array");
210    return parsed as OpsAwaitingSeat[];
211  } catch (err) {
212    logEvent("error", "OPS_FEED_AWAITING_UNREADABLE", { message: `OPS_FEED_AWAITING_UNREADABLE ${AWAITING_SEAT_KEY}: ${err instanceof Error ? err.message : String(err)}` });
213    return [];
214  }
215}
216
217/** When the next on-demand pass is allowed, or null when one is allowed now. A stamp
218 *  that does not parse allows one: the limiter's put writes a fresh stamp. */
219export function refreshAllowedAt(last: string | null, now: Date): string | null {
220  if (last === null) return null;
221  const at = Date.parse(last);
222  if (Number.isNaN(at)) return null;
223  const next = at + OPS_REFRESH_INTERVAL_MS;
224  return next > now.getTime() ? new Date(next).toISOString() : null;
225}
226
227// The store's size against D1's cap. Every D1 result carries the database's size in
228// meta.size_after, so the jobs read below reports it and the feed makes no extra read
229// (Foxhound reads it the same way: PRAGMA page_count is not authorized on D1). 10 GB is
230// the per-database cap on Workers Paid (developers.cloudflare.com/d1/platform/limits);
231// on Workers Free it is 500 MB, and this is the one line to change.
232export const D1_CAP_BYTES = 10 * 1024 ** 3;
233
234/** The size a D1 result reports, or null when its meta has no number for it. */
235export function sizeFrom(meta: unknown): number | null {
236  const size = (meta as { size_after?: unknown } | null | undefined)?.size_after;
237  return typeof size === "number" && Number.isFinite(size) && size >= 0 ? size : null;
238}
239
240async function liveJobs(db: D1Database, now: Date, secret: string | undefined): Promise<{ jobs: OpsJob[]; sizeBytes: number | null }> {
241  const open = OPEN_JOB_STATUSES.map((_, i) => `?${i + 2}`).join(", ");
242  // datetime() on both sides: updated_at is written both as ISO and as D1's default
243  // text form, and the two do not compare as text. The watcher's finding row joins
244  // on the job it is currently filed as, so a watcher job shows how often its finding
245  // has been seen; one row per fingerprint, so the join cannot multiply a job. A parked
246  // job is not open (src/jobs-schema.ts) but is listed whatever its age, so it does not
247  // vanish from the Queue a day after it was parked.
248  const { results, meta } = await db
249    .prepare(
250      `SELECT j.id, j.namespace, j.title, j.status, j.priority, j.posted_by, j.claimed_by, j.created_at, j.updated_at,
251              j.blocked_count, j.resumed_count, j.gate_required, j.result_ref, j.result_summary, j.summary_sig,
252              j.started_by, j.started_at,
253              wf.seen_count AS finding_seen_count, wf.last_seen_at AS finding_last_seen
254       FROM jobs j LEFT JOIN watcher_findings wf ON wf.job_id = j.id
255       WHERE j.status IN (${open}) OR j.status = 'parked' OR datetime(j.updated_at) >= datetime(?1)
256       ORDER BY j.updated_at DESC`
257    )
258    .bind(new Date(now.getTime() - DAY_MS).toISOString(), ...OPEN_JOB_STATUSES)
259    .all<JobFeedRow>();
260  const rows = results ?? [];
261  const checks = await Promise.all(
262    rows.map((r) =>
263      r.status === "blocked" && r.result_summary !== null
264        ? checkJobText(secret, "summary", r.id, { summary: r.result_summary }, r.summary_sig ?? null)
265        : Promise.resolve(null)
266    )
267  );
268  return { jobs: rows.map((r, i) => opsJobFrom(r, checks[i])), sizeBytes: sizeFrom(meta) };
269}
270
271async function livePrs(db: D1Database, now: Date): Promise<OpsPr[]> {
272  const { results } = await db
273    .prepare(
274      `SELECT job_id, pr_url, merged, merge_verified_at, recorded_at FROM job_outcome_prs
275       WHERE recorded_at >= ?1 ORDER BY recorded_at DESC`
276    )
277    .bind(sqliteTime(new Date(now.getTime() - WEEK_MS)))
278    .all<{ job_id: string; pr_url: string; merged: number | null; merge_verified_at: string | null; recorded_at: string }>();
279  return (results ?? []).map((r) => ({
280    job_id: r.job_id,
281    pr_url: r.pr_url,
282    // Three states, kept: null is never checked, not closed (migrations/0015).
283    merged: r.merged === null ? null : r.merged === 1,
284    merge_verified_at: isoTime(r.merge_verified_at),
285    recorded_at: isoTime(r.recorded_at),
286  }));
287}
288
289// The row as D1 holds it: needs_input is SQLite's 0 or 1, and the incident is decided
290// when the feed is read.
291export type SessionFeedRow = Omit<OpsSession, "needs_input" | "incident"> & { needs_input: number };
292
293/** A session as the feed shows it, with its incident decided now: a failure a person
294 *  must act on, or input awaited for more than ten minutes (src/ops-hooks.ts). */
295export function opsSessionFrom(row: SessionFeedRow, now: Date): OpsSession {
296  const needsInput = row.needs_input === 1;
297  const waitedMs = now.getTime() - Date.parse(isoTime(row.last_event_at));
298  const incident =
299    row.last_failure !== null && INCIDENT_FAILURES.includes(row.last_failure)
300      ? "failure"
301      : needsInput && waitedMs > NEEDS_INPUT_INCIDENT_MS
302        ? "waiting"
303        : null;
304  return {
305    session_id: row.session_id,
306    agent: row.agent,
307    job_id: row.job_id,
308    namespace: row.namespace,
309    source: row.source,
310    model: row.model,
311    permission_mode: row.permission_mode,
312    started_at: isoTime(row.started_at),
313    last_event_at: isoTime(row.last_event_at),
314    last_event: row.last_event,
315    last_notification_type: row.last_notification_type,
316    needs_input: needsInput,
317    last_failure: row.last_failure,
318    incident,
319  };
320}
321
322// Through agent_sessions_live (ended_at, last_event_at): the NULL ended_at and the
323// range on last_event_at are both on the index, which also gives the order.
324async function liveSessions(db: D1Database, now: Date): Promise<OpsSession[]> {
325  const { results } = await db
326    .prepare(
327      `SELECT session_id, agent, job_id, namespace, source, model, permission_mode, started_at, last_event_at, last_event,
328              last_notification_type, needs_input, last_failure
329       FROM agent_sessions WHERE ended_at IS NULL AND last_event_at >= ?1
330       ORDER BY last_event_at DESC LIMIT ?2`
331    )
332    .bind(new Date(now.getTime() - DAY_MS).toISOString(), SESSION_ROWS_LIMIT)
333    .all<SessionFeedRow>();
334  return (results ?? []).map((row) => opsSessionFrom(row, now));
335}
336
337// The policy as the tick loads it. A throw is reported as not loading, not raised: one
338// unreadable policy must not take the whole live feed down.
339async function liveMergePolicy(env: Env): Promise<OpsLive["merge_policy"]> {
340  try {
341    const loaded = await loadMergePolicy(env);
342    return "error" in loaded ? { loads: false, reason: loaded.error } : { loads: true, version: loaded.policy.version, enabled: loaded.policy.enabled };
343  } catch (err) {
344    return { loads: false, reason: `the policy could not be read: ${err instanceof Error ? err.message : String(err)}` };
345  }
346}
347
348async function opsLive(env: Env, now: Date): Promise<OpsLive> {
349  const [jobRead, agents, prs, awaitingRaw, mergeSafety, sites, sessions, packages, overnight, canonProposals, inbox, findings, mergePolicy] = await Promise.all([
350    liveJobs(env.DB, now, env.IMPROVE_SCORE_SECRET),
351    agentSummaries(env.DB),
352    livePrs(env.DB, now),
353    env.APP_KV.get(AWAITING_SEAT_KEY),
354    readSafetyReport(env.APP_KV),
355    readSiteConfig(env.DB),
356    liveSessions(env.DB, now),
357    readPackageConfig(env.DB),
358    overnightState(env),
359    pendingProposals(env.DB),
360    gatherInbox(env, now, null),
361    openQuietFindings(env.DB),
362    liveMergePolicy(env),
363  ]);
364  return {
365    generated: now.toISOString(),
366    jobs: jobRead.jobs,
367    agents: agents.map(opsAgentFrom),
368    prs,
369    awaiting_seat: awaitingFrom(awaitingRaw),
370    merge_safety: mergeSafety,
371    merge_policy: mergePolicy,
372    overnight,
373    sessions,
374    namespaces: APP_ROSTER.map((name) => ({ name })),
375    sites,
376    packages,
377    store: { size_bytes: jobRead.sizeBytes, cap_bytes: D1_CAP_BYTES },
378    canon_proposals: canonProposals,
379    inbox,
380    findings,
381  };
382}
383
384// The feed without its csrf, which comes off the request (portalCsrf) and is added by
385// the handler, so opsFeed reads storage only.
386export type OpsFeedData = Omit<OpsFeed, "csrf" | "user">;
387
388/** The run ledger as the feed shows it. A failed read is said in the panel, not
389 *  turned into a failure of the whole feed. */
390async function scheduled(env: Env): Promise<{ read: Awaited<ReturnType<typeof readTaskRuns>> } | { error: string }> {
391  try {
392    return { read: await readTaskRuns(env.DB) };
393  } catch (err) {
394    const message = err instanceof Error ? err.message : String(err);
395    logEvent("error", "OPS_FEED_TASK_RUNS_UNREADABLE", { message: `OPS_FEED_TASK_RUNS_UNREADABLE ${message}` });
396    return { error: message };
397  }
398}
399
400export async function opsFeed(env: Env, now: Date): Promise<OpsFeedData> {
401  const [snapshot, live, last, runs] = await Promise.all([readSnapshot(env), opsLive(env, now), env.APP_KV.get(OPS_REFRESH_KEY), scheduled(env)]);
402  return {
403    snapshot,
404    live,
405    scheduled:
406      "error" in runs
407        ? { tasks: null, error: runs.error }
408        : { tasks: taskStates(runs.read, now, snapshot?.cadence_min ?? DEFAULT_CADENCE_MINUTES), error: null },
409    refresh_allowed_at: refreshAllowedAt(last, now),
410    cloudflare_configured: Boolean(env.CF_OPS_TOKEN),
411  };
412}
413
414function feedResponse(request: Request, user: PortalUser, data: OpsFeedData, extra: Record<string, string> = {}): Response {
415  const csrf = portalCsrf(request);
416  const feed: OpsFeed = { ...data, csrf: csrf.value, user: portalDisplay(user) };
417  const headers = new Headers({ "Content-Type": "application/json", "Cache-Control": "no-store", ...extra });
418  if (csrf.setCookie) headers.set("Set-Cookie", csrf.setCookie);
419  return new Response(JSON.stringify(feed), { status: 200, headers });
420}
421
422// What the two handlers can be handed in a test in place of the live reads. The
423// watcher pass itself is not injectable: the refresh always calls watcherTick with
424// force, and only the gather under it is swapped.
425export interface OpsDeps {
426  feed?: (env: Env, now: Date) => Promise<OpsFeedData>;
427  gather?: (env: Env, now: Date) => Promise<Gathered>;
428}
429
430export async function handleOpsFeed(request: Request, env: Env, now: Date = new Date(), deps: OpsDeps = {}): Promise<Response> {
431  const gate = await portalGate(request, env, now, OPS_RETURN_TO);
432  if (!gate.ok) return gate.response;
433  return feedResponse(request, gate.user, await (deps.feed ?? opsFeed)(env, now));
434}
435
436export async function handleOpsRefresh(request: Request, env: Env, now: Date = new Date(), deps: OpsDeps = {}): Promise<Response> {
437  const gate = await portalGate(request, env, now, OPS_RETURN_TO);
438  if (!gate.ok) return gate.response;
439  if (request.headers.get(OPS_REFRESH_HEADER) !== OPS_REFRESH_HEADER_VALUE) {
440    return textResponse(`forbidden: a refresh must carry the header ${OPS_REFRESH_HEADER}: ${OPS_REFRESH_HEADER_VALUE}. The dashboard sends it; a cross-site form cannot.`, 403);
441  }
442
443  // The rate limit fails closed: a stamp that cannot be read or written refuses the
444  // pass, because an unlimited Refresh is a button that runs every GitHub read the
445  // watcher makes as often as it is clicked.
446  let last: string | null;
447  try {
448    last = await env.APP_KV.get(OPS_REFRESH_KEY);
449  } catch (err) {
450    const message = err instanceof Error ? err.message : String(err);
451    logEvent("error", "OPS_REFRESH_LIMIT_UNREADABLE", { message: `OPS_REFRESH_LIMIT_UNREADABLE ${message}` });
452    return textResponse(`the refresh rate limit could not be read (${message}), so no pass was run. Try again shortly.`, 503);
453  }
454  const allowedAt = refreshAllowedAt(last, now);
455  if (allowedAt !== null) {
456    const seconds = Math.max(1, Math.ceil((Date.parse(allowedAt) - now.getTime()) / 1000));
457    return textResponse(`a watcher pass was run on demand less than two minutes ago. The next is allowed at ${allowedAt}.`, 429, {
458      "Retry-After": String(seconds),
459    });
460  }
461  try {
462    await env.APP_KV.put(OPS_REFRESH_KEY, now.toISOString());
463  } catch (err) {
464    const message = err instanceof Error ? err.message : String(err);
465    logEvent("error", "OPS_REFRESH_LIMIT_UNWRITABLE", { message: `OPS_REFRESH_LIMIT_UNWRITABLE ${message}` });
466    return textResponse(`the refresh rate limit could not be written (${message}), so no pass was run. Try again shortly.`, 503);
467  }
468
469  const actor = adminAgentForEmail(gate.user.email).actor;
470  const gather = deps.gather ?? gatherFindings;
471  let report: WatcherReport;
472  try {
473    report = await watcherTick(env, now, () => gather(env, now), { force: true });
474  } catch (err) {
475    const message = err instanceof Error ? err.message : String(err);
476    logEvent("error", "OPS_REFRESH_PASS_FAILED", { message: `OPS_REFRESH_PASS_FAILED ${message}` });
477    await env.DB.batch([auditStatement(env.DB, actor, "portal-ops-refresh", null, null, { ran: false, error: message, source_address: sourceAddress(request) })]);
478    return textResponse(`the watcher pass failed: ${message}`, 500);
479  }
480
481  // The click's own audit row, as every Portal action writes one (src/portal-actions.ts).
482  let warning: string | null = null;
483  try {
484    await env.DB.batch([
485      auditStatement(env.DB, actor, "portal-ops-refresh", null, null, { ran: report.ran, note: report.note, posted: report.posted, cleared: report.cleared, source_address: sourceAddress(request) }),
486    ]);
487  } catch (err) {
488    // The pass happened; only its audit row failed. Said in a header and the log, not
489    // turned into a failure of a pass that ran.
490    warning = `the pass ran, but the Portal audit row naming ${actor} was not written: ${err instanceof Error ? err.message : String(err)}`;
491    logEvent("error", "OPS_REFRESH_AUDIT_FAILED", { message: warning });
492  }
493  const feed = await (deps.feed ?? opsFeed)(env, now);
494  return feedResponse(request, gate.user, feed, warning ? { "X-Capsid-Warning": warning.replace(/[^\x20-\x7e]+/g, " ") } : {});
495}