trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRunsg rEventsFlows

Link health, probe and flow actions between forges (limit of #9451) #9564

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/9561-link-health-probe-and-flow-actions-betwe → main 3 commits fork 4 threads · 4 unresolved

  • Merged. scout/9561-link-health-probe-and-flow-actions-betwe was merged into main as c53b795.

Review

Sign in to review this change.
12 files changed +399 −20
docs/design/areas/flows.md added +3
@@ -0,0 +1,3 @@
1 # Flows
2
3 No new component class. A flow line is the existing `row` with one `badge`: its text is the line's state ("2 changes behind", "up to date", "not read lately") and its tone follows the link's health (`attn` for news or not read lately, `danger` for diverged or unreadable, default otherwise). An action under the *Flows in* card is the existing secondary `btn`.
docs/features/flows.md +7 −1
@@ -8,4 +8,10 @@
88 - **Bounded.** A run asks four packages at a time within 30 s total; a repository has one run at a time; one identity gets four runs a minute. Followed and upstream URLs are shown without userinfo, query or fragment.
99 - **Pins.** Only exact floors (`1.2.3`, `^1.2.3`, `~1.2.3`) are compared; tags and URLs are skipped. Names are validated before they reach a URL.
1010
11 Left out: the CHANGELOG sections between pin and newest (dropped until a lookup can read one within a byte bound); a schedule and the issue the crew files from a report; advisories and code scanning (and the release block on a new high finding); acting on a line in place (update, follow, send back); GitHub release tags; the all-repositories admin view; "last moved" for forks. Each is its own pull on top of `dep_latest` and `readFlows`.
11 - **Link health (#9561).** Every line carries a health badge: healthy, news, diverged, unreadable, not read lately (no comparison in 90 minutes) or paused (`followHealth`, `src/flows/view.ts`). A mirror's own origin shows as a line from the head reading in `origin-ahead.md`.
12 - **Probe.** The clock already compares heads (`checkDue` for the upstream and sources, `runAheadReads` for an origin); an admin's *Compare heads now* (`POST /:owner/:name/flows/probe`, `src/flows/probe.ts`) runs the same reads on ask: admin only, never an agent, one at a time, four a minute.
13 - **Bounded.** A probe reads at most 8 enabled sources, four at a time, and answers within 30 s; it announces itself (`repo.links_probed`, readers of the repository) so an open page offers the refresh. An origin line is stale on the same 90-minute rule; every origin reading records when it was taken.
14 - **Terminal.** The Flows screen shows the same lines with health tones and the same gates: `c` compares heads (admin), `p` pulls in what's new (writer, `takeFromOrigin`), `o` names the page that offers a pull to the parent. A guest sees only what they may read.
15 - **Actions.** A writer of a mirror with news sees *Pull in what's new* (the existing `sync/take`, which refuses a diverged mirror); a writer of a fork sees *Offer what's useful* (the existing pull-to-parent page).
16
17 Left out: the CHANGELOG sections between pin and newest (dropped until a lookup can read one within a byte bound); a schedule and the issue the crew files from a report; advisories and code scanning (and the release block on a new high finding); per-line buttons (actions are listed under the columns); counting a fork's commits ahead of its parent; GitHub release tags; the all-repositories admin view; "last moved" for forks. Each is its own pull on top of `dep_latest` and `readFlows`.
src/core/event_audience.ts +1
@@ -139,6 +139,7 @@ const READER_KINDS: ReadonlySet<string> = new Set([
139139 "repo.created",
140140 "repo.vision_kept",
141141 "repo.forked",
142 "repo.links_probed",
142143 "repo.detached",
143144 "repo.deleted",
144145 "label.updated",
src/flows/probe.ts added +82
@@ -0,0 +1,82 @@
1 /**
2 * The head-comparison probe (#9561): one door that re-reads every link a
3 * repository follows — its upstream, its sources, its origin — so the flows
4 * view shows a fresh health for each. The clock already runs the same reads
5 * (`checkDue`, `runAheadReads`); this is the same read, on a person's ask.
6 * Admin only, never an agent, one at a time per repository, a few a minute.
7 */
8
9 import { checkSource, checkUpstream } from "../actions/upstream_follow.ts";
10 import type { App } from "../app.ts";
11 import { getFollow } from "../core/upstream_follow.ts";
12 import { listSources, type FollowSource } from "../core/follow_sources.ts";
13 import { RateLimiter } from "../core/ratelimit.ts";
14 import type { Repo } from "../core/repos.ts";
15 import type { User } from "../core/users.ts";
16 import { readAhead } from "../origins/ahead.ts";
17 import { mayStartRadar } from "./start.ts";
18
19 export type Probed = { ok: true; read: number } | { ok: false; why: string };
20
21 /** The reads one probe makes; tests stand in for the network. */
22 export interface Reads {
23 readonly origin: (app: App, repo: Repo) => Promise<boolean>;
24 readonly upstream: (app: App, repo: Repo) => Promise<boolean>;
25 readonly source: (app: App, repo: Repo, source: FollowSource) => Promise<boolean>;
26 }
27
28 const realReads: Reads = {
29 origin: async (app, repo) => (await readAhead(app, repo)) !== undefined,
30 upstream: async (app, repo) => (await checkUpstream(app, repo)).ok,
31 source: async (app, repo, source) => (await checkSource(app, repo, source)).ok,
32 };
33
34 /** Sources compared in one probe, how many at once, and the total time, as the radar's run is held. */
35 export const MAX_SOURCES = 8;
36 export const LANES = 4;
37 export const DEADLINE_MS = 30_000;
38
39 const running = new Set<string>();
40 const defaultLimiter = new RateLimiter({ perMinute: 4, adminPerMinute: 4, enabled: true });
41
42 export async function probeLinks(
sentinelcrew outdated

Sequential, unbounded network reads over every source. Add a total time budget and a cap, as the radar run has.
43 app: App,
44 repo: Repo,
45 user: User | undefined,
46 limiter: RateLimiter = defaultLimiter,
47 reads: Reads = realReads,
48 deadlineMs = DEADLINE_MS,
49 ): Promise<Probed> {
50 if (!mayStartRadar(app, repo, user)) return { ok: false, why: "Only an admin of this repository can do this." };
51 if (running.has(repo.id)) return { ok: false, why: "A comparison is already running for this repository." };
52 const verdict = limiter.take(`flows-probe:${user.id}`, "rw");
53 if (!verdict.allowed) return { ok: false, why: `Too many comparisons: try again in ${verdict.retryAfter}s.` };
54 running.add(repo.id);
55 try {
56 const stopAt = Date.now() + deadlineMs;
57 let read = 0;
58 const jobs: (() => Promise<boolean>)[] = [];
59 if (repo.origin !== null) jobs.push(() => reads.origin(app, repo));
60 else {
61 if (getFollow(app.db, repo.id)?.enabled === true) jobs.push(() => reads.upstream(app, repo));
62 for (const s of listSources(app.db, repo.id).filter((x) => x.enabled).slice(0, MAX_SOURCES))
63 jobs.push(() => reads.source(app, repo, s));
64 }
65 // A few at a time; none starts once the deadline has passed, and a read still running
66 // at the deadline is left to its own timeout while the probe answers.
67 const lanes = Array.from({ length: Math.min(LANES, jobs.length) }, async () => {
68 for (let job = jobs.shift(); job !== undefined && Date.now() < stopAt; job = jobs.shift()) {
69 try {
70 if (await job()) read++;
71 } catch (error) {
72 app.log.warn("link probe failed", { repo: repo.id, error: String(error) });
73 }
74 }
75 });
76 await Promise.race([Promise.all(lanes), Bun.sleep(deadlineMs)]);
77 app.bus.append({ kind: "repo.links_probed", repoId: repo.id, payload: {} });
78 return { ok: true, read };
79 } finally {
80 running.delete(repo.id);
81 }
82 }
src/flows/view.ts +51 −4
@@ -6,8 +6,23 @@
66 */
77
88 import type { Database } from "bun:sqlite";
9 import { canRead, getRepoById, visibleForks, type Repo } from "../core/repos.ts";
9 import { getCursor } from "../core/origins.ts";
10 import { canRead, canWrite, getRepoById, visibleForks, type Repo } from "../core/repos.ts";
1011 import type { User } from "../core/users.ts";
12 import { aheadWords, parseAhead } from "../origins/ahead.ts";
13
14 /** A followed link is stale when its last head comparison is older than this (three follow periods). */
15 export const STALE_MS = 3 * 30 * 60_000;
16
17 /** How a link stands: `ok` level, `behind` has news, `diverged` both moved, `unread` could not be read, `stale` not read lately, `off` not following, `none` nothing to compare. */
18 export type Health = "ok" | "behind" | "diverged" | "unread" | "stale" | "off" | "none";
19
20 /** One thing a person may do about a link: a same-origin POST, or a link to a page. */
21 export interface FlowAction {
22 readonly label: string;
23 readonly href: string;
24 readonly post: boolean;
25 }
1126
1227 export interface FlowLine {
1328 readonly kind: "origin" | "source" | "upstream" | "fork" | "mirror";
@@ -18,6 +33,8 @@ export interface FlowLine {
1833 /** When it last moved (ISO), or null. */
1934 readonly at: string | null;
2035 readonly problem: boolean;
36 readonly health: Health;
37 readonly action?: FlowAction;
2138 }
2239
2340 export interface FlowView {
@@ -47,13 +64,28 @@ function behindWords(behind: number, error: string): string {
4764 return behind === 0 ? "up to date" : `${behind} ${behind === 1 ? "change" : "changes"} behind`;
4865 }
4966
50 export function readFlows(db: Database, repo: Repo, viewer: User | undefined): FlowView {
67 /** Health of a followed link from its last comparison; `now` is injectable for tests. */
68 export function followHealth(
69 r: { enabled: boolean; checkedAt: string | null; behind: number; error: string },
70 now: number,
71 ): Health {
72 if (!r.enabled) return "off";
73 if (r.error !== "") return "unread";
74 if (r.checkedAt === null || now - Date.parse(r.checkedAt) > STALE_MS) return "stale";
75 return r.behind > 0 ? "behind" : "ok";
76 }
77
78 export function readFlows(db: Database, repo: Repo, viewer: User | undefined, now = Date.now()): FlowView {
79 const writer = viewer !== undefined && viewer.kind !== "agent" && canWrite(db, repo, viewer);
5180 const inbound: FlowLine[] = [];
5281 const outbound: FlowLine[] = [];
5382 if (repo.forkParentId !== null) {
5483 const parent = getRepoById(db, repo.forkParentId);
5584 if (parent && canRead(db, parent, viewer)) {
56 inbound.push({ kind: "origin", label: `${parent.ownerName}/${parent.name}`, href: base(parent), state: "forked from", at: null, problem: false });
85 inbound.push({
86 kind: "origin", label: `${parent.ownerName}/${parent.name}`, href: base(parent), state: "forked from", at: null, problem: false, health: "none",
87 ...(writer ? { action: { label: `Offer what's useful to ${parent.ownerName}/${parent.name}`, href: `${base(repo)}/pulls/new`, post: false } } : {}),
88 });
5789 }
5890 }
5991 const up = db
@@ -64,6 +96,7 @@ export function readFlows(db: Database, repo: Repo, viewer: User | undefined): F
6496 if (up) {
6597 inbound.push({
6698 kind: "upstream", label: `${plainUrl(up.url)} (${up.branch})`, href: "", at: up.checked_at, problem: up.error !== "",
99 health: followHealth({ enabled: up.enabled !== 0, checkedAt: up.checked_at, behind: up.behind, error: up.error }, now),
67100 state: up.enabled === 0 ? "not following" : behindWords(up.behind, up.error),
68101 });
69102 }
@@ -74,11 +107,24 @@ export function readFlows(db: Database, repo: Repo, viewer: User | undefined): F
74107 .all(repo.id)) {
75108 inbound.push({
76109 kind: "source", label: `${plainUrl(s.url)} (${s.branch})`, href: "", at: s.checked_at, problem: s.error !== "",
110 health: followHealth({ enabled: s.enabled !== 0, checkedAt: s.checked_at, behind: s.behind, error: s.error }, now),
77111 state: s.enabled === 0 ? "not following" : behindWords(s.behind, s.error),
78112 });
79113 }
114 if (repo.origin !== null) {
115 const ahead = parseAhead(getCursor(db, repo.origin.originId, `repo:${repo.id}:ahead`));
116 const read = getCursor(db, repo.origin.originId, `repo:${repo.id}:ahead-read`);
117 const seenAt = ahead?.at ?? read ?? null;
118 const old = seenAt === null || now - Date.parse(seenAt) > STALE_MS;
119 const health: Health = old ? "stale" : ahead !== undefined ? (ahead.diverged ? "diverged" : "behind") : "ok";
120 inbound.push({
121 kind: "origin", label: "its origin", href: "", at: seenAt, problem: ahead?.diverged === true, health,
122 state: seenAt === null ? "not read yet" : old ? "not read lately" : ahead === undefined ? "up to date" : aheadWords(ahead),
123 ...(writer && health === "behind" ? { action: { label: "Pull in what's new", href: `${base(repo)}/sync/take`, post: true } } : {}),
sentinelcrew outdated

Origin health ignores age: any read cursor gives ok. Apply STALE_MS here too.
124 });
125 }
80126 for (const f of visibleForks(db, repo.id, viewer)) {
81 outbound.push({ kind: "fork", label: `${f.ownerName}/${f.name}`, href: base(f), state: "fork", at: null, problem: false });
127 outbound.push({ kind: "fork", label: `${f.ownerName}/${f.name}`, href: base(f), state: "fork", at: null, problem: false, health: "none" });
82128 }
83129 const mirror = db
84130 .query<{ full_name: string; pushed_at: string | null; last_error: string }, [string]>(
@@ -89,6 +135,7 @@ export function readFlows(db: Database, repo: Repo, viewer: User | undefined): F
89135 outbound.push({
90136 kind: "mirror", label: mirror.full_name, href: "", at: mirror.pushed_at, problem: mirror.last_error !== "",
91137 state: mirror.last_error !== "" ? "last push failed" : "mirrored",
138 health: mirror.last_error !== "" ? "unread" : "ok",
92139 });
93140 }
94141 return { inbound, outbound };
src/http/server.ts +2 −1
@@ -151,7 +151,7 @@ import { reconnectSubmit } from "../web/pages/frozen.ts";
151151 import { activityAtom, changelogAtom } from "../web/feeds.ts";
152152 import { auditCsv, auditPage, auditProbeSubmit } from "../web/pages/audit.ts";
153153 import { apiPage, appApiPage } from "../web/pages/api.ts";
154 import { flowsCheck, flowsPage } from "../web/pages/flows.ts";
154 import { flowsCheck, flowsPage, flowsProbe } from "../web/pages/flows.ts";
155155 import { betAct, betPage, betPropose, betsPage } from "../web/pages/bets.ts";
156156 import { changelogJson, changelogPage } from "../web/pages/changelog.ts";
157157 import { releaseHoldSubmit, liftSubmit, releasesPage, rollbackSubmit } from "../web/pages/releases.ts";
@@ -1144,6 +1144,7 @@ export function buildRouter(): Router<Handler> {
11441144 router.add("POST", "/:owner/:name/issues/:number/first", firstSubmit);
11451145 router.add("GET", "/:owner/:name/flows", flowsPage);
11461146 router.add("POST", "/:owner/:name/flows/check", flowsCheck);
1147 router.add("POST", "/:owner/:name/flows/probe", flowsProbe);
11471148 router.add("GET", "/:owner/:name/bets", betsPage);
11481149 router.add("POST", "/:owner/:name/bets/propose", betPropose);
11491150 router.add("GET", "/:owner/:name/bets/:id", betPage);
src/origins/ahead.ts +4 −3
@@ -50,10 +50,10 @@ const aheadKey = (repoId: string): string => `repo:${repoId}:ahead`;
5050 /** The last reading of a mirror, or undefined when there is none or it is level. */
5151 export function aheadOf(app: App, repo: Repo): Ahead | undefined {
5252 if (repo.origin === null) return undefined;
53 return parse(getCursor(app.db, repo.origin.originId, aheadKey(repo.id)));
53 return parseAhead(getCursor(app.db, repo.origin.originId, aheadKey(repo.id)));
5454 }
5555
56 function parse(value: string | undefined): Ahead | undefined {
56 export function parseAhead(value: string | undefined): Ahead | undefined {
5757 const [count, tip, diverged, at] = (value ?? "").split(" ");
5858 const n = Number(count);
5959 if (!Number.isInteger(n) || n < 1 || !isSafeOid(tip ?? "") || at === undefined) return undefined;
@@ -87,6 +87,7 @@ async function read(app: App, repo: Repo): Promise<Ahead | undefined> {
8787 const fetched = await fetchRefFrom(dir, link.cloneUrl, `refs/heads/${repo.defaultBranch}`, authEnv);
8888 if (!fetched.ok) return undefined;
8989 const tip = fetched.value;
90 setCursor(app.db, link.origin.id, `repo:${repo.id}:ahead-read`, new Date().toISOString());
9091 const count = tip === main ? 0 : await aheadCount(dir, main, tip);
9192 const key = aheadKey(repo.id);
9293 if (count === 0) {
@@ -185,7 +186,7 @@ export function mirrorsBehind(app: App, viewer: User | undefined): { repo: Repo;
185186 const out: { repo: Repo; ahead: Ahead }[] = [];
186187 for (const row of rows) {
187188 const repo = getRepoById(app.db, row.id);
188 const ahead = parse(row.value);
189 const ahead = parseAhead(row.value);
189190 if (repo !== undefined && ahead !== undefined && canRead(app.db, repo, viewer)) out.push({ repo, ahead });
190191 }
191192 return out;
src/tui/screens/flows.ts +39 −5
@@ -5,20 +5,24 @@
55 * decides nothing of its own. Access is read at every draw.
66 */
77
8 import { takeFromOrigin } from "../../actions/origins.ts";
89 import { canRead, type Repo } from "../../core/repos.ts";
910 import { readRadar, type DepLine } from "../../flows/radar.ts";
1011 import { pinsOf } from "../../flows/run.ts";
12 import { probeLinks } from "../../flows/probe.ts";
1113 import { mayStartRadar, startRadar } from "../../flows/start.ts";
12 import { readFlows, type FlowLine } from "../../flows/view.ts";
14 import { readFlows, type FlowLine, type Health } from "../../flows/view.ts";
1315 import type { Ctx, Screen } from "../app/app.ts";
1416 import { binding } from "../app/keys.ts";
1517 import type { Session } from "../session.ts";
16 import { ListState, drawList, runAsync, type ListRow } from "./common.ts";
18 import { ListState, drawList, live, runAsync, type ListRow } from "./common.ts";
19
20 const HEALTH_TONE: Partial<Record<Health, "danger" | "attn">> = { diverged: "danger", unread: "danger", behind: "attn", stale: "attn" };
1721
1822 const flowRow = (side: string) => (l: FlowLine): ListRow => ({
1923 title: `${side} ${l.label}`,
2024 secondary: l.state,
21 ...(l.problem ? { tone: "danger" as const } : {}),
25 ...(l.problem ? { tone: "danger" as const } : HEALTH_TONE[l.health] ? { tone: HEALTH_TONE[l.health]! } : {}),
2226 meta: l.at === null ? "never moved" : l.at.slice(0, 16).replace("T", " "),
2327 });
2428
@@ -38,18 +42,48 @@ export function flowsScreen(f: Session, ctx: Ctx, repo: Repo): Screen {
3842 const view = readFlows(f.db, repo, f.user);
3943 return [...view.inbound.map(flowRow("in ")), ...deps.map(depRow), ...view.outbound.map(flowRow("out"))];
4044 };
41 const load = async (): Promise<void> => {
45 /** The actions the page would offer this person: the same reads decide. */
46 const actions = () => (visible() ? readFlows(f.db, repo, f.user).inbound.flatMap((l) => (l.action ? [l.action] : [])) : []);
47 const load = async (signal?: AbortSignal): Promise<void> => {
48 if (signal) live(f, signal, { repoId: repo.id, kinds: ["repo.links_probed", "repo.upstream_checked"] }, () => ctx.invalidate());
4249 deps = visible() ? readRadar(f.db, repo.id, await pinsOf(f.app, repo)) : [];
4350 ctx.invalidate();
4451 };
4552 return {
4653 title: "Flows",
47 keys: [binding(["j", "k"], "move cursor"), ...(mayStartRadar(f.app, repo, f.user) ? [binding(["r"], "read upstream now")] : []), binding(["esc"], "back")],
54 keys: [binding(["j", "k"], "move cursor"), ...(mayStartRadar(f.app, repo, f.user) ? [binding(["r"], "read upstream now"), binding(["c"], "compare heads now")] : []), ...(actions().some((a) => a.post) ? [binding(["p"], "pull in what's new")] : []), ...(actions().some((a) => !a.post) ? [binding(["o"], "offer what's useful")] : []), binding(["esc"], "back")],
4855 load,
4956 render(area, s) {
5057 drawList(s, area, ctx.theme.c, state, rows());
5158 },
5259 key(e) {
60 if (e.ctrl || e.alt) return state.key(e, rows().length);
61 if (e.key === "c" && mayStartRadar(f.app, repo, f.user)) {
62 runAsync(ctx, async () => {
63 const probed = await probeLinks(f.app, repo, f.user);
64 if (probed.ok) ctx.toast(`Compared ${probed.read} links.`);
65 else ctx.toast(probed.why, "danger");
66 await load();
67 });
68 return true;
69 }
70 if (e.key === "p" && actions().some((a) => a.post)) {
71 runAsync(ctx, async () => {
72 const done = await takeFromOrigin(f.app, f.user, repo);
73 if (!done.ok) ctx.toast("that could not be taken", "danger");
74 else if (done.value.kind === "conflict") ctx.toast("The mirror holds commits the origin lacks: nothing moved.", "danger");
75 else if (done.value.kind === "unread") ctx.toast("The origin could not be read just now.", "danger");
76 else ctx.toast("Took what's new from the origin.");
77 await load();
78 });
79 return true;
80 }
81 if (e.key === "o") {
82 const offer = actions().find((a) => !a.post);
83 if (!offer) return false;
84 ctx.toast(`Open ${offer.href} on the web to offer it.`);
85 return true;
86 }
5387 if (e.key === "r") {
5488 runAsync(ctx, async () => {
5589 const started = await startRadar(f.app, repo, f.user);
src/web/live.ts +1 −1
@@ -304,7 +304,7 @@ export const liveClientJs = /* js */ `
304304 'label.created', 'label.updated', 'label.deleted',
305305 'run.queued', 'run.started', 'run.finished', 'app.serving',
306306 'app.created', 'app.updated', 'app.deleted', 'app.stopped',
307 'repo.upstream_checked', 'repo.source_added', 'repo.source_removed', 'repo.upstream_follow', 'repo.following', 'limb.moved', 'limb.replan', 'limb.replan_taken_back', 'issue.assigned', 'pull.assigned', 'crew.added', 'crew.removed',
307 'repo.upstream_checked', 'repo.links_probed', 'repo.source_added', 'repo.source_removed', 'repo.upstream_follow', 'repo.following', 'limb.moved', 'limb.replan', 'limb.replan_taken_back', 'issue.assigned', 'pull.assigned', 'crew.added', 'crew.removed',
308308 ];
309309 var extra = (cfg.getAttribute('data-repos') || '').split(' ').filter(Boolean);
310310 var watching = [repo].concat(extra).filter(function (r, i, all) {
src/web/pages/flows.ts +31 −4
@@ -10,23 +10,34 @@
1010 import { notePhase, takeSessionTime } from "../../core/route_timing.ts";
1111 import { readRadar } from "../../flows/radar.ts";
1212 import { pinsOf } from "../../flows/run.ts";
13 import { probeLinks } from "../../flows/probe.ts";
1314 import { mayStartRadar, startRadar } from "../../flows/start.ts";
14 import { readFlows, type FlowLine } from "../../flows/view.ts";
15 import { readFlows, type FlowAction, type FlowLine, type Health } from "../../flows/view.ts";
1516 import type { Ctx } from "../../http/context.ts";
1617 import { crumbsFor, loadRepo, notFoundPage, pageHead, repoBase, repoHeader, repoPalette } from "../chrome.ts";
1718 import { readForm } from "../form.ts";
1819 import { html, type Html } from "../html.ts";
1920 import { flashTo, page } from "../layout.ts";
20 import { badge, btn, card, listCard, row, time } from "../ui.ts";
21 import { badge, btn, type Tone, card, listCard, row, time } from "../ui.ts";
2122
2223 const lineRow = (l: FlowLine): Html =>
2324 row({
2425 ...(l.href === "" ? {} : { href: l.href }),
2526 title: html`${l.label}`,
2627 meta: html`${l.at === null ? "never moved" : time(l.at)}`,
27 trail: badge(l.state, { tone: l.problem ? "danger" : "default" }),
28 trail: badge(l.state, { tone: TONE[l.health] }),
sentinelcrew outdated

Two badges per row. Use one status badge, and keep the extra words out of the tone map unless DESIGN.md names them.
2829 });
2930
31 /** A link's tone from its health: `attn` and `danger` are Tone values the badge already has. */
32 const TONE: Record<Health, Tone> = {
33 ok: "default", behind: "attn", diverged: "danger", unread: "danger", stale: "attn", off: "default", none: "default",
34 };
35
36 const actionForm = (a: FlowAction): Html =>
37 a.post
38 ? html`<form method="post" action="${a.href}" class="mt">${btn(a.label, { variant: "secondary" })}</form>`
39 : html`<p class="mt">${btn(a.label, { variant: "secondary", href: a.href })}</p>`;
40
3041 function side(title: string, lines: readonly Html[], empty: string): Html {
3142 return card(lines.length === 0 ? html`<p class="muted">${empty}</p>` : listCard([...lines]), { head: html`${title}`, pad: lines.length === 0 });
3243 }
@@ -40,6 +51,7 @@ export async function flowsPage(ctx: Ctx): Promise<Response> {
4051 const flows = readFlows(ctx.app.db, repo, ctx.auth.user);
4152 const deps = readRadar(ctx.app.db, repo.id, await pinsOf(ctx.app, repo));
4253 const admin = mayStartRadar(ctx.app, repo, ctx.auth.user);
54 const actions = flows.inbound.flatMap((l) => (l.action ? [actionForm(l.action)] : []));
4355 const depRows = deps.map((d) =>
4456 row({
4557 title: html`${d.name}`,
@@ -54,12 +66,14 @@ export async function flowsPage(ctx: Ctx): Promise<Response> {
5466 <div class="split">
5567 <div>
5668 ${side("Flows in", flows.inbound.map(lineRow), "Nothing flows in: no origin and no followed source.")}
69 ${actions}
5770 ${side("Packages", depRows, "No pinned packages.")}
5871 </div>
5972 <div>${side("Flows out", flows.outbound.map(lineRow), "Nothing flows out: no forks and no mirror.")}</div>
6073 </div>
6174 ${admin
62 ? html`<form method="post" action="${repoBase(repo)}/flows/check" class="mt">${btn("Read upstream releases now", { variant: "secondary" })}</form>`
75 ? html`<form method="post" action="${repoBase(repo)}/flows/probe" class="mt">${btn("Compare heads now", { variant: "secondary" })}</form>
76 <form method="post" action="${repoBase(repo)}/flows/check" class="mt">${btn("Read upstream releases now", { variant: "secondary" })}</form>`
6377 : null}`;
6478 const response = page({ ctx, title: `Flows · ${repo.name}`, crumbs: crumbsFor(repo), palette: repoPalette(ctx, repo), body });
6579 notePhase(ctx.request, "render", performance.now() - rendering);
@@ -78,3 +92,16 @@ export async function flowsCheck(ctx: Ctx): Promise<Response> {
7892 const n = started.read;
7993 return flashTo(ctx, back, { message: n === 0 ? "No exact pins to read." : `Read upstream for ${n} ${n === 1 ? "package" : "packages"}.` });
8094 }
95
96 // POST /:owner/:name/flows/probe
97 export async function flowsProbe(ctx: Ctx): Promise<Response> {
98 const repo = loadRepo(ctx);
99 if (!repo) return notFoundPage(ctx);
100 const form = await readForm(ctx, "rw");
101 if (form instanceof Response) return form;
102 const back = `${repoBase(repo)}/flows`;
103 const probed = await probeLinks(ctx.app, repo, ctx.auth.user);
104 if (!probed.ok) return flashTo(ctx, back, { tone: "danger", message: probed.why });
105 const n = probed.read;
106 return flashTo(ctx, back, { message: n === 0 ? "No link could be compared." : `Compared ${n} ${n === 1 ? "link" : "links"}.` });
107 }
test/flows-tui.test.ts +15
@@ -78,6 +78,21 @@ describe("flows in the terminal", () => {
7878 reader.stop();
7979 });
8080
81 test("compare heads is offered to an admin and not to a reader or a guest", async () => {
82 const admin = boot(ada);
83 await admin.open();
84 expect(admin.text()).toContain("compare heads now");
85 admin.stop();
86 for (const who of [eve, undefined]) {
87 const other = boot(who);
88 await other.open();
89 expect(other.text()).not.toContain("compare heads now");
90 await other.press("c");
91 expect(other.text()).not.toContain("Compared");
92 other.stop();
93 }
94 });
95
8196 test("a private repository shows a guest nothing", async () => {
8297 forge.db.query("UPDATE repo SET visibility = 'private' WHERE id = ?").run(repo.id);
8398 const guest = boot(undefined);
test/flows.test.ts +163 −1
@@ -11,8 +11,10 @@ import { readRadar, refreshRadar } from "../src/flows/radar.ts";
1111 import { compareSemver, gapOf, parseSemver } from "../src/flows/semver.ts";
1212 import { RateLimiter } from "../src/core/ratelimit.ts";
1313 import { getUserById, type User } from "../src/core/users.ts";
14 import { createRepo } from "../src/actions/repos.ts";
15 import { MAX_SOURCES, probeLinks, type Reads } from "../src/flows/probe.ts";
1416 import { startRadar } from "../src/flows/start.ts";
15 import { plainUrl, readFlows } from "../src/flows/view.ts";
17 import { followHealth, plainUrl, readFlows } from "../src/flows/view.ts";
1618 import { serve, type Servers } from "../src/http/server.ts";
1719 import { PREVIEW_ADMIN, PREVIEW_REPO } from "../src/spawn/preview-looks.ts";
1820 import { seedPreview } from "../src/spawn/preview-seed.ts";
@@ -180,3 +182,163 @@ describe("the page", () => {
180182 }
181183 });
182184 });
185
186 describe("link health and actions (#9561)", () => {
187 const repoNow = () => getRepo(app.db, PREVIEW_ADMIN, PREVIEW_REPO)!;
188 const admin = () => getUserById(app.db, app.db.query<{ id: string }, [string]>("SELECT id FROM user WHERE username = ?").get(PREVIEW_ADMIN)!.id)!;
189 const now = Date.parse("2026-11-08T12:00:00Z");
190
191 test("followHealth names each standing", () => {
192 const at = "2026-11-08T11:50:00Z";
193 expect(followHealth({ enabled: false, checkedAt: at, behind: 3, error: "" }, now)).toBe("off");
194 expect(followHealth({ enabled: true, checkedAt: at, behind: 0, error: "x" }, now)).toBe("unread");
195 expect(followHealth({ enabled: true, checkedAt: null, behind: 0, error: "" }, now)).toBe("stale");
196 expect(followHealth({ enabled: true, checkedAt: "2026-11-08T06:00:00Z", behind: 0, error: "" }, now)).toBe("stale");
197 expect(followHealth({ enabled: true, checkedAt: at, behind: 2, error: "" }, now)).toBe("behind");
198 expect(followHealth({ enabled: true, checkedAt: at, behind: 0, error: "" }, now)).toBe("ok");
199 });
200
201 test("a mirror behind its origin shows news and Pull in what's new, for a writer only", () => {
202 const repo = repoNow();
203 const mirrored = { ...repo, origin: { ...(repo.origin ?? {}), originId: "o9" } } as typeof repo;
204 app.db.exec("PRAGMA foreign_keys = OFF");
205 app.db.query("INSERT OR REPLACE INTO origin_cursor (origin_id, key, value) VALUES ('o9', ?1, ?2)").run(`repo:${repo.id}:ahead`, `3 ${"a".repeat(40)} 0 2026-11-08T11:00:00Z`);
206 const mine = readFlows(app.db, mirrored, admin(), now).inbound.find((l) => l.label === "its origin")!;
207 expect(mine.health).toBe("behind");
208 expect(mine.action).toEqual({ label: "Pull in what's new", href: `/${PREVIEW_ADMIN}/${PREVIEW_REPO}/sync/take`, post: true });
209 const guest = readFlows(app.db, mirrored, undefined, now).inbound.find((l) => l.label === "its origin")!;
210 expect(guest.health).toBe("behind");
211 expect(guest.action).toBeUndefined();
212 app.db.query(`UPDATE origin_cursor SET value = ?1 WHERE origin_id = 'o9'`).run(`3 ${"a".repeat(40)} 1 2026-11-08T11:00:00Z`);
213 const diverged = readFlows(app.db, mirrored, admin(), now).inbound.find((l) => l.label === "its origin")!;
214 expect(diverged.health).toBe("diverged");
215 expect(diverged.action).toBeUndefined();
216 app.db.query("DELETE FROM origin_cursor WHERE origin_id = 'o9'").run();
217 app.db.exec("PRAGMA foreign_keys = ON");
218 });
219
220 test("the probe is admin only and a cross-origin POST changes nothing", async () => {
221 const repo = repoNow();
222 const stranger = { ...admin(), id: "nobody", isAdmin: false } as User;
223 expect(await probeLinks(app, repo, stranger)).toEqual({ ok: false, why: "Only an admin of this repository can do this." });
224 expect(await probeLinks(app, repo, undefined)).toMatchObject({ ok: false });
225 const said = `u1|trevato|admin|${Date.now() + 60_000}`;
226 const headers = { "x-forge-identity": said, "x-forge-signature": createHmac("sha256", KEY).update(said).digest("base64url") };
227 const url = `${servers.url}/${PREVIEW_ADMIN}/${PREVIEW_REPO}/flows`;
228 const page = await (await fetch(url, { headers })).text();
229 expect(page).toContain("Compare heads now");
230 const post = await fetch(`${url}/probe`, { method: "POST", redirect: "manual", headers: { ...headers, origin: "https://evil.example" } });
231 expect(post.status).toBe(403);
232 });
233
sentinelcrew outdated

This returns early and asserts nothing when there is one repo. Build a real parent and fork.
234 test("a fork's writer is offered a pull to its parent; a guest is not", async () => {
235 const parent = repoNow();
236 const made = await createRepo(app, admin(), { name: "kid", owner: PREVIEW_ADMIN, visibility: "public", description: "" });
237 if (!made.ok) throw new Error(JSON.stringify(made.error));
238 app.db.query("UPDATE repo SET fork_parent_id = ?1 WHERE id = ?2").run(parent.id, made.value.id);
239 const fork = getRepo(app.db, PREVIEW_ADMIN, "kid")!;
240 const mine = readFlows(app.db, fork, admin(), now).inbound.find((l) => l.state === "forked from")!;
241 expect(mine.label).toBe(`${PREVIEW_ADMIN}/${PREVIEW_REPO}`);
242 expect(mine.action).toEqual({ label: `Offer what's useful to ${PREVIEW_ADMIN}/${PREVIEW_REPO}`, href: `/${PREVIEW_ADMIN}/kid/pulls/new`, post: false });
243 expect(readFlows(app.db, fork, undefined, now).inbound.find((l) => l.state === "forked from")!.action).toBeUndefined();
244 const agent = { ...admin(), kind: "agent" } as User;
245 expect(readFlows(app.db, fork, agent, now).inbound.find((l) => l.state === "forked from")!.action).toBeUndefined();
246 });
247
248 test("an origin line is ok when read lately, stale when read long ago, not read yet when never", () => {
249 const repo = repoNow();
250 const mirrored = { ...repo, origin: { ...(repo.origin ?? {}), originId: "o8" } } as typeof repo;
251 const line = () => readFlows(app.db, mirrored, undefined, now).inbound.find((l) => l.label === "its origin")!;
252 expect([line().health, line().state]).toEqual(["stale", "not read yet"]);
253 app.db.exec("PRAGMA foreign_keys = OFF");
254 const put = (at: string) => app.db.query("INSERT OR REPLACE INTO origin_cursor (origin_id, key, value) VALUES ('o8', ?1, ?2)").run(`repo:${repo.id}:ahead-read`, at);
255 put("2026-11-08T11:50:00Z");
256 expect([line().health, line().state]).toEqual(["ok", "up to date"]);
257 put("2026-11-08T06:00:00Z");
258 expect([line().health, line().state]).toEqual(["stale", "not read lately"]);
259 app.db.query("DELETE FROM origin_cursor WHERE origin_id = 'o8'").run();
260 app.db.exec("PRAGMA foreign_keys = ON");
261 });
262
263 describe("the probe", () => {
264 const reads = (log: string[], over: Partial<Reads> = {}): Reads => ({
265 origin: async () => (log.push("origin"), true),
266 upstream: async () => (log.push("upstream"), true),
267 source: async (_a, _r, s) => (log.push(`source:${s.id}`), true),
268 ...over,
269 });
270 const limiter = (n: number) => new RateLimiter({ perMinute: n, adminPerMinute: n, enabled: true });
271 const addSources = (repoId: string, n: number) => {
272 for (let i = 0; i < n; i++)
273 app.db.query("INSERT OR IGNORE INTO repo_source (id, repo_id, url, branch, by_id, enabled, created_at) VALUES (?1, ?2, 'https://example.com/p.git', 'main', 'u1', 1, '2026-11-03T00:00:00Z')").run(`p${i}`, repoId);
274 };
275
276 test("a successful probe reads the enabled sources and announces itself", async () => {
277 const repo = repoNow();
278 addSources(repo.id, 2);
279 const log: string[] = [];
280 const seen: string[] = [];
281 const off = app.bus.subscribe((e) => seen.push(e.kind));
282 const out = await probeLinks(app, repo, admin(), limiter(4), reads(log));
283 off();
284 expect(out.ok).toBe(true);
285 expect(log).toContain("source:p0");
286 expect(log).toContain("source:s1");
287 expect(seen).toContain("repo.links_probed");
288 });
289
290 test("an agent is refused and reads nothing", async () => {
291 const log: string[] = [];
292 const agent = { ...admin(), kind: "agent" } as User;
293 expect(await probeLinks(app, repoNow(), agent, limiter(4), reads(log))).toEqual({ ok: false, why: "Only an admin of this repository can do this." });
294 expect(log).toEqual([]);
295 });
296
297 test("one at a time, and four a minute", async () => {
298 const repo = repoNow();
299 let release!: () => void;
300 const gate = new Promise<void>((r) => (release = r));
301 const slow = reads([], { source: async () => { await gate; return true; } });
302 const lim = limiter(4);
303 const first = probeLinks(app, repo, admin(), lim, slow);
304 await Promise.resolve();
305 expect(await probeLinks(app, repo, admin(), lim, slow)).toEqual({ ok: false, why: "A comparison is already running for this repository." });
306 release();
307 expect((await first).ok).toBe(true);
308 const tight = limiter(1);
309 await probeLinks(app, repo, admin(), tight, reads([]));
310 const third = await probeLinks(app, repo, admin(), tight, reads([]));
311 expect(!third.ok && third.why).toMatch(/^Too many comparisons/);
312 });
313
314 test("it is capped in sources and bounded in time", async () => {
315 const repo = repoNow();
316 addSources(repo.id, 12);
317 const log: string[] = [];
318 await probeLinks(app, repo, admin(), limiter(4), reads(log));
319 expect(log.filter((l) => l.startsWith("source:")).length).toBeLessThanOrEqual(MAX_SOURCES);
320 const hang = reads([], { source: () => new Promise<boolean>(() => {}) });
321 const t0 = Date.now();
322 const out = await probeLinks(app, repo, admin(), limiter(4), hang, 60);
323 expect(out.ok).toBe(true);
324 expect(Date.now() - t0).toBeLessThan(2000);
325 });
326
327 test("a failing read does not stop the others", async () => {
328 const repo = repoNow();
329 const log: string[] = [];
330 const out = await probeLinks(app, repo, admin(), limiter(4), reads(log, { upstream: async () => { throw new Error("boom"); } }));
331 expect(out.ok).toBe(true);
332 expect(log.some((l) => l.startsWith("source:"))).toBe(true);
333 });
334
335 test("the route flashes the result for an admin", async () => {
336 const said = `u1|trevato|admin|${Date.now() + 60_000}`;
337 const headers = { "x-forge-identity": said, "x-forge-signature": createHmac("sha256", KEY).update(said).digest("base64url") };
338 const url = `${servers.url}/${PREVIEW_ADMIN}/${PREVIEW_REPO}/flows`;
339 const post = await fetch(`${url}/probe`, { method: "POST", redirect: "manual", headers: { ...headers, origin: servers.url } });
340 expect(post.status).toBe(303);
341 expect(post.headers.get("location")).toBe(`/${PREVIEW_ADMIN}/${PREVIEW_REPO}/flows`);
342 });
343 });
344 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs