|
1 | 1 | import type { ServerWebSocket } from "bun"; |
2 | | -import { App } from "octokit"; |
| 2 | +import { App, type Octokit } from "octokit"; |
3 | 3 |
|
4 | 4 | import { config } from "../config"; |
5 | 5 | import { logger } from "../logger"; |
| 6 | +import { addReaction } from "../utils/reactions"; |
| 7 | +import { findById, findInflightByOwner, type WorkflowRunRow } from "../workflows/runs-store"; |
| 8 | +import { setState } from "../workflows/tracking-mirror"; |
6 | 9 |
|
7 | 10 | // Read orchestrator app version at module load so we can detect daemon drift |
8 | 11 | // in handleRegister and request an update via daemon:update-required. |
@@ -152,11 +155,144 @@ async function cleanupAfterDisconnect(daemonId: string): Promise<void> { |
152 | 155 | "Cleaned up orphaned executions after daemon disconnect", |
153 | 156 | ); |
154 | 157 | } |
| 158 | + |
| 159 | + // User-facing notification: any in-flight workflow_runs owned by this |
| 160 | + // daemon will be flipped to 'failed' by the liveness reaper. We update |
| 161 | + // the user's tracking comment + react on the trigger comment now so the |
| 162 | + // user sees the failure immediately instead of staring at a stale |
| 163 | + // "starting…" comment. |
| 164 | + await notifyOrphanedWorkflowRuns(daemonId); |
155 | 165 | } catch (err) { |
156 | 166 | logger.error({ err, daemonId }, "Failed to cleanup after daemon disconnect"); |
157 | 167 | } |
158 | 168 | } |
159 | 169 |
|
| 170 | +/** |
| 171 | + * Update the user-facing tracking comment + add a `confused` reaction on the |
| 172 | + * originating comment for every in-flight workflow_run owned by the dying |
| 173 | + * daemon. Walks the parent chain so a child step's failure shows up on the |
| 174 | + * top-level run's surface (the surface the user is actually watching) rather |
| 175 | + * than on a per-child comment they may not have noticed. |
| 176 | + * |
| 177 | + * Best-effort throughout — a missing GitHub App config or a comment-update |
| 178 | + * failure must never bubble up and prevent the rest of cleanup from running. |
| 179 | + */ |
| 180 | +async function notifyOrphanedWorkflowRuns(daemonId: string): Promise<void> { |
| 181 | + let inflight: WorkflowRunRow[]; |
| 182 | + try { |
| 183 | + inflight = await findInflightByOwner("daemon", daemonId); |
| 184 | + } catch (err) { |
| 185 | + logger.error({ err, daemonId }, "Failed to query in-flight workflow_runs for orphan cleanup"); |
| 186 | + return; |
| 187 | + } |
| 188 | + |
| 189 | + if (inflight.length === 0) return; |
| 190 | + |
| 191 | + // Dedupe by ancestor so a single ship cascade (ship → plan → implement) |
| 192 | + // only updates one comment + one reaction even if multiple of its rows |
| 193 | + // were owned by this daemon at the moment of disconnect. |
| 194 | + const ancestorIds = new Set<string>(); |
| 195 | + for (const row of inflight) { |
| 196 | + // eslint-disable-next-line no-await-in-loop |
| 197 | + const ancestor = await findTopAncestor(row); |
| 198 | + if (ancestor === null) continue; |
| 199 | + if (ancestorIds.has(ancestor.id)) continue; |
| 200 | + ancestorIds.add(ancestor.id); |
| 201 | + |
| 202 | + try { |
| 203 | + // eslint-disable-next-line no-await-in-loop |
| 204 | + await postOrphanNotification(ancestor); |
| 205 | + } catch (err) { |
| 206 | + logger.warn( |
| 207 | + { |
| 208 | + err: err instanceof Error ? err.message : String(err), |
| 209 | + ancestorRunId: ancestor.id, |
| 210 | + daemonId, |
| 211 | + }, |
| 212 | + "Orphan notification (comment/reaction) failed", |
| 213 | + ); |
| 214 | + } |
| 215 | + } |
| 216 | +} |
| 217 | + |
| 218 | +/** |
| 219 | + * Walk parent_run_id up to the topmost row. Returns the input row if it has |
| 220 | + * no parent, or null if the parent chain is broken (orphaned mid-walk) or |
| 221 | + * exceeds the safety cap. Returning the cap-iteration row would be a silent |
| 222 | + * bug: it still has a non-null parent_run_id, so we'd update the wrong |
| 223 | + * (mid-chain) tracking comment. |
| 224 | + */ |
| 225 | +async function findTopAncestor(row: WorkflowRunRow): Promise<WorkflowRunRow | null> { |
| 226 | + let current: WorkflowRunRow | null = row; |
| 227 | + // Bound at 8 levels of nesting — defensive cap for a chain that should |
| 228 | + // realistically never exceed depth 2 (ship → step). A null parent ends |
| 229 | + // the walk naturally. |
| 230 | + for (let i = 0; i < 8; i++) { |
| 231 | + if (current === null) return null; |
| 232 | + if (current.parent_run_id === null) return current; |
| 233 | + // eslint-disable-next-line no-await-in-loop |
| 234 | + current = await findById(current.parent_run_id); |
| 235 | + } |
| 236 | + logger.warn( |
| 237 | + { startRunId: row.id, lastSeenRunId: current?.id ?? null }, |
| 238 | + "findTopAncestor: parent chain exceeded 8 levels — skipping orphan notification to avoid touching the wrong comment", |
| 239 | + ); |
| 240 | + return null; |
| 241 | +} |
| 242 | + |
| 243 | +async function postOrphanNotification(ancestor: WorkflowRunRow): Promise<void> { |
| 244 | + if (config.appId === undefined || config.privateKey === undefined) { |
| 245 | + logger.debug( |
| 246 | + { ancestorRunId: ancestor.id }, |
| 247 | + "Skipping orphan notification — GitHub App credentials not configured", |
| 248 | + ); |
| 249 | + return; |
| 250 | + } |
| 251 | + |
| 252 | + const app = getOrCreateApp(); |
| 253 | + const { data: installation } = await app.octokit.rest.apps.getRepoInstallation({ |
| 254 | + owner: ancestor.target_owner, |
| 255 | + repo: ancestor.target_repo, |
| 256 | + }); |
| 257 | + const octokit = await app.getInstallationOctokit(installation.id); |
| 258 | + |
| 259 | + const humanMessage = [ |
| 260 | + `❌ **Daemon disconnected during execution** — likely an OOM kill on the workflow pod.`, |
| 261 | + ``, |
| 262 | + `The in-flight step has been marked failed. Its workflow_run row will be flipped`, |
| 263 | + `to \`failed\` by the liveness reaper. To resume, re-trigger the workflow:`, |
| 264 | + ``, |
| 265 | + `- For \`ship\`: re-apply the \`bot:ship\` label, or comment again. Resume picks up`, |
| 266 | + ` from the failed step and reuses prior succeeded steps.`, |
| 267 | + `- For standalone workflows: re-comment with the same trigger.`, |
| 268 | + ].join("\n"); |
| 269 | + |
| 270 | + // Re-uses tracking-mirror.setState so the cascade refresh and `_lastHumanMessage` |
| 271 | + // bookkeeping stay consistent — and so the parent's composite body picks up the |
| 272 | + // failure narrative on the next render. |
| 273 | + const installationOctokit = octokit as unknown as Octokit; |
| 274 | + await setState( |
| 275 | + { octokit: installationOctokit, logger }, |
| 276 | + { |
| 277 | + runId: ancestor.id, |
| 278 | + patch: { phase: "orphaned" }, |
| 279 | + humanMessage, |
| 280 | + }, |
| 281 | + ); |
| 282 | + |
| 283 | + if (ancestor.trigger_comment_id !== null && ancestor.trigger_event_type !== null) { |
| 284 | + await addReaction({ |
| 285 | + octokit: installationOctokit, |
| 286 | + logger, |
| 287 | + owner: ancestor.target_owner, |
| 288 | + repo: ancestor.target_repo, |
| 289 | + commentId: ancestor.trigger_comment_id, |
| 290 | + eventType: ancestor.trigger_event_type, |
| 291 | + content: "confused", |
| 292 | + }); |
| 293 | + } |
| 294 | +} |
| 295 | + |
160 | 296 | /** Route validated daemon messages to type-specific handlers. */ |
161 | 297 | export function handleDaemonMessage( |
162 | 298 | ws: ServerWebSocket<WsConnectionData>, |
|
0 commit comments