src/watcher-gather.ts

Functions

Imports

It imports src/anomaly.ts, src/deploy-outcomes.ts, src/domain-expiry.ts, src/env.ts, src/github.ts, src/health.ts, src/live-checks.ts, src/log.ts, src/manual-steps.ts, src/ops-cloudflare.ts, src/ops-packages.ts, src/ops-sites.ts, src/ops-snapshot.ts, src/ops-types/sites.ts, src/roster.ts, src/secret-presence.ts, src/site-watch.ts, src/stale-canon.ts, src/status.ts, src/unmapped-repos.ts, src/watcher-checks.ts, src/watcher-detect.ts, src/watcher.ts. Imported by src/ops-feed.ts, src/tick.ts.

Source

1import { readActionHours, readToolHours, anomalyFindings, toolRateFindings } from "./anomaly";
2import { DEPLOY_ROSTER, readDeployOutcomes, deployFindings } from "./deploy-outcomes";
3import { registrableDomain, readExpiries, expiryFindings } from "./domain-expiry";
4import type { Env } from "./env";
5import { defaultBranchSha, ciStatus, listRepoTree } from "./github";
6import { healthReport } from "./health";
7import { readLiveConfig, liveChecks, defaultBranchHead } from "./live-checks";
8import { logEvent } from "./log";
9import { checkManualSteps, liveStepReaders } from "./manual-steps";
10import { cloudflareCredentials, readWebAnalyticsSites, readCloudflare, readSecretNames, readAccessAppNames } from "./ops-cloudflare";
11import { readPackageConfig, readPackage, npmVersions } from "./ops-packages";
12import { readSiteConfig, type OpsSite, sitesFrom, siteMapDrift } from "./ops-sites";
13import { type SiteProbe, probeSite, readSnapshot } from "./ops-snapshot";
14import type { PackageSnapshot } from "./ops-types/sites";
15import { APP_ROSTER } from "./roster";
16import { secretPresence, storeSecrets } from "./secret-presence";
17import { watchSites } from "./site-watch";
18import { readStaleCanon } from "./stale-canon";
19import { capsidStatus } from "./status";
20import { readRepoMap } from "./unmapped-repos";
21import { type Gathered, type Observed, WATCHER_ACTOR } from "./watcher";
22import type { WatcherCheck } from "./watcher-checks";
23import { type Finding, type DeployVerdict, deployVerdict, healthFindings, statusFindings, MIRROR_DUMP_PREFIX, MIRROR_REPO_LABEL, type MirrorRun, newestDump, mirrorFindings, latestMirrorRun, type CiRun, ciFindings, siteMapFindings, unmappedRepoFindings, repoMapUnreadableFinding, siteDownFindings, liveFinding, finding, siteErrorFindings } from "./watcher-detect";
24
25// The watcher's reads (src/watcher.ts runs the pass): every surface it checks, read
26// once per pass and turned into findings by src/watcher-detect.ts.
27//
28// Every read is wrapped so one failing surface does not stop the others being
29// checked. A watcher that goes silent because GitHub was slow is worse than one that
30// reports three of its four checks, because silence reads as health.
31
32async function attempt<T>(what: string, fn: () => Promise<T>): Promise<T | null> {
33  try {
34    return await fn();
35  } catch (err) {
36    logEvent("error", "WATCHER_READ_FAILED", { message: `WATCHER_READ_FAILED ${what}: ${err instanceof Error ? err.message : String(err)}` });
37    return null;
38  }
39}
40
41const basename = (path: string): string => path.split("/").filter(Boolean).pop() ?? "";
42
43/** The newest migration filename on master, compared against the live schema_version.
44 *  Sorted by name: the files are zero-padded. */
45export function newestMigration(names: string[]): string | null {
46  const sql = names.filter((n) => n.endsWith(".sql")).sort();
47  return sql.length ? sql[sql.length - 1] : null;
48}
49
50export async function gatherFindings(env: Env, now: Date, fetchImpl: typeof fetch = fetch): Promise<Gathered> {
51  const out: Finding[] = [];
52  const ran = new Set<WatcherCheck>();
53  const observed: Observed = { health: null, mirror: null, ci: [], siteMap: null, probes: null };
54
55  // No casts on a repo reader's result, so a renamed field fails `npm run check`.
56  // attempt() swallows a throw, so a cast would hide a check that never runs.
57  // test/watcher-gather.test.ts drives both reads end to end.
58  const health = await attempt("health", () => healthReport(env));
59  observed.health = health;
60  if (health) {
61    ran.add("health");
62    const head = await attempt("master head", () => defaultBranchSha(env, "capsid"));
63    // Master head's own CI run, read only when the deployed sha differs. A deploy
64    // still in flight leaves the check un-judged this pass: nothing is posted and an
65    // open drift finding is not cleared on it.
66    let deploy: DeployVerdict | undefined;
67    if (head !== null && health.sha && health.sha !== "unknown" && health.sha !== head) {
68      const runs = await attempt("head ci run", async () => (await ciStatus(env, "capsid", undefined, { ref: head, limit: 10 })).runs);
69      deploy = deployVerdict(runs, now);
70    }
71    if (head !== null && !deploy?.wait) ran.add("master head");
72    // Wrapped in an object so a tree with no migrations (null) is told apart from a
73    // read that failed (also null from attempt).
74    const migrations = await attempt("migrations", async () => {
75      const tree = await listRepoTree(env, "capsid", "migrations");
76      return { newest: newestMigration(tree.entries.map((e) => basename(e.path))) };
77    });
78    if (migrations !== null) ran.add("migrations");
79    out.push(...healthFindings(health, head ?? null, migrations?.newest ?? null, "capsid", deploy));
80  }
81
82  const status = await attempt("status", () => capsidStatus(env));
83  if (status) {
84    ran.add("status");
85    out.push(...statusFindings(status));
86  }
87
88  // The off-account mirror, resolved through the capsid namespace mapping (selector
89  // "backups"). A failed dump read skips the check: "cannot see the mirror" is not
90  // "the mirror is dead", and an outage must not file a job every half hour. A failed
91  // run read degrades to an empty list, so the stale-dump finding still fires.
92  const dumps = await attempt("mirror dumps", async () => {
93    const tree = await listRepoTree(env, "capsid", MIRROR_DUMP_PREFIX, undefined, MIRROR_REPO_LABEL);
94    return tree.entries;
95  });
96  if (dumps) {
97    ran.add("mirror dumps");
98    const runs: MirrorRun[] | null = await attempt("mirror runs", async () => (await ciStatus(env, "capsid", MIRROR_REPO_LABEL, { limit: 10 })).runs);
99    if (runs) ran.add("mirror runs");
100    const newest = newestDump(dumps);
101    out.push(...mirrorFindings("capsid", newest, runs ?? [], now));
102    const last = runs ? latestMirrorRun(runs) : null;
103    observed.mirror = {
104      newest_dump: newest?.toISOString() ?? null,
105      last_run: last ? { at: last.created_at ?? null, conclusion: last.conclusion ?? null, url: last.url ?? null } : null,
106    };
107  }
108
109  // The ci check counts as run only when every roster repo was read. A ci-red
110  // fingerprint names its namespace, but ownership is per check, not per repo, so a
111  // repo that could not be read must not let the others' pass clear its incident.
112  let ciRead = 0;
113  for (const namespace of APP_ROSTER) {
114    const runs: CiRun[] | null = await attempt(`ci ${namespace}`, async () => (await ciStatus(env, namespace, undefined, { limit: 5 })).runs);
115    if (runs) {
116      ciRead++;
117      out.push(...ciFindings(namespace, runs, now));
118    }
119    const latest = runs?.find((r) => r.status === "completed") ?? runs?.[0] ?? null;
120    observed.ci.push({
121      namespace,
122      latest: latest
123        ? { head_sha: latest.head_sha, status: latest.status, conclusion: latest.conclusion, created_at: latest.created_at, url: latest.url ?? null }
124        : null,
125    });
126  }
127  if (ciRead === APP_ROSTER.length) ran.add("ci");
128
129  // What became of each roster repo's newest deploys (src/deploy-outcomes.ts, the merge
130  // pipeline's PR 6): recorded in deploy_outcomes, and a rollback is a finding. The check
131  // counts as run only when every repo was read, so a failed read never clears one.
132  let deploysRead = 0;
133  for (const namespace of DEPLOY_ROSTER) {
134    // Wrapped, since a repo with no deploy yet reads as null and a failed read does too.
135    const read = await attempt(`deploys ${namespace}`, async () => ({ newest: await readDeployOutcomes(env, namespace) }));
136    if (!read) continue;
137    deploysRead++;
138    if (read.newest) out.push(...deployFindings(namespace, read.newest));
139  }
140  if (deploysRead === DEPLOY_ROSTER.length) ran.add("deploys");
141
142  // Canon documents a merged pull request left citing a removed, renamed or changed file
143  // (src/stale-canon.ts, improvement item 23). A digest finding: it posts no job, and it
144  // clears when the document is next written or stops citing the path.
145  const staleCanon = await attempt("stale canon", () => readStaleCanon(env, now));
146  if (staleCanon) {
147    if (staleCanon.complete) ran.add("stale canon");
148    out.push(...staleCanon.findings);
149    logEvent("log", "WATCHER_STALE_CANON", { message: `WATCHER_STALE_CANON ${staleCanon.note}` });
150  }
151
152  // The site configuration, read each pass (src/ops-sites.ts). Unreadable, every site
153  // check below cannot run this pass: nothing is probed on a guess, and an open site
154  // finding is not cleared on no evidence.
155  const config = await attempt("site config", () => readSiteConfig(env.DB));
156  const sites: OpsSite[] | null = config ? sitesFrom(config) : null;
157
158  // The configuration against the registered namespaces, both ways.
159  const registered = config
160    ? await attempt("site map", async () => {
161        const { results } = await env.DB.prepare("SELECT namespace FROM namespaces ORDER BY namespace").all<{ namespace: string }>();
162        return (results ?? []).map((r) => r.namespace);
163      })
164    : null;
165  if (config && registered) {
166    ran.add("site map");
167    observed.siteMap = siteMapDrift(registered, config);
168    out.push(...siteMapFindings(observed.siteMap));
169  }
170
171  // The App's repos against the namespace mapping. Read whole or not at all: a failed
172  // or empty read is its own finding and leaves the check un-run (src/unmapped-repos.ts).
173  try {
174    const repoMap = await readRepoMap(env);
175    ran.add("repo map");
176    out.push(...unmappedRepoFindings(repoMap));
177  } catch (err) {
178    const reason = err instanceof Error ? err.message : String(err);
179    logEvent("error", "WATCHER_READ_FAILED", { message: `WATCHER_READ_FAILED repo map: ${reason}` });
180    out.push(repoMapUnreadableFinding(reason));
181  }
182
183  // Every configured site, probed in turn. With none configured nothing is probed and
184  // the check has nothing to judge. A probe that gets no answer is a result, not a
185  // failure of the check; the check fails only if probing itself throws.
186  const probes = sites
187    ? await attempt("site probes", async () => {
188        const results: SiteProbe[] = [];
189        for (const site of sites) results.push(await probeSite(site, fetchImpl, now, health));
190        return results;
191      })
192    : null;
193  // The previous pass's rings, for "down on two probes in a row". A snapshot that is
194  // absent or unreadable (readSnapshot logs it) has no previous probe to compare; a
195  // KV read that throws means the check cannot judge, so it does not count as run and
196  // an open site-down job is not cleared on no evidence.
197  const prev = await attempt("previous snapshot", async () => ({ snapshot: await readSnapshot(env) }));
198  if (probes) {
199    observed.probes = probes;
200    if (prev) {
201      ran.add("site probes");
202      out.push(...siteDownFindings(prev.snapshot, probes));
203    }
204  }
205
206  // The live checks (src/live-checks.ts): what capsid/policy/live-checks.md says to
207  // look at, read through the probes this pass just made. No document, no check. A site
208  // that could not be read this pass leaves the check un-run, so an open finding is not
209  // cleared on no evidence.
210  if (sites && probes) {
211    const live = await attempt("live checks", async () => {
212      const config = await readLiveConfig(env.DB);
213      const analytics = async () => {
214        const credentials = cloudflareCredentials(env);
215        if (!credentials.ok) throw new Error(credentials.reason);
216        return readWebAnalyticsSites(fetchImpl, credentials.token, credentials.account);
217      };
218      return liveChecks(config, sites, probes, { fetchImpl, head: (namespace) => defaultBranchHead(env, namespace), analytics }, now);
219    });
220    if (live) {
221      if (live.ran) ran.add("live checks");
222      out.push(...live.findings.map(liveFinding));
223      // The count beside the verdict: no findings over zero pages read is not a clean bill.
224      logEvent("log", "WATCHER_LIVE", { message: `WATCHER_LIVE pages ${live.summary.pages_read}/${live.summary.pages_expected} shas ${live.summary.shas_read}/${live.summary.shas_expected} analytics ${live.summary.analytics_read}/${live.summary.analytics_expected} findings ${live.findings.length}` });
225    }
226  }
227
228  // The sites that opted into Capsid calling their operator API (src/site-watch.ts): the
229  // repair each failing health check maps to, the weekly tools, and the red and green
230  // mail. A site whose health could not be read leaves the check un-run, so its open
231  // finding is not cleared on no evidence.
232  if (config && config.some((c) => c.origin && c.operator)) {
233    const repairs = await attempt("site repairs", () => watchSites(env, config, WATCHER_ACTOR, fetchImpl, now));
234    if (repairs) {
235      if (repairs.complete) ran.add("site repairs");
236      out.push(...repairs.findings.map((f) => finding(f.namespace, f.fingerprint, f.headline, f.evidence)));
237      if (repairs.lines.length) logEvent("log", "WATCHER_SITE_REPAIRS", { message: `WATCHER_SITE_REPAIRS ${repairs.lines.join(" | ")}` });
238    }
239  } else if (config) {
240    // No site opted in: nothing to judge, so nothing open can be stale.
241    ran.add("site repairs");
242  }
243
244  // Cloudflare's view of each site: deploys and hourly errors (src/ops-cloudflare.ts).
245  // With no token nothing is fetched, each site says why, and the check is not run.
246  // With no site configured there is nothing to read, and nothing is fetched.
247  if (sites && sites.length > 0) {
248    const cf = await attempt("cloudflare", () => readCloudflare(env, sites, fetchImpl, now));
249    if (cf) {
250      observed.cloudflare = cf.bySite;
251      if (cf.ran) ran.add("cloudflare");
252      out.push(...siteErrorFindings(sites, cf.bySite));
253    } else {
254      observed.cloudflare = Object.fromEntries(
255        sites.map((s) => [s.namespace, { state: "error", reason: "the Cloudflare read threw this pass (WATCHER_READ_FAILED cloudflare in the log)" } as const])
256      );
257    }
258  }
259
260  // The secret names of each opted-in site's Worker (src/secret-presence.ts), for the
261  // Portal's convergence view: names only, set or not, read here because no Portal request
262  // calls Cloudflare. Each site's read stands alone and stores its own reason on failure.
263  for (const site of config ?? []) {
264    if (!site.origin || !site.operator || site.operator.secrets.length === 0 || site.platform !== "cloudflare") continue;
265    const read_at = now.toISOString();
266    const expected = site.operator.secrets;
267    const cfState = observed.cloudflare?.[site.namespace];
268    const script = site.script ?? (cfState?.state === "ok" ? cfState.script : null);
269    const credentials = cloudflareCredentials(env);
270    const stored = !credentials.ok
271      ? { state: "error" as const, script, rows: [], reason: credentials.reason, read_at }
272      : !script
273        ? { state: "error" as const, script: null, rows: [], reason: "the Worker script is not known: name it on the site's row, or let the Cloudflare read resolve it", read_at }
274        : await readSecretNames(fetchImpl, credentials.token, credentials.account, script).then(
275            (names) => ({ state: "ok" as const, script, rows: secretPresence(expected, names), reason: null, read_at }),
276            (err: unknown) => ({ state: "error" as const, script, rows: [], reason: (err instanceof Error ? err.message : String(err)).slice(0, 300), read_at })
277          );
278    await attempt(`secret names ${site.namespace}`, () => storeSecrets(env, site.namespace, stored));
279  }
280
281  // Agents acting out of their usual pattern, from audit_log (src/anomaly.ts, OWASP item
282  // 6). A finding only, never a suspension (Dustin, D4 of 2026-10-03).
283  // The hook receiver's tool-use rows, on their own series, rule and check (D1 of
284  // 2026-10-03), so a failed read of one never holds the other's findings open. The two
285  // reads are independent and run together.
286  const [actionHours, toolHours] = await Promise.all([
287    attempt("agent anomalies", () => readActionHours(env.DB, now, WATCHER_ACTOR)),
288    attempt("tool anomalies", () => readToolHours(env.DB, now, WATCHER_ACTOR)),
289  ]);
290  if (actionHours) {
291    ran.add("agent anomalies");
292    out.push(...anomalyFindings(actionHours, now));
293  }
294  if (toolHours) {
295    ran.add("tool anomalies");
296    out.push(...toolRateFindings(toolHours, now));
297  }
298
299  // Each configured site's domain registration, by RDAP (src/domain-expiry.ts): a
300  // finding 30 and 7 days before it expires. The check counts as run only when every
301  // domain was read, so a failed read never clears an open finding.
302  if (config) {
303    const domains = [...new Set(config.map((c) => (c.origin ? registrableDomain(c.origin) : null)).filter((d): d is string => d !== null))].sort();
304    const read = await attempt("domains", () => readExpiries(env.APP_KV, domains, fetchImpl, now));
305    if (read) {
306      out.push(...expiryFindings(read.expiries, now));
307      if (read.failed.length === 0) ran.add("domains");
308      for (const f of read.failed) logEvent("warn", "WATCHER_DOMAIN_UNREAD", { message: `${f.domain}: ${f.error}` });
309    }
310  }
311
312  // The configured packages, for the Portal's Packages view (src/ops-packages.ts).
313  // Read, never judged: no finding comes from a package. With none configured nothing
314  // is fetched; each package's sources fail on their own inside readPackage.
315  const packageConfig = await attempt("package config", () => readPackageConfig(env.DB));
316  if (packageConfig) {
317    const read = await attempt("packages", async () => {
318      const results: PackageSnapshot[] = [];
319      for (const cfg of packageConfig) results.push(await readPackage(env, cfg, fetchImpl, now));
320      return results;
321    });
322    if (read) observed.packages = read;
323  }
324
325  // The manual steps blocked jobs name with a "check:" line (src/manual-steps.ts): after
326  // the secret names above, so a secret set since the last pass is seen on this one. Read,
327  // never judged: no finding comes from a step, and the counts go to stale:steps for the
328  // inbox and the stale view.
329  const steps = await attempt("manual steps", () =>
330    checkManualSteps(
331      env,
332      now,
333      liveStepReaders(env, {
334        npmVersions: (name) => npmVersions(fetchImpl, name),
335        accessAppNames: async () => {
336          const credentials = cloudflareCredentials(env);
337          if (!credentials.ok) throw new Error(credentials.reason);
338          return readAccessAppNames(fetchImpl, credentials.token, credentials.account);
339        },
340      })
341    )
342  );
343  if (steps && steps.seen.length > 0) logEvent("log", "WATCHER_STEPS_SEEN", { message: `WATCHER_STEPS_SEEN ${steps.note}` });
344
345  return { findings: out, ran, observed };
346}