Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions apps/ade-cli/src/bootstrap.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1138,8 +1138,8 @@ export async function createAdeRuntime(args: {
// `pr.listQueueStates` call over the local runtime fails with "is not
// callable". Mirror the desktop main-process wiring (see main.ts) so the PRs
// tab loads against the local runtime.
// Fan-out for the push publisher: PR merge-ready / checks-failing notifications
// are bridged here so the publisher never has to poll GitHub itself. Populated
// Fan-out for the push publisher: PR lifecycle/status notifications are
// bridged here so the publisher never has to poll GitHub itself. Populated
// by pushPublisherService.start() (declared below), so it stays empty and inert
// when push publishing is not running.
const pushPrNotificationSubscribers = new Set<(notification: PushPrNotification) => void>();
Expand Down
45 changes: 45 additions & 0 deletions apps/ade-cli/src/services/push/pushPublisherService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import {
parseHhMm,
shouldDeliverAlertForPrefs,
type AgentRunState,
type PushPrNotification,
} from "./pushPublisherService";

function run(overrides: Partial<AgentRunState>): AgentRunState {
Expand Down Expand Up @@ -226,6 +227,50 @@ describe("createPushPublisherService flush", () => {
publisher.dispose();
});

it("publishes PR lifecycle alerts into the aggregate Live Activity", async () => {
const { publisher, publish } = makeHarness();
let prCb: (event: PushPrNotification) => void = () => {
throw new Error("PR notification source was not attached");
};
publisher.attachSources("scope-pr", {
subscribePrNotifications: (cb) => {
prCb = cb;
return () => {};
},
resolveLaneName: (laneId: string) => laneId === "lane-42" ? "Mobile PR lane" : laneId,
});
await publisher.start();

prCb?.({ kind: "merged", prNumber: 42, prTitle: "Ship mobile PR view", laneId: "lane-42" });
await vi.advanceTimersByTimeAsync(2_500);

const payload = publish.mock.calls[0][0];
expect(payload.notifications[0]).toMatchObject({
title: "PR #42 merged",
body: "Ship mobile PR view",
deepLink: "ade://pr/42",
threadId: "pr-42",
});
expect(payload.liveActivity[0].contentState.prs[0]).toMatchObject({
id: "pr-42",
prNumber: 42,
title: "Ship mobile PR view",
phase: "merged",
lane: "Mobile PR lane",
});
expect(payload.liveActivity[0].phase).toBe("running");

await vi.advanceTimersByTimeAsync(45 * 60 * 1000 + 1_000);
const endPayload = publish.mock.calls.at(-1)?.[0];
expect(endPayload.liveActivity[0]).toMatchObject({
event: "end",
phase: "terminal",
});
expect(endPayload.liveActivity[0].contentState.prs).toEqual([]);

publisher.dispose();
});

it("carries actionable fields: category + sessionId/itemId on the alert, itemId on the waiting LA row, badge count", async () => {
const { publisher, publish, emit } = makeHarness();
await publisher.start();
Expand Down
169 changes: 131 additions & 38 deletions apps/ade-cli/src/services/push/pushPublisherService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,14 @@ export function resolvePushRelayStateFile(secretsDir: string): string {
export const APPROVAL_NOTIFICATION_CATEGORY = "ADE_APPROVAL";
const SHUTDOWN_PUBLISH_TIMEOUT_MS = 2_500;
const AGENT_RUNS_MAX = 3;
const PR_LIVE_ACTIVITY_MAX = 2;
const DETAIL_MAX_CHARS = 160;
/** Fixed lock-screen copy for failed runs — never leak error text to the widget. */
const FAILED_DETAIL = "Run failed";

const RUNNING_TTL_MS = 2 * 60 * 60 * 1000; // 2h for running/starting
const WAITING_TTL_MS = 24 * 60 * 60 * 1000; // 24h for waiting_for_*
const PR_LIVE_ACTIVITY_TTL_MS = 45 * 60 * 1000; // keep recent PR status visible, then age it out
const DEFAULT_FLUSH_DEBOUNCE_MS = 2_000;
const DEFAULT_PROMPT_FLUSH_MS = 150;
const PUBLISH_RETRY_MS = 30_000;
Expand Down Expand Up @@ -86,6 +88,15 @@ export type PushPrNotification = {
laneId: string | null;
};

export type PrLiveActivityState = {
id: string;
prNumber: number;
title: string;
phase: PrNotificationKind;
lane: string | null;
updatedAt: number;
};

type PendingAlert = {
sessionId: string | null;
dedupeKey: string;
Expand Down Expand Up @@ -243,6 +254,7 @@ function capDetail(detail: string | null | undefined): string | null {
export function buildAgentRunsContentState(
runs: AgentRunState[],
nowMs: number,
prs: PrLiveActivityState[] = [],
): {
updatedAt: number;
activeCount: number;
Expand All @@ -255,9 +267,18 @@ export function buildAgentRunsContentState(
detail: string | null;
itemId?: string;
}>;
prs: Array<{
id: string;
prNumber: number;
title: string;
phase: PrNotificationKind;
lane: string | null;
updatedAt: number;
}>;
} {
const activeCount = runs.filter((run) => isActivePhase(run.phase)).length;
const ordered = [...runs].sort((left, right) => right.lastActiveAt - left.lastActiveAt).slice(0, AGENT_RUNS_MAX);
const orderedPrs = [...prs].sort((left, right) => right.updatedAt - left.updatedAt).slice(0, PR_LIVE_ACTIVITY_MAX);
return {
updatedAt: Math.floor(nowMs / 1000),
activeCount,
Expand All @@ -272,6 +293,14 @@ export function buildAgentRunsContentState(
// screen render Approve/Deny intents against the pending item.
...(run.phase === "waiting_for_approval" && run.itemId ? { itemId: run.itemId } : {}),
})),
prs: orderedPrs.map((pr) => ({
id: pr.id,
prNumber: pr.prNumber,
title: pr.title,
phase: pr.phase,
lane: pr.lane,
updatedAt: Math.floor(pr.updatedAt / 1000),
})),
};
}

Expand Down Expand Up @@ -358,6 +387,7 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
const promptFlushMs = deps.promptFlushMs ?? DEFAULT_PROMPT_FLUSH_MS;

const runs = new Map<string, AgentRunState>();
const prActivities = new Map<string, PrLiveActivityState>();
let pendingAlerts: PendingAlert[] = [];
const lastAlertFingerprintByKey = new Map<string, string>();
let lastLiveActivityFingerprint: string | null = null;
Expand All @@ -371,6 +401,7 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
const lastSentBadgeByDevice = new Map<string, number>();

let flushTimer: NodeJS.Timeout | null = null;
let prExpiryTimer: NodeJS.Timeout | null = null;
let flushFireAt = 0;
let flushing = false;
let disposed = false;
Expand Down Expand Up @@ -469,6 +500,31 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
}
};

const prunePrActivities = (nowMs: number): void => {
for (const [id, pr] of prActivities) {
if (nowMs - pr.updatedAt > PR_LIVE_ACTIVITY_TTL_MS) {
prActivities.delete(id);
}
}
};

const schedulePrActivityExpiry = (nowMs: number): void => {
if (prExpiryTimer) {
clearTimeout(prExpiryTimer);
prExpiryTimer = null;
}
let nextExpiryMs = Number.POSITIVE_INFINITY;
for (const pr of prActivities.values()) {
nextExpiryMs = Math.min(nextExpiryMs, pr.updatedAt + PR_LIVE_ACTIVITY_TTL_MS);
}
if (!Number.isFinite(nextExpiryMs)) return;
const delayMs = Math.max(250, nextExpiryMs - nowMs + 250);
prExpiryTimer = setTimeout(() => {
prExpiryTimer = null;
scheduleFlush(true);
}, delayMs);
};

const resolveMissingMeta = async (): Promise<void> => {
for (const run of runs.values()) {
if (run.metaResolved) continue;
Expand Down Expand Up @@ -517,17 +573,21 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
): { item: PushRelayLiveActivityItem; commit: () => void } | null => {
if (deviceIds.length === 0) return null;
const allRuns = [...runs.values()];
const contentState = buildAgentRunsContentState(allRuns, nowMs);
prunePrActivities(nowMs);
schedulePrActivityExpiry(nowMs);
const allPrActivities = [...prActivities.values()];
const contentState = buildAgentRunsContentState(allRuns, nowMs, allPrActivities);
const activeCount = contentState.activeCount;
const prActivityCount = contentState.prs.length;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
// Stale rows (quiet CLIs) don't count as active but must not END the
// activity either — a 12s-quiet CLI that resumes output would otherwise
// churn end→push-to-start cycles. Only an all-completed/failed (or empty)
// roster ends the aggregate.
const dormantCount = allRuns.filter((run) => run.phase === "stale").length;

let event: "start" | "update" | "end";
if (activeCount > 0 && !liveActivityStarted) event = "start";
else if (activeCount === 0 && dormantCount === 0 && liveActivityStarted) event = "end";
if ((activeCount > 0 || prActivityCount > 0) && !liveActivityStarted) event = "start";
else if (activeCount === 0 && dormantCount === 0 && prActivityCount === 0 && liveActivityStarted) event = "end";
Comment thread
arul28 marked this conversation as resolved.
Resolved
else if (!liveActivityStarted) return null; // nothing live to report (stale-only never starts)
else event = "update";

Expand All @@ -536,10 +596,14 @@ export function createPushPublisherService(deps: PushPublisherDeps) {

const anyWaiting = allRuns.some(
(run) => run.phase === "waiting_for_approval" || run.phase === "waiting_for_input",
);
) || allPrActivities.some((pr) =>
pr.phase === "checks_failing"
|| pr.phase === "changes_requested"
|| pr.phase === "review_requested"
|| pr.phase === "merge_ready");
const phase: "running" | "waiting" | "terminal" = anyWaiting
? "waiting"
: activeCount > 0
: (activeCount > 0 || prActivityCount > 0)
? "running"
: "terminal";

Expand All @@ -556,9 +620,12 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
item.attributesType = AGENT_RUNS_ATTRIBUTES_TYPE;
item.attributes = { machineName: deps.machineName };
const mostRecent = allRuns.sort((left, right) => right.lastActiveAt - left.lastActiveAt)[0];
const mostRecentPr = allPrActivities.sort((left, right) => right.updatedAt - left.updatedAt)[0];
item.alert = {
title: activeCount === 1 && mostRecent ? `${runSubject(mostRecent)} is running` : `${activeCount} agent runs active`,
body: mostRecent ? laneTitleLine(mostRecent) : null,
title: activeCount > 0
? (activeCount === 1 && mostRecent ? `${runSubject(mostRecent)} is running` : `${activeCount} agent runs active`)
: (prActivityCount === 1 && mostRecentPr ? `PR #${mostRecentPr.prNumber} updated` : `${prActivityCount} pull requests updated`),
body: mostRecent ? laneTitleLine(mostRecent) : mostRecentPr?.title ?? null,
Comment thread
arul28 marked this conversation as resolved.
Resolved
};
}
if (event === "end") {
Expand Down Expand Up @@ -944,37 +1011,56 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
scheduleFlush(false);
};

const onPrNotification = (notification: PushPrNotification): void => {
if (notification.kind === "merge_ready") {
enqueueAlert({
sessionId: null,
dedupeKey: `alert:pr:${notification.prNumber}:merge_ready`,
render: () => ({
title: `PR #${notification.prNumber} is ready to merge`,
body: notification.prTitle ?? "Required checks passed and it has approval.",
}),
deepLink: `ade://pr/${notification.prNumber}`,
threadId: `pr-${notification.prNumber}`,
phase: "terminal",
interruptionLevel: "active",
});
scheduleFlush(false);
} else if (notification.kind === "checks_failing") {
enqueueAlert({
sessionId: null,
dedupeKey: `alert:pr:${notification.prNumber}:checks_failing`,
render: () => ({
title: `PR #${notification.prNumber} checks are failing`,
body: notification.prTitle ?? "One or more required checks failed.",
}),
deepLink: `ade://pr/${notification.prNumber}`,
threadId: `pr-${notification.prNumber}`,
phase: "terminal",
interruptionLevel: "active",
});
scheduleFlush(false);
const prNotificationCopy = (notification: PushPrNotification): { title: string; body: string; interruptionLevel: PendingAlert["interruptionLevel"] } => {
const fallbackTitle = notification.prTitle?.trim() || `Pull request #${notification.prNumber}`;
switch (notification.kind) {
case "opened":
return { title: `PR #${notification.prNumber} opened`, body: fallbackTitle, interruptionLevel: "passive" };
case "reopened":
return { title: `PR #${notification.prNumber} reopened`, body: fallbackTitle, interruptionLevel: "passive" };
case "closed":
return { title: `PR #${notification.prNumber} closed`, body: fallbackTitle, interruptionLevel: "passive" };
case "merged":
return { title: `PR #${notification.prNumber} merged`, body: fallbackTitle, interruptionLevel: "active" };
case "merge_ready":
return { title: `PR #${notification.prNumber} is ready to merge`, body: notification.prTitle ?? "Required checks passed and it has approval.", interruptionLevel: "active" };
case "checks_failing":
return { title: `PR #${notification.prNumber} checks are failing`, body: notification.prTitle ?? "One or more required checks failed.", interruptionLevel: "active" };
case "changes_requested":
return { title: `Changes requested on PR #${notification.prNumber}`, body: fallbackTitle, interruptionLevel: "active" };
case "review_requested":
return { title: `Review requested on PR #${notification.prNumber}`, body: fallbackTitle, interruptionLevel: "active" };
}
// review_requested / changes_requested are intentionally not pushed.
};

const onPrNotification = (
notification: PushPrNotification,
resolveLaneName?: (laneId: string) => string | null | undefined,
): void => {
const lane = notification.laneId
? (resolveLaneName?.(notification.laneId) ?? notification.laneId)
: null;
prActivities.set(`pr-${notification.prNumber}`, {
id: `pr-${notification.prNumber}`,
Comment thread
arul28 marked this conversation as resolved.
Resolved
prNumber: notification.prNumber,
title: notification.prTitle?.trim() || `Pull request #${notification.prNumber}`,
phase: notification.kind,
lane,
updatedAt: now(),
});
schedulePrActivityExpiry(now());

const copy = prNotificationCopy(notification);
enqueueAlert({
sessionId: null,
dedupeKey: `alert:pr:${notification.prNumber}:${notification.kind}`,
render: () => ({ title: copy.title, body: copy.body }),
deepLink: `ade://pr/${notification.prNumber}`,
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
threadId: `pr-${notification.prNumber}`,
phase: copy.interruptionLevel === "passive" ? "terminal" : "waiting",
interruptionLevel: copy.interruptionLevel,
});
scheduleFlush(false);
};

const relayApnsConfigured = async (): Promise<boolean | null> => {
Expand Down Expand Up @@ -1033,6 +1119,9 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
if (flushTimer) clearTimeout(flushTimer);
flushTimer = null;
flushFireAt = 0;
if (prExpiryTimer) clearTimeout(prExpiryTimer);
prExpiryTimer = null;
prActivities.clear();
Comment thread
arul28 marked this conversation as resolved.
Resolved
pendingAlerts = [];
}
};
Expand All @@ -1041,7 +1130,9 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
disposed = true;
for (const scopeKey of [...scopes.keys()]) detachScope(scopeKey);
if (flushTimer) clearTimeout(flushTimer);
if (prExpiryTimer) clearTimeout(prExpiryTimer);
flushTimer = null;
prExpiryTimer = null;
};

return {
Expand All @@ -1068,7 +1159,9 @@ export function createPushPublisherService(deps: PushPublisherDeps) {
scopeUnsubscribes.push(sources.ptyService.onExit((event) => onPtyExit(scopeKey, event)));
}
if (sources.subscribePrNotifications) {
scopeUnsubscribes.push(sources.subscribePrNotifications(onPrNotification));
scopeUnsubscribes.push(sources.subscribePrNotifications((notification) => {
onPrNotification(notification, sources.resolveLaneName);
}));
}
scopes.set(scopeKey, {
agentChatService: sources.agentChatService ?? null,
Expand Down
Loading