From da70967a1a2207f4368a6e760ac7ca3106775afa Mon Sep 17 00:00:00 2001 From: Chris Lee Date: Sun, 26 Apr 2026 16:23:54 +1000 Subject: [PATCH 1/2] feat(workflows): up-front tracking comments, trigger reactions, parent cascade MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the silent-multi-minute UX gap during triage/plan/implement runs and the silent-failure window when a daemon dies mid-job. - Triage, plan, and implement post a starting comment with input snapshot before the agent runs, so users see progress instead of an empty thread. - 4-stage GitHub reactions on the trigger comment: eyes (queued), rocket (dispatched), hooray (success), confused (failure / OOM disconnect). - Composite parents (ship) now render as a verbose composite body — each child step shows status, cost, turns, and a deep link to its own comment. Cascade fires automatically on child setState via tracking-mirror. - Orphan/disconnect cleanup in the orchestrator now updates the existing ancestor tracking comment with a failure message and adds a confused reaction, instead of failing silently. - Persist trigger_comment_id + trigger_event_type on workflow_runs and executions so the orphan path (no live BotContext) can reach the right comment after a daemon dies. Co-Authored-By: Claude Opus 4.7 --- docs/BOT-WORKFLOWS.md | 36 ++++ src/daemon/workflow-executor.ts | 23 +++ src/db/migrations/007_trigger_comment.sql | 21 +++ src/orchestrator/connection-handler.ts | 131 +++++++++++++- src/orchestrator/history.ts | 23 ++- src/utils/reactions.ts | 66 ++++++++ src/webhook/events/issue-comment.ts | 16 ++ src/webhook/events/review-comment.ts | 13 ++ src/workflows/dispatcher.ts | 26 +++ src/workflows/execution-row.ts | 25 ++- src/workflows/handlers/implement.ts | 36 ++++ src/workflows/handlers/plan.ts | 34 ++++ src/workflows/handlers/triage.ts | 36 ++++ src/workflows/orchestrator.ts | 46 ++++- src/workflows/runs-store.ts | 52 ++++-- src/workflows/tracking-mirror.ts | 197 ++++++++++++++++++---- test/db/migrate.test.ts | 6 +- test/utils/reactions.test.ts | 126 ++++++++++++++ test/webhook/events/issue-comment.test.ts | 4 + 19 files changed, 855 insertions(+), 62 deletions(-) create mode 100644 src/db/migrations/007_trigger_comment.sql create mode 100644 src/utils/reactions.ts create mode 100644 test/utils/reactions.test.ts diff --git a/docs/BOT-WORKFLOWS.md b/docs/BOT-WORKFLOWS.md index b4cab311..a7410318 100644 --- a/docs/BOT-WORKFLOWS.md +++ b/docs/BOT-WORKFLOWS.md @@ -143,6 +143,42 @@ Composite workflows like `ship` insert a child row per step. When the child comp - **Cost note**: every ship pays for both `review` and `resolve` agent runs. Per project direction (2026-04-25), accuracy beats cost — closing the loop justifies the extra spend. - **Example trigger**: add label `bot:ship`, or comment "`@chrisleekr-bot ship this`" +## User-facing surfaces + +Each workflow run produces two GitHub-visible signals: a **tracking comment** (the bot's working/result body) and a **reaction set** on the user's trigger comment. + +### Tracking comments + +- `triage`, `plan`, and `implement` post an **up-front "starting…" comment** as soon as they fetch the issue title, before the (multi-minute) agent run. The terminal `setState` call rewrites the same comment with the verdict / plan / PR link. Skipping the up-front write would leave the user staring at an empty issue while the daemon worked. +- `review` and `resolve` already post upfront; behaviour unchanged. +- For composite parents (`ship`), the tracking comment is rendered as a **verbose composite**: the parent's narrative followed by one `### — ` block per child step, each linking back to the child's own tracking comment via deep `#issuecomment-` anchors. The composite refresh is triggered automatically by `tracking-mirror.setState` whenever a child run writes — the cascade walks `parent_run_id` and re-renders the parent's body so the user always sees the latest child status on the surface they're already watching. + +### Trigger-comment reactions + +Comment-driven workflows stack four GitHub reactions on the user's trigger comment so the lifecycle is visible without scrolling: + +| Stage | Reaction | Where it fires | +| ------------------------------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------------------ | +| Trigger detected, before classifier | 👀 `eyes` | `src/webhook/events/issue-comment.ts`, `review-comment.ts` (after allowlist) | +| Job dispatched to a daemon | 🚀 `rocket` | `src/workflows/dispatcher.ts` (after `enqueueJob`) | +| Workflow succeeded | 🎉 `hooray` | `src/daemon/workflow-executor.ts` for atomic runs; `src/workflows/orchestrator.ts` for composite parents (cascade) | +| Workflow failed (handler error, daemon disconnect, OOM) | 😕 `confused` | `workflow-executor.ts`, `orchestrator.ts`, `src/orchestrator/connection-handler.ts` (orphan path) | + +GitHub reactions are additive — the combined set is the audit trail. Label-triggered runs (`bot:ship` via label apply) skip reactions silently because no comment exists to react on. Reaction failures (e.g., missing `reactions:write` scope) are logged at warn level and swallowed; they never block a workflow. + +### Failure surface on daemon disconnect + +When a daemon dies abruptly (OOM, pod eviction, network partition), `connection-handler.cleanupAfterDisconnect` walks every in-flight `workflow_runs` row owned by that daemon, finds the topmost ancestor (so a child step failure shows up on the parent's surface), and: + +1. Updates the ancestor's tracking comment with an `❌ Daemon disconnected (likely OOM)` message and resume instructions. +2. Adds 😕 `confused` to the user's trigger comment. + +This closes the silent-failure window that previously left users staring at a stale "starting…" comment after an OOM. The liveness reaper still flips the `workflow_runs.status` to `failed`; the cleanup path only owns the user-visible surface. + +### Re-trigger / resume + +Re-triggering `ship` (re-applying the `bot:ship` label or re-commenting the intent) walks the prior runs via `computeStartIndex` in `src/workflows/handlers/ship.ts`: succeeded `triage`/`plan` rows are reused, succeeded `implement` is reused only while its PR is still open, and `review`/`resolve` always re-run. A failed `implement` row from a prior crash means resume picks up at `implement` — the row is not "succeeded" so `isFresh` returns false and the step is re-queued. + ## Comment intent classifier Comments that mention `@chrisleekr-bot` are routed through `src/workflows/intent-classifier.ts`, which returns `{ workflow, confidence, rationale }` using a single-turn Haiku call. Rules: diff --git a/src/daemon/workflow-executor.ts b/src/daemon/workflow-executor.ts index 4b96d8c0..8df83c7e 100644 --- a/src/daemon/workflow-executor.ts +++ b/src/daemon/workflow-executor.ts @@ -3,6 +3,7 @@ import { Octokit } from "octokit"; import { logger } from "../logger"; import type { SerializableBotContext } from "../shared/daemon-types"; import { createMessageEnvelope, type JobPayloadMessage } from "../shared/ws-messages"; +import { addReaction, type ReactionContent } from "../utils/reactions"; import { type CompletionResult, onStepComplete } from "../workflows/orchestrator"; import { getByName, type WorkflowRunContext } from "../workflows/registry"; import { markFailed, markRunning, markSucceeded, mergeState } from "../workflows/runs-store"; @@ -59,6 +60,22 @@ export async function executeWorkflowRun( const octokit = new Octokit({ auth: installationToken }); + // Best-effort reaction on the user's trigger comment. No-op for child runs + // (commentId === 0 because children inherit nothing from the parent's + // dispatch payload) and for label-triggered runs (no comment exists). + const reactOnTrigger = (content: ReactionContent): void => { + if (context.commentId === 0) return; + void addReaction({ + octokit, + logger: log, + owner: context.owner, + repo: context.repo, + commentId: context.commentId, + eventType: context.eventName, + content, + }); + }; + try { const entry = getByName(workflowRun.workflowName); const daemonId = getDaemonId(); @@ -163,6 +180,8 @@ export async function executeWorkflowRun( "Workflow run completed", ); + reactOnTrigger("hooray"); + completion = { status: "succeeded" }; send({ @@ -202,6 +221,8 @@ export async function executeWorkflowRun( "Workflow run reported failure", ); + reactOnTrigger("confused"); + completion = { status: "failed", reason: result.reason }; send({ @@ -250,6 +271,8 @@ export async function executeWorkflowRun( "Workflow handler threw", ); + reactOnTrigger("confused"); + try { await onStepComplete({ octokit, logger: log }, workflowRun.runId, { status: "failed", diff --git a/src/db/migrations/007_trigger_comment.sql b/src/db/migrations/007_trigger_comment.sql new file mode 100644 index 00000000..ee80e1d0 --- /dev/null +++ b/src/db/migrations/007_trigger_comment.sql @@ -0,0 +1,21 @@ +-- Migration 007: persist the user's trigger comment on workflow_runs and executions. +-- +-- Required so the orphan/disconnect cleanup path (which has no live BotContext) +-- can update the right tracking comment and add a failure reaction (👀 → ❌) +-- when a daemon dies mid-job. Also lets every workflow stage update the same +-- trigger-comment reaction set throughout its lifecycle (eyes → rocket → +-- hooray → confused). +-- +-- Both columns are NULL because label-triggered workflows (e.g. bot:ship via +-- issues.labeled) have no originating comment to react on. No backfill — the +-- reaction lifecycle only matters for jobs dispatched after this migration. + +ALTER TABLE workflow_runs + ADD COLUMN trigger_comment_id BIGINT NULL, + ADD COLUMN trigger_event_type TEXT NULL + CHECK (trigger_event_type IN ('issue_comment', 'pull_request_review_comment')); + +ALTER TABLE executions + ADD COLUMN trigger_comment_id BIGINT NULL, + ADD COLUMN trigger_event_type TEXT NULL + CHECK (trigger_event_type IN ('issue_comment', 'pull_request_review_comment')); diff --git a/src/orchestrator/connection-handler.ts b/src/orchestrator/connection-handler.ts index 777b3bda..29ee9c6f 100644 --- a/src/orchestrator/connection-handler.ts +++ b/src/orchestrator/connection-handler.ts @@ -1,8 +1,11 @@ import type { ServerWebSocket } from "bun"; -import { App } from "octokit"; +import { App, type Octokit } from "octokit"; import { config } from "../config"; import { logger } from "../logger"; +import { addReaction } from "../utils/reactions"; +import { findById, findInflightByOwner, type WorkflowRunRow } from "../workflows/runs-store"; +import { setState } from "../workflows/tracking-mirror"; // Read orchestrator app version at module load so we can detect daemon drift // in handleRegister and request an update via daemon:update-required. @@ -152,11 +155,137 @@ async function cleanupAfterDisconnect(daemonId: string): Promise { "Cleaned up orphaned executions after daemon disconnect", ); } + + // User-facing notification: any in-flight workflow_runs owned by this + // daemon will be flipped to 'failed' by the liveness reaper. We update + // the user's tracking comment + react on the trigger comment now so the + // user sees the failure immediately instead of staring at a stale + // "starting…" comment. + await notifyOrphanedWorkflowRuns(daemonId); } catch (err) { logger.error({ err, daemonId }, "Failed to cleanup after daemon disconnect"); } } +/** + * Update the user-facing tracking comment + add a `confused` reaction on the + * originating comment for every in-flight workflow_run owned by the dying + * daemon. Walks the parent chain so a child step's failure shows up on the + * top-level run's surface (the surface the user is actually watching) rather + * than on a per-child comment they may not have noticed. + * + * Best-effort throughout — a missing GitHub App config or a comment-update + * failure must never bubble up and prevent the rest of cleanup from running. + */ +async function notifyOrphanedWorkflowRuns(daemonId: string): Promise { + let inflight: WorkflowRunRow[]; + try { + inflight = await findInflightByOwner("daemon", daemonId); + } catch (err) { + logger.error({ err, daemonId }, "Failed to query in-flight workflow_runs for orphan cleanup"); + return; + } + + if (inflight.length === 0) return; + + // Dedupe by ancestor so a single ship cascade (ship → plan → implement) + // only updates one comment + one reaction even if multiple of its rows + // were owned by this daemon at the moment of disconnect. + const ancestorIds = new Set(); + for (const row of inflight) { + // eslint-disable-next-line no-await-in-loop + const ancestor = await findTopAncestor(row); + if (ancestor === null) continue; + if (ancestorIds.has(ancestor.id)) continue; + ancestorIds.add(ancestor.id); + + try { + // eslint-disable-next-line no-await-in-loop + await postOrphanNotification(ancestor); + } catch (err) { + logger.warn( + { + err: err instanceof Error ? err.message : String(err), + ancestorRunId: ancestor.id, + daemonId, + }, + "Orphan notification (comment/reaction) failed", + ); + } + } +} + +/** + * Walk parent_run_id up to the topmost row. Returns the input row if it has + * no parent, or null if the parent chain is broken (orphaned mid-walk). + */ +async function findTopAncestor(row: WorkflowRunRow): Promise { + let current: WorkflowRunRow | null = row; + // Bound at 8 levels of nesting — defensive cap for a chain that should + // realistically never exceed depth 2 (ship → step). A null parent ends + // the walk naturally. + for (let i = 0; i < 8; i++) { + if (current === null) return null; + if (current.parent_run_id === null) return current; + // eslint-disable-next-line no-await-in-loop + current = await findById(current.parent_run_id); + } + return current; +} + +async function postOrphanNotification(ancestor: WorkflowRunRow): Promise { + if (config.appId === undefined || config.privateKey === undefined) { + logger.debug( + { ancestorRunId: ancestor.id }, + "Skipping orphan notification — GitHub App credentials not configured", + ); + return; + } + + const app = getOrCreateApp(); + const { data: installation } = await app.octokit.rest.apps.getRepoInstallation({ + owner: ancestor.target_owner, + repo: ancestor.target_repo, + }); + const octokit = await app.getInstallationOctokit(installation.id); + + const humanMessage = [ + `❌ **Daemon disconnected during execution** — likely an OOM kill on the workflow pod.`, + ``, + `The in-flight step has been marked failed. Its workflow_run row will be flipped`, + `to \`failed\` by the liveness reaper. To resume, re-trigger the workflow:`, + ``, + `- For \`ship\`: re-apply the \`bot:ship\` label, or comment again. Resume picks up`, + ` from the failed step and reuses prior succeeded steps.`, + `- For standalone workflows: re-comment with the same trigger.`, + ].join("\n"); + + // Re-uses tracking-mirror.setState so the cascade refresh and `_lastHumanMessage` + // bookkeeping stay consistent — and so the parent's composite body picks up the + // failure narrative on the next render. + const installationOctokit = octokit as unknown as Octokit; + await setState( + { octokit: installationOctokit, logger }, + { + runId: ancestor.id, + patch: { phase: "orphaned" }, + humanMessage, + }, + ); + + if (ancestor.trigger_comment_id !== null && ancestor.trigger_event_type !== null) { + await addReaction({ + octokit: installationOctokit, + logger, + owner: ancestor.target_owner, + repo: ancestor.target_repo, + commentId: ancestor.trigger_comment_id, + eventType: ancestor.trigger_event_type, + content: "confused", + }); + } +} + /** Route validated daemon messages to type-specific handlers. */ export function handleDaemonMessage( ws: ServerWebSocket, diff --git a/src/orchestrator/history.ts b/src/orchestrator/history.ts index 1246f4a4..7d8a9c07 100644 --- a/src/orchestrator/history.ts +++ b/src/orchestrator/history.ts @@ -37,6 +37,13 @@ export interface CreateExecutionParams { triageConfidence?: number; triageCostUsd?: number; contextJson?: SerializableBotContext; + /** + * REST id of the user comment that triggered this run, persisted so the + * orphan/disconnect path (which has no live BotContext) can still react + * on the right comment after a daemon dies. NULL for label/system runs. + */ + triggerCommentId?: number | null; + triggerEventType?: "issue_comment" | "pull_request_review_comment" | null; } /** @@ -63,6 +70,9 @@ export async function createExecution(params: CreateExecutionParams): Promise { + const { octokit, logger, owner, repo, commentId, eventType, content } = params; + + try { + if (eventType === "issue_comment") { + await octokit.rest.reactions.createForIssueComment({ + owner, + repo, + comment_id: commentId, + content, + }); + } else { + await octokit.rest.reactions.createForPullRequestReviewComment({ + owner, + repo, + comment_id: commentId, + content, + }); + } + } catch (err) { + logger.warn( + { + err: err instanceof Error ? err.message : String(err), + owner, + repo, + commentId, + eventType, + content, + }, + "Failed to add reaction — continuing without it", + ); + } +} diff --git a/src/webhook/events/issue-comment.ts b/src/webhook/events/issue-comment.ts index 7980510b..9a40659a 100644 --- a/src/webhook/events/issue-comment.ts +++ b/src/webhook/events/issue-comment.ts @@ -3,6 +3,7 @@ import type { Octokit } from "octokit"; import { containsTrigger } from "../../core/trigger"; import { logger } from "../../logger"; +import { addReaction } from "../../utils/reactions"; import { dispatchByIntent } from "../../workflows/dispatcher"; import { isOwnerAllowed } from "../authorize"; @@ -41,6 +42,19 @@ export function handleIssueComment( return; } + // Acknowledge receipt before the (slow) intent classifier kicks off so the + // user sees an immediate reaction. Subsequent dispatch/handler stages stack + // rocket / hooray / confused on top. + void addReaction({ + octokit, + logger: log, + owner: payload.repository.owner.login, + repo: payload.repository.name, + commentId: payload.comment.id, + eventType: "issue_comment", + content: "eyes", + }); + const isPR = payload.issue.pull_request !== undefined; void dispatchByIntent({ @@ -55,6 +69,8 @@ export function handleIssueComment( }, senderLogin, deliveryId, + triggerCommentId: payload.comment.id, + triggerEventType: "issue_comment", }).catch((err: unknown) => { log.error({ err }, "dispatchByIntent threw for issue_comment"); }); diff --git a/src/webhook/events/review-comment.ts b/src/webhook/events/review-comment.ts index 93829bbb..8ef7bf36 100644 --- a/src/webhook/events/review-comment.ts +++ b/src/webhook/events/review-comment.ts @@ -3,6 +3,7 @@ import type { Octokit } from "octokit"; import { containsTrigger } from "../../core/trigger"; import { logger } from "../../logger"; +import { addReaction } from "../../utils/reactions"; import { dispatchByIntent } from "../../workflows/dispatcher"; import { isOwnerAllowed } from "../authorize"; @@ -38,6 +39,16 @@ export function handleReviewComment( return; } + void addReaction({ + octokit, + logger: log, + owner: payload.repository.owner.login, + repo: payload.repository.name, + commentId: payload.comment.id, + eventType: "pull_request_review_comment", + content: "eyes", + }); + void dispatchByIntent({ octokit, logger: log, @@ -50,6 +61,8 @@ export function handleReviewComment( }, senderLogin, deliveryId, + triggerCommentId: payload.comment.id, + triggerEventType: "pull_request_review_comment", }).catch((err: unknown) => { log.error({ err }, "dispatchByIntent threw for review_comment"); }); diff --git a/src/workflows/dispatcher.ts b/src/workflows/dispatcher.ts index 3a1244e6..302bef76 100644 --- a/src/workflows/dispatcher.ts +++ b/src/workflows/dispatcher.ts @@ -4,6 +4,7 @@ import type pino from "pino"; import { config } from "../config"; import { getInstanceId } from "../orchestrator/instance-id"; import { enqueueJob } from "../orchestrator/job-queue"; +import { addReaction } from "../utils/reactions"; import { recordWorkflowExecution } from "./execution-row"; import { classify, type ClassifyResult } from "./intent-classifier"; import { enforceSingleBotLabel } from "./label-mutex"; @@ -18,6 +19,14 @@ export interface DispatchTarget { readonly number: number; } +/** + * Trigger event type for the user-facing comment that started this workflow. + * Drives which Octokit reactions endpoint is used when the bot reacts on the + * trigger comment (eyes → rocket → hooray → confused). NULL when the trigger + * carries no comment (label apply, branch event). + */ +export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; + export interface DispatchByLabelParams { readonly octokit: Octokit; readonly logger: pino.Logger; @@ -179,6 +188,8 @@ export interface DispatchByIntentParams { readonly target: DispatchTarget; readonly senderLogin: string; readonly deliveryId: string; + readonly triggerCommentId: number; + readonly triggerEventType: TriggerEventType; } /** @@ -194,6 +205,7 @@ export interface DispatchByIntentParams { */ export async function dispatchByIntent(params: DispatchByIntentParams): Promise { const { octokit, logger, commentBody, target, senderLogin, deliveryId } = params; + const { triggerCommentId, triggerEventType } = params; const verdict = await classify(commentBody); logger.info( @@ -265,6 +277,8 @@ export async function dispatchByIntent(params: DispatchByIntentParams): Promise< deliveryId, ownerKind: "orchestrator", ownerId: getInstanceId(), + triggerCommentId, + triggerEventType, }); } catch (err) { if (isInflightCollision(err)) { @@ -294,6 +308,8 @@ export async function dispatchByIntent(params: DispatchByIntentParams): Promise< runId: runRow.id, labels: [entry.label], logger, + triggerCommentId, + triggerEventType, }); await enqueueJob({ deliveryId, @@ -338,6 +354,16 @@ export async function dispatchByIntent(params: DispatchByIntentParams): Promise< "Workflow run dispatched via intent", ); + void addReaction({ + octokit, + logger, + owner: target.owner, + repo: target.repo, + commentId: triggerCommentId, + eventType: triggerEventType, + content: "rocket", + }); + return { status: "dispatched", runId: runRow.id, workflowName: entry.name }; } diff --git a/src/workflows/execution-row.ts b/src/workflows/execution-row.ts index 857a1604..47c24bbf 100644 --- a/src/workflows/execution-row.ts +++ b/src/workflows/execution-row.ts @@ -2,7 +2,7 @@ import type pino from "pino"; import { createExecution } from "../orchestrator/history"; import type { SerializableBotContext } from "../shared/daemon-types"; -import type { DispatchTarget } from "./dispatcher"; +import type { DispatchTarget, TriggerEventType } from "./dispatcher"; /** * Builds the `context_json` shape the accept handler on @@ -14,26 +14,30 @@ import type { DispatchTarget } from "./dispatcher"; * (`deliveryId`, `owner`, `repo`, `entityNumber`). All other fields are * carried for forward-compat parity with the legacy pipeline shape. * - * `eventName` is fixed to `"issue_comment"` to match `BotContext`'s narrow - * union and the existing synthetic-context pattern in `handlers/plan.ts`. + * `eventName` defaults to `"issue_comment"` for label-triggered runs (no + * originating comment) and is set to the real event when the dispatcher + * was called from a comment webhook — that drives which Octokit reactions + * endpoint is used downstream. */ export function buildWorkflowContextJson(params: { target: DispatchTarget; senderLogin: string; deliveryId: string; labels?: readonly string[]; + triggerCommentId?: number; + triggerEventType?: TriggerEventType; }): SerializableBotContext { - const { target, senderLogin, deliveryId, labels } = params; + const { target, senderLogin, deliveryId, labels, triggerCommentId, triggerEventType } = params; return { owner: target.owner, repo: target.repo, entityNumber: target.number, isPR: target.type === "pr", - eventName: "issue_comment", + eventName: triggerEventType ?? "issue_comment", triggerUsername: senderLogin, triggerTimestamp: new Date().toISOString(), triggerBody: "", - commentId: 0, + commentId: triggerCommentId ?? 0, deliveryId, labels: labels !== undefined ? [...labels] : [], defaultBranch: "", @@ -57,14 +61,19 @@ export async function recordWorkflowExecution(params: { runId: string; labels?: readonly string[]; logger: pino.Logger; + triggerCommentId?: number; + triggerEventType?: TriggerEventType; }): Promise { const { deliveryId, target, senderLogin, workflowName, runId, labels, logger } = params; + const { triggerCommentId, triggerEventType } = params; const contextJson = buildWorkflowContextJson({ target, senderLogin, deliveryId, ...(labels !== undefined ? { labels } : {}), + ...(triggerCommentId !== undefined ? { triggerCommentId } : {}), + ...(triggerEventType !== undefined ? { triggerEventType } : {}), }); await createExecution({ @@ -73,11 +82,13 @@ export async function recordWorkflowExecution(params: { repoName: target.repo, entityNumber: target.number, entityType: target.type === "pr" ? "pull_request" : "issue", - eventName: "issue_comment", + eventName: triggerEventType ?? "issue_comment", triggerUsername: senderLogin, dispatchMode: "daemon", dispatchReason: "persistent-daemon", contextJson, + ...(triggerCommentId !== undefined ? { triggerCommentId } : {}), + ...(triggerEventType !== undefined ? { triggerEventType } : {}), }); logger.info( diff --git a/src/workflows/handlers/implement.ts b/src/workflows/handlers/implement.ts index 179fc780..5803c475 100644 --- a/src/workflows/handlers/implement.ts +++ b/src/workflows/handlers/implement.ts @@ -46,6 +46,12 @@ export const handler: WorkflowHandler = async (ctx) => { issue_number: target.number, }); + await postStartingComment(ctx, { + title: issue.title, + number: target.number, + author: issue.user?.login ?? null, + }); + const { data: repoData } = await octokit.rest.repos.get({ owner: target.owner, repo: target.repo, @@ -183,3 +189,33 @@ async function findRecentOpenedPr( } return null; } + +/** + * Up-front tracking comment so the user sees implement has started before the + * (5–15 minute) agent run finishes and opens a PR. Best-effort: if the write + * fails, the handler still proceeds and the terminal setState writes the PR + * link + IMPLEMENT.md report. + */ +async function postStartingComment( + ctx: Parameters[0], + input: { title: string; number: number; author: string | null }, +): Promise { + const author = input.author === null ? "" : ` (opened by @${input.author})`; + const body = [ + `🛠️ **Implement starting** — executing plan for issue #${String(input.number)}${author}`, + ``, + `> ${input.title}`, + ``, + `Cloning the repo, creating a feature branch, and asking the agent to implement`, + `the plan and open a PR. This typically takes 5–15 minutes. The PR link and`, + `IMPLEMENT.md summary replace this comment when the agent finishes.`, + ].join("\n"); + try { + await ctx.setState({ phase: "starting" }, body); + } catch (err) { + ctx.logger.warn( + { err: err instanceof Error ? err.message : String(err) }, + "implement starting-comment write failed — continuing without up-front comment", + ); + } +} diff --git a/src/workflows/handlers/plan.ts b/src/workflows/handlers/plan.ts index 68715d00..b74ad456 100644 --- a/src/workflows/handlers/plan.ts +++ b/src/workflows/handlers/plan.ts @@ -33,6 +33,12 @@ export const handler: WorkflowHandler = async (ctx) => { issue_number: target.number, }); + await postStartingComment(ctx, { + title: issue.title, + number: target.number, + author: issue.user?.login ?? null, + }); + const defaultBranch = await resolveDefaultBranch(octokit, target.owner, target.repo); const { token: installationToken } = (await octokit.auth({ @@ -180,3 +186,31 @@ function buildPlanPrompt(input: { `6. When PLAN.md is written and saved, your job is done.`, ].join("\n"); } + +/** + * Up-front tracking comment so the user sees the bot has started before the + * (multi-minute) agent run produces PLAN.md. Best-effort: if the write fails + * the handler still proceeds and the terminal setState writes the plan. + */ +async function postStartingComment( + ctx: Parameters[0], + input: { title: string; number: number; author: string | null }, +): Promise { + const author = input.author === null ? "" : ` (opened by @${input.author})`; + const body = [ + `📋 **Plan starting** — analyzing issue #${String(input.number)}${author}`, + ``, + `> ${input.title}`, + ``, + `Cloning the repo and asking the agent to produce PLAN.md describing the task`, + `decomposition. The plan replaces this comment when the agent finishes.`, + ].join("\n"); + try { + await ctx.setState({ phase: "starting" }, body); + } catch (err) { + ctx.logger.warn( + { err: err instanceof Error ? err.message : String(err) }, + "plan starting-comment write failed — continuing without up-front comment", + ); + } +} diff --git a/src/workflows/handlers/triage.ts b/src/workflows/handlers/triage.ts index 88c8319a..a5d62d2c 100644 --- a/src/workflows/handlers/triage.ts +++ b/src/workflows/handlers/triage.ts @@ -102,6 +102,12 @@ export const handler: WorkflowHandler = async (ctx) => { issue_number: target.number, }); + await postStartingComment(ctx, { + title: issue.title, + number: target.number, + author: issue.user?.login ?? null, + }); + const defaultBranch = await resolveDefaultBranch(octokit, target.owner, target.repo); const { token: installationToken } = (await octokit.auth({ @@ -341,6 +347,36 @@ function buildTriagePrompt(input: { ].join("\n"); } +/** + * Up-front tracking comment so the user sees the bot has started before the + * (multi-minute) agent run produces its final TRIAGE.md. Best-effort: if the + * comment write fails the handler still proceeds and the terminal setState + * call posts the verdict. + */ +async function postStartingComment( + ctx: Parameters[0], + input: { title: string; number: number; author: string | null }, +): Promise { + const author = input.author === null ? "" : ` (opened by @${input.author})`; + const body = [ + `🔍 **Triage starting** — analyzing issue #${String(input.number)}${author}`, + ``, + `> ${input.title}`, + ``, + `Cloning the repo and running the agent. Bug-class issues will be reproduced by`, + `running code, so this can take a few minutes. The full report and verdict`, + `replace this comment when triage finishes.`, + ].join("\n"); + try { + await ctx.setState({ phase: "starting" }, body); + } catch (err) { + ctx.logger.warn( + { err: err instanceof Error ? err.message : String(err) }, + "triage starting-comment write failed — continuing without up-front comment", + ); + } +} + function composeComment( report: string, verdict: Verdict, diff --git a/src/workflows/orchestrator.ts b/src/workflows/orchestrator.ts index 5e25ffc0..aae10da0 100644 --- a/src/workflows/orchestrator.ts +++ b/src/workflows/orchestrator.ts @@ -5,9 +5,10 @@ import type pino from "pino"; import { requireDb } from "../db"; import { getInstanceId } from "../orchestrator/instance-id"; import { enqueueJob } from "../orchestrator/job-queue"; +import { addReaction, type ReactionContent } from "../utils/reactions"; import { recordWorkflowExecution } from "./execution-row"; import { getByName, type WorkflowName } from "./registry"; -import { markFailed, type WorkflowRunRow } from "./runs-store"; +import { findById, markFailed, type WorkflowRunRow } from "./runs-store"; import { setState } from "./tracking-mirror"; /** @@ -244,13 +245,50 @@ export async function onStepComplete( } if (postCommit.parentTerminal !== null && postCommit.parentRunId !== null) { + const parentRunId = postCommit.parentRunId; + const terminal = postCommit.parentTerminal; await setState(deps, { - runId: postCommit.parentRunId, + runId: parentRunId, patch: {}, - humanMessage: postCommit.parentTerminal.humanMessage, + humanMessage: terminal.humanMessage, }).catch((err: unknown) => { - logger.warn({ err, parentId: postCommit.parentRunId }, "parent tracking emit failed"); + logger.warn({ err, parentId: parentRunId }, "parent tracking emit failed"); }); + + // Composite parents (e.g., ship) terminate here, not in the daemon + // executor — so this is the right point to react on the user's trigger + // comment with the chain's final outcome. + await reactOnParentTrigger( + deps, + parentRunId, + terminal.status === "succeeded" ? "hooray" : "confused", + ); + } +} + +async function reactOnParentTrigger( + deps: OnStepCompleteDeps, + parentRunId: string, + content: ReactionContent, +): Promise { + try { + const row = await findById(parentRunId); + if (row === null) return; + if (row.trigger_comment_id === null || row.trigger_event_type === null) return; + await addReaction({ + octokit: deps.octokit, + logger: deps.logger, + owner: row.target_owner, + repo: row.target_repo, + commentId: row.trigger_comment_id, + eventType: row.trigger_event_type, + content, + }); + } catch (err) { + deps.logger.warn( + { err: err instanceof Error ? err.message : String(err), parentRunId, content }, + "reactOnParentTrigger failed", + ); } } diff --git a/src/workflows/runs-store.ts b/src/workflows/runs-store.ts index 56a31570..3cffb1a8 100644 --- a/src/workflows/runs-store.ts +++ b/src/workflows/runs-store.ts @@ -13,6 +13,8 @@ export type WorkflowRunStatus = "queued" | "running" | "succeeded" | "failed"; export type WorkflowOwnerKind = "orchestrator" | "daemon"; +export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; + export interface WorkflowRunRow { id: string; workflow_name: WorkflowName; @@ -28,6 +30,8 @@ export interface WorkflowRunRow { delivery_id: string | null; owner_kind: WorkflowOwnerKind | null; owner_id: string | null; + trigger_comment_id: number | null; + trigger_event_type: TriggerEventType | null; created_at: Date; updated_at: Date; } @@ -39,14 +43,15 @@ export interface WorkflowRunRow { * up through the rest of the codebase. */ function normalizeRow(row: WorkflowRunRow): WorkflowRunRow { - const raw = row.tracking_comment_id as unknown; - const tracking_comment_id = - raw === null || raw === undefined - ? null - : typeof raw === "string" - ? Number(raw) - : (raw as number); - return { ...row, tracking_comment_id }; + const tracking_comment_id = coerceBigintId(row.tracking_comment_id as unknown); + const trigger_comment_id = coerceBigintId(row.trigger_comment_id as unknown); + return { ...row, tracking_comment_id, trigger_comment_id }; +} + +function coerceBigintId(raw: unknown): number | null { + if (raw === null || raw === undefined) return null; + if (typeof raw === "string") return Number(raw); + return raw as number; } export interface InsertQueuedParams { @@ -68,6 +73,12 @@ export interface InsertQueuedParams { */ ownerKind: WorkflowOwnerKind; ownerId: string; + /** + * REST id of the user comment that triggered this run. NULL for + * label-triggered or system-spawned runs (no comment to react on). + */ + triggerCommentId?: number | null; + triggerEventType?: TriggerEventType | null; } /** @@ -82,17 +93,19 @@ export async function insertQueued( const parentStepIndex = params.parentStepIndex ?? null; const deliveryId = params.deliveryId ?? null; const state = params.initialState ?? {}; + const triggerCommentId = params.triggerCommentId ?? null; + const triggerEventType = params.triggerEventType ?? null; const rows: WorkflowRunRow[] = await sql` INSERT INTO workflow_runs ( workflow_name, target_type, target_owner, target_repo, target_number, parent_run_id, parent_step_index, status, state, delivery_id, - owner_kind, owner_id + owner_kind, owner_id, trigger_comment_id, trigger_event_type ) VALUES ( ${params.workflowName}, ${params.target.type}, ${params.target.owner}, ${params.target.repo}, ${params.target.number}, ${parentRunId}, ${parentStepIndex}, 'queued', ${state}::jsonb, ${deliveryId}, - ${params.ownerKind}, ${params.ownerId} + ${params.ownerKind}, ${params.ownerId}, ${triggerCommentId}, ${triggerEventType} ) RETURNING * `; @@ -312,6 +325,25 @@ export async function findLatestSucceededForTarget( return row === undefined ? null : normalizeRow(row); } +/** + * In-flight rows owned by a specific (kind, id) — used by the disconnect + * cleanup path to find workflow_runs that need a user-facing failure + * notification when their owning daemon dies abruptly. + */ +export async function findInflightByOwner( + ownerKind: WorkflowOwnerKind, + ownerId: string, + sql: SQL = requireDb(), +): Promise { + const rows = (await sql` + SELECT * FROM workflow_runs + WHERE owner_kind = ${ownerKind} + AND owner_id = ${ownerId} + AND status IN ('queued', 'running') + `) as unknown as WorkflowRunRow[]; + return rows.map(normalizeRow); +} + /** * Children of a composite parent, ordered by step index. Used by the * orchestrator to compute the next step. diff --git a/src/workflows/tracking-mirror.ts b/src/workflows/tracking-mirror.ts index 23ce4107..362278be 100644 --- a/src/workflows/tracking-mirror.ts +++ b/src/workflows/tracking-mirror.ts @@ -4,11 +4,21 @@ import type pino from "pino"; import type { WorkflowName } from "./registry"; import { findById, + listChildrenByParent, mergeState, tryReserveTrackingCommentId, type WorkflowRunRow, } from "./runs-store"; +/** + * State key used to persist the last human-readable message written by a + * setState call. The cascade refresh re-uses this when re-rendering the + * parent's composite tracking comment so the parent's own narrative survives + * across child step updates. Underscore prefix marks it as an internal field + * not meant for handler-visible state. + */ +const LAST_HUMAN_MESSAGE_KEY = "_lastHumanMessage"; + /** * FR-026: the tracking comment is a projection of `workflow_runs.state`, not * an independent record. `setState` writes the partial state and the @@ -54,7 +64,10 @@ export async function setState( const { octokit, logger } = deps; const { runId, patch, humanMessage } = params; - await mergeState(runId, patch); + // Persist the human message alongside the caller's patch so the cascade + // refresh can re-render the parent's composite body without losing this + // run's narrative. + await mergeState(runId, { ...patch, [LAST_HUMAN_MESSAGE_KEY]: humanMessage }); const row = await findById(runId); if (row === null) { @@ -63,6 +76,7 @@ export async function setState( const body = renderCommentBody(row, humanMessage); + let resultRow: WorkflowRunRow; if (row.tracking_comment_id === null) { const created = await octokit.rest.issues.createComment({ owner: row.target_owner, @@ -76,54 +90,173 @@ export async function setState( { runId, commentId: created.data.id, workflowName: row.workflow_name }, "Created tracking comment", ); - return { ...row, tracking_comment_id: created.data.id }; - } - - // Lost the race: another concurrent setState already reserved a comment. - // Delete the duplicate we just created so a single canonical comment - // remains. A delete failure is not fatal — the DB still points at the - // winning comment, and the duplicate is cosmetic. - logger.warn( - { - runId, - losingCommentId: created.data.id, - winningCommentId: reservation.trackingCommentId, - workflowName: row.workflow_name, - }, - "Lost tracking-comment reservation race; deleting duplicate comment", - ); - try { - await octokit.rest.issues.deleteComment({ - owner: row.target_owner, - repo: row.target_repo, - comment_id: created.data.id, - }); - } catch (deleteErr) { + resultRow = { ...row, tracking_comment_id: created.data.id }; + } else { + // Lost the race: another concurrent setState already reserved a comment. + // Delete the duplicate we just created so a single canonical comment + // remains. A delete failure is not fatal — the DB still points at the + // winning comment, and the duplicate is cosmetic. logger.warn( { runId, losingCommentId: created.data.id, - err: deleteErr instanceof Error ? deleteErr.message : String(deleteErr), + winningCommentId: reservation.trackingCommentId, + workflowName: row.workflow_name, }, - "Failed to delete duplicate tracking comment", + "Lost tracking-comment reservation race; deleting duplicate comment", ); + try { + await octokit.rest.issues.deleteComment({ + owner: row.target_owner, + repo: row.target_repo, + comment_id: created.data.id, + }); + } catch (deleteErr) { + logger.warn( + { + runId, + losingCommentId: created.data.id, + err: deleteErr instanceof Error ? deleteErr.message : String(deleteErr), + }, + "Failed to delete duplicate tracking comment", + ); + } + await octokit.rest.issues.updateComment({ + owner: row.target_owner, + repo: row.target_repo, + comment_id: reservation.trackingCommentId, + body, + }); + resultRow = { ...row, tracking_comment_id: reservation.trackingCommentId }; } + } else { await octokit.rest.issues.updateComment({ owner: row.target_owner, repo: row.target_repo, - comment_id: reservation.trackingCommentId, + comment_id: row.tracking_comment_id, body, }); - return { ...row, tracking_comment_id: reservation.trackingCommentId }; + resultRow = row; } + // Cascade: when this run is a child of a composite (e.g., ship), refresh + // the parent's tracking comment so the user sees this child's status + // reflected on the parent's comment in real time. Best-effort — a cascade + // failure must never bubble up because the child's own write already + // succeeded. + if (resultRow.parent_run_id !== null) { + await refreshParentCompositeBody(deps, resultRow.parent_run_id).catch((err: unknown) => { + logger.warn( + { + err: err instanceof Error ? err.message : String(err), + parentRunId: resultRow.parent_run_id, + childRunId: runId, + }, + "Cascade refresh of parent composite body failed", + ); + }); + } + + return resultRow; +} + +/** + * Re-render the parent's tracking comment with this run's narrative plus a + * verbose block per child step. No-op when the parent never created its own + * comment (no `tracking_comment_id`) — handlers that opted out cannot have + * their comment refreshed. + */ +async function refreshParentCompositeBody( + deps: TrackingMirrorDeps, + parentRunId: string, +): Promise { + const { octokit } = deps; + + const parent = await findById(parentRunId); + if (parent === null) return; + if (parent.tracking_comment_id === null) return; + + const children = await listChildrenByParent(parentRunId); + const body = renderCompositeBody(parent, children); + await octokit.rest.issues.updateComment({ - owner: row.target_owner, - repo: row.target_repo, - comment_id: row.tracking_comment_id, + owner: parent.target_owner, + repo: parent.target_repo, + comment_id: parent.tracking_comment_id, body, }); - return row; +} + +/** + * Composite render for a parent (e.g., ship) and its child steps. Verbose by + * design — each child gets its own block with status, narrative, and a deep + * link to the child's own tracking comment so the user can drill in. + */ +export function renderCompositeBody( + parent: WorkflowRunRow, + children: readonly WorkflowRunRow[], +): string { + const parentMessage = readLastHumanMessage(parent); + const parentBody = renderCommentBody(parent, parentMessage ?? ""); + + if (children.length === 0) return parentBody; + + const childBlocks = children.map((child) => renderChildBlock(child)).join("\n\n"); + + return `${parentBody}\n\n---\n\n## Steps\n\n${childBlocks}`; +} + +function renderChildBlock(child: WorkflowRunRow): string { + const emoji = statusEmoji(child.status); + const link = + child.tracking_comment_id !== null + ? ` · [open comment](https://github.com/${child.target_owner}/${child.target_repo}/issues/${String(child.target_number)}#issuecomment-${String(child.tracking_comment_id)})` + : ""; + + const meta = renderChildMeta(child); + const message = readLastHumanMessage(child); + const messageBlock = message === null ? "" : `\n${truncateForComposite(message)}`; + + return `### ${emoji} \`${child.workflow_name}\` — ${child.status}${link}${meta}${messageBlock}`; +} + +function readLastHumanMessage(row: WorkflowRunRow): string | null { + const raw = row.state[LAST_HUMAN_MESSAGE_KEY]; + return typeof raw === "string" && raw.length > 0 ? raw : null; +} + +function renderChildMeta(child: WorkflowRunRow): string { + const cost = child.state["costUsd"]; + const turns = child.state["turns"]; + const parts: string[] = []; + if (typeof cost === "number") parts.push(`cost: $${cost.toFixed(4)}`); + if (typeof turns === "number") parts.push(`turns: ${String(turns)}`); + return parts.length > 0 ? `\n_${parts.join(" · ")}_` : ""; +} + +function statusEmoji(status: WorkflowRunRow["status"]): string { + switch (status) { + case "queued": + return "⏳"; + case "running": + return "🔄"; + case "succeeded": + return "✅"; + case "failed": + return "❌"; + } +} + +/** + * The composite body sits inside one GitHub comment alongside the parent's + * own narrative — keep each child's excerpt short so the comment stays + * readable. Drill-in users follow the per-step link to read the full body. + */ +function truncateForComposite(text: string): string { + const limit = 600; + const trimmed = text.trim(); + if (trimmed.length <= limit) return trimmed; + return `${trimmed.slice(0, limit)}…`; } /** diff --git a/test/db/migrate.test.ts b/test/db/migrate.test.ts index 567ee4f7..073af17d 100644 --- a/test/db/migrate.test.ts +++ b/test/db/migrate.test.ts @@ -62,12 +62,14 @@ describe.skipIf(sql === null)("runMigrations", () => { const versions: { version: string }[] = await requireDb()` SELECT version FROM _migrations ORDER BY version `; - expect(versions.length).toBe(5); + expect(versions.length).toBe(7); expect(versions[0]?.version).toBe("001_initial"); expect(versions[1]?.version).toBe("002_repo_knowledge"); expect(versions[2]?.version).toBe("003_dispatch_decisions"); expect(versions[3]?.version).toBe("004_collapse_dispatch_to_daemon"); expect(versions[4]?.version).toBe("005_workflow_runs"); + expect(versions[5]?.version).toBe("006_workflow_runs_ownership"); + expect(versions[6]?.version).toBe("007_trigger_comment"); }); it("is idempotent — second run is a no-op", async () => { @@ -77,7 +79,7 @@ describe.skipIf(sql === null)("runMigrations", () => { const versions: { version: string }[] = await requireDb()` SELECT version FROM _migrations ORDER BY version `; - expect(versions.length).toBe(5); + expect(versions.length).toBe(7); }); it("creates the executions table with expected columns", async () => { diff --git a/test/utils/reactions.test.ts b/test/utils/reactions.test.ts new file mode 100644 index 00000000..6ae56638 --- /dev/null +++ b/test/utils/reactions.test.ts @@ -0,0 +1,126 @@ +/** + * Unit tests for the reactions helper. Covers the two endpoint dispatches + * (issue comment vs PR review comment) and the swallowed-error contract — + * a failing reactions API call must never bubble up because reactions are + * a cosmetic UX layer on top of the workflow. + */ + +import { describe, expect, it, mock } from "bun:test"; +import type pino from "pino"; + +import { addReaction } from "../../src/utils/reactions"; + +function silentLogger(): pino.Logger { + return { + warn: () => undefined, + info: () => undefined, + error: () => undefined, + debug: () => undefined, + trace: () => undefined, + fatal: () => undefined, + child: () => silentLogger(), + } as unknown as pino.Logger; +} + +describe("addReaction", () => { + it("routes issue_comment through createForIssueComment", async () => { + const create = mock(() => Promise.resolve({})); + const octokit = { + rest: { + reactions: { + createForIssueComment: create, + createForPullRequestReviewComment: mock(() => { + throw new Error("wrong endpoint called"); + }), + }, + }, + } as never; + + await addReaction({ + octokit, + logger: silentLogger(), + owner: "acme", + repo: "repo", + commentId: 42, + eventType: "issue_comment", + content: "eyes", + }); + + expect(create).toHaveBeenCalledTimes(1); + expect(create.mock.calls[0]?.[0]).toEqual({ + owner: "acme", + repo: "repo", + comment_id: 42, + content: "eyes", + }); + }); + + it("routes pull_request_review_comment through createForPullRequestReviewComment", async () => { + const create = mock(() => Promise.resolve({})); + const octokit = { + rest: { + reactions: { + createForIssueComment: mock(() => { + throw new Error("wrong endpoint called"); + }), + createForPullRequestReviewComment: create, + }, + }, + } as never; + + await addReaction({ + octokit, + logger: silentLogger(), + owner: "acme", + repo: "repo", + commentId: 99, + eventType: "pull_request_review_comment", + content: "rocket", + }); + + expect(create).toHaveBeenCalledTimes(1); + expect(create.mock.calls[0]?.[0]).toEqual({ + owner: "acme", + repo: "repo", + comment_id: 99, + content: "rocket", + }); + }); + + it("swallows API errors so a missing reactions:write scope can't break the workflow", async () => { + const octokit = { + rest: { + reactions: { + createForIssueComment: mock(() => + Promise.reject(new Error("403 Resource not accessible by integration")), + ), + createForPullRequestReviewComment: mock(() => Promise.resolve({})), + }, + }, + } as never; + + let warned: { msg: string; bindings: unknown } | null = null; + const log = { + ...silentLogger(), + warn: ((bindings: unknown, msg: string): undefined => { + warned = { bindings, msg }; + return undefined; + }) as never, + } as unknown as pino.Logger; + + // Should NOT throw — the assertion is the absence of a thrown error and + // the presence of the warn-level log entry. + await addReaction({ + octokit, + logger: log, + owner: "acme", + repo: "repo", + commentId: 7, + eventType: "issue_comment", + content: "confused", + }); + + expect(warned).not.toBeNull(); + expect(warned?.msg).toBe("Failed to add reaction — continuing without it"); + }); +}); diff --git a/test/webhook/events/issue-comment.test.ts b/test/webhook/events/issue-comment.test.ts index 48df9551..3541f899 100644 --- a/test/webhook/events/issue-comment.test.ts +++ b/test/webhook/events/issue-comment.test.ts @@ -162,6 +162,8 @@ describe.skipIf(sql === null)("issue-comment → dispatchByIntent integration (T target: { type: "issue", owner: "acme", repo: "repo", number: 402 }, senderLogin: "acme", deliveryId: "delivery-intent-402", + triggerCommentId: 555_402, + triggerEventType: "issue_comment", }); expect(intentOutcome.status).toBe("dispatched"); if (intentOutcome.status !== "dispatched") throw new Error("expected dispatched"); @@ -211,6 +213,8 @@ describe.skipIf(sql === null)("issue-comment → dispatchByIntent integration (T target: { type: "issue", owner: "acme", repo: "repo", number: 403 }, senderLogin: "acme", deliveryId: "delivery-intent-403", + triggerCommentId: 555_403, + triggerEventType: "issue_comment", }); expect(outcome.status).toBe("ignored"); expect(mockEnqueueJob).not.toHaveBeenCalled(); From 882b2a7d36fe0bb738c38563949056ffeb131581 Mon Sep 17 00:00:00 2001 From: Chris Lee Date: Sun, 26 Apr 2026 16:39:34 +1000 Subject: [PATCH 2/2] fix(workflows): address CodeRabbit review on PR #61 - src/shared/dispatch-types.ts: TriggerEventType becomes the single source of truth; dispatcher / runs-store / execution-row / utils/reactions all import from here so the union can't silently drift. - src/orchestrator/connection-handler.ts: findTopAncestor now returns null (with a warn log) when the parent chain hits the 8-level safety cap, instead of returning a non-topmost row and updating the wrong tracking comment. - src/workflows/tracking-mirror.ts: lost-race branch now re-fetches the row before re-rendering so we don't clobber the winner's freshly written body with our stale snapshot. - src/utils/reactions.ts: documented the floating-promise contract that makes parameter destructuring outside the try/catch safe under strict TS. Co-Authored-By: Claude Opus 4.7 --- src/orchestrator/connection-handler.ts | 11 +++++++++-- src/shared/dispatch-types.ts | 13 +++++++++++++ src/utils/reactions.ts | 11 +++++++++-- src/workflows/dispatcher.ts | 9 +-------- src/workflows/execution-row.ts | 3 ++- src/workflows/runs-store.ts | 3 +-- src/workflows/tracking-mirror.ts | 13 +++++++++---- 7 files changed, 44 insertions(+), 19 deletions(-) diff --git a/src/orchestrator/connection-handler.ts b/src/orchestrator/connection-handler.ts index 29ee9c6f..637428a2 100644 --- a/src/orchestrator/connection-handler.ts +++ b/src/orchestrator/connection-handler.ts @@ -217,7 +217,10 @@ async function notifyOrphanedWorkflowRuns(daemonId: string): Promise { /** * Walk parent_run_id up to the topmost row. Returns the input row if it has - * no parent, or null if the parent chain is broken (orphaned mid-walk). + * no parent, or null if the parent chain is broken (orphaned mid-walk) or + * exceeds the safety cap. Returning the cap-iteration row would be a silent + * bug: it still has a non-null parent_run_id, so we'd update the wrong + * (mid-chain) tracking comment. */ async function findTopAncestor(row: WorkflowRunRow): Promise { let current: WorkflowRunRow | null = row; @@ -230,7 +233,11 @@ async function findTopAncestor(row: WorkflowRunRow): Promise { diff --git a/src/shared/dispatch-types.ts b/src/shared/dispatch-types.ts index 304ac8e6..67dbe821 100644 --- a/src/shared/dispatch-types.ts +++ b/src/shared/dispatch-types.ts @@ -13,6 +13,19 @@ export const DISPATCH_TARGETS = ["daemon"] as const; export type DispatchTarget = (typeof DISPATCH_TARGETS)[number]; +/** + * TriggerEventType — the GitHub webhook event class for the user comment that + * started a workflow run. Drives which Octokit reactions endpoint is used + * downstream (`createForIssueComment` vs `createForPullRequestReviewComment`). + * + * Persisted on `workflow_runs.trigger_event_type` and `executions.trigger_event_type` + * (see migration `007_trigger_comment.sql`). NULL on label-triggered runs. + * + * Single source of truth — `dispatcher.ts`, `runs-store.ts`, `execution-row.ts`, + * and `utils/reactions.ts` import from here so the union can't silently drift. + */ +export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; + export const DispatchTargetSchema = z.enum(DISPATCH_TARGETS); /** diff --git a/src/utils/reactions.ts b/src/utils/reactions.ts index 870f38d2..2a2fe217 100644 --- a/src/utils/reactions.ts +++ b/src/utils/reactions.ts @@ -1,6 +1,8 @@ import type { Octokit } from "octokit"; import type pino from "pino"; +import type { TriggerEventType } from "../shared/dispatch-types"; + /** * GitHub comment reaction lifecycle for bot-driven workflows. * @@ -14,8 +16,6 @@ import type pino from "pino"; */ export type ReactionContent = "eyes" | "rocket" | "hooray" | "confused"; -export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; - export interface AddReactionParams { octokit: Octokit; logger: pino.Logger; @@ -30,6 +30,13 @@ export interface AddReactionParams { * Best-effort reaction add. Failures are logged at warn level and swallowed * so a missing reactions:write scope (or a deleted comment) never blocks * the workflow that produced the reaction. + * + * Floating-promise contract for callers (`void addReaction(...)`): + * the parameter destructuring below cannot throw at runtime because every + * call site passes a literal `AddReactionParams` object under strict TS + * (`exactOptionalPropertyTypes`, `noUncheckedIndexedAccess`). The only + * runtime error surface is the awaited Octokit call itself, which is + * inside the try/catch. */ export async function addReaction(params: AddReactionParams): Promise { const { octokit, logger, owner, repo, commentId, eventType, content } = params; diff --git a/src/workflows/dispatcher.ts b/src/workflows/dispatcher.ts index 302bef76..efcdfa7f 100644 --- a/src/workflows/dispatcher.ts +++ b/src/workflows/dispatcher.ts @@ -4,6 +4,7 @@ import type pino from "pino"; import { config } from "../config"; import { getInstanceId } from "../orchestrator/instance-id"; import { enqueueJob } from "../orchestrator/job-queue"; +import type { TriggerEventType } from "../shared/dispatch-types"; import { addReaction } from "../utils/reactions"; import { recordWorkflowExecution } from "./execution-row"; import { classify, type ClassifyResult } from "./intent-classifier"; @@ -19,14 +20,6 @@ export interface DispatchTarget { readonly number: number; } -/** - * Trigger event type for the user-facing comment that started this workflow. - * Drives which Octokit reactions endpoint is used when the bot reacts on the - * trigger comment (eyes → rocket → hooray → confused). NULL when the trigger - * carries no comment (label apply, branch event). - */ -export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; - export interface DispatchByLabelParams { readonly octokit: Octokit; readonly logger: pino.Logger; diff --git a/src/workflows/execution-row.ts b/src/workflows/execution-row.ts index 47c24bbf..1170a7cb 100644 --- a/src/workflows/execution-row.ts +++ b/src/workflows/execution-row.ts @@ -2,7 +2,8 @@ import type pino from "pino"; import { createExecution } from "../orchestrator/history"; import type { SerializableBotContext } from "../shared/daemon-types"; -import type { DispatchTarget, TriggerEventType } from "./dispatcher"; +import type { TriggerEventType } from "../shared/dispatch-types"; +import type { DispatchTarget } from "./dispatcher"; /** * Builds the `context_json` shape the accept handler on diff --git a/src/workflows/runs-store.ts b/src/workflows/runs-store.ts index 3cffb1a8..19e3395b 100644 --- a/src/workflows/runs-store.ts +++ b/src/workflows/runs-store.ts @@ -1,6 +1,7 @@ import type { SQL } from "bun"; import { requireDb } from "../db"; +import type { TriggerEventType } from "../shared/dispatch-types"; import type { WorkflowName } from "./registry"; /** @@ -13,8 +14,6 @@ export type WorkflowRunStatus = "queued" | "running" | "succeeded" | "failed"; export type WorkflowOwnerKind = "orchestrator" | "daemon"; -export type TriggerEventType = "issue_comment" | "pull_request_review_comment"; - export interface WorkflowRunRow { id: string; workflow_name: WorkflowName; diff --git a/src/workflows/tracking-mirror.ts b/src/workflows/tracking-mirror.ts index 362278be..70d30221 100644 --- a/src/workflows/tracking-mirror.ts +++ b/src/workflows/tracking-mirror.ts @@ -121,13 +121,18 @@ export async function setState( "Failed to delete duplicate tracking comment", ); } + // Re-render against the freshest row so we don't clobber the winner's + // newer body with our stale snapshot. Both racers wrote `_lastHumanMessage` + // before the reservation, so the latest row already contains the merged + // human message — we just need its post-merge view. + const latest = (await findById(runId)) ?? row; await octokit.rest.issues.updateComment({ - owner: row.target_owner, - repo: row.target_repo, + owner: latest.target_owner, + repo: latest.target_repo, comment_id: reservation.trackingCommentId, - body, + body: renderCommentBody(latest, humanMessage), }); - resultRow = { ...row, tracking_comment_id: reservation.trackingCommentId }; + resultRow = { ...latest, tracking_comment_id: reservation.trackingCommentId }; } } else { await octokit.rest.issues.updateComment({