From edf346d7b0ce9482f7f77ce798574167cf718c57 Mon Sep 17 00:00:00 2001 From: mathuraditya724 Date: Wed, 26 Aug 2026 19:50:48 +0530 Subject: [PATCH 1/2] feat(github): Close outstanding PR discussions Persist admitted review feedback, prompt Jared with the exact unresolved inbox, and verify substantive replies through signed GitHub webhooks. Retry missed items with a bounded escalation path. --- .../skills/respond-to-comment/SKILL.md | 19 + .../container/flue/src/agents/instructions.ts | 11 + .../skills/respond-to-comment/SKILL.md | 19 + .../migrations/0001_conscious_runaways.sql | 24 + .../server/migrations/meta/0001_snapshot.json | 673 ++++++++++++++++++ apps/server/migrations/meta/_journal.json | 7 + apps/server/src/agents/instructions.ts | 11 + apps/server/src/cloudflare.ts | 12 + apps/server/src/db/schema.ts | 35 +- .../events/__tests__/discussion-retry.test.ts | 24 + .../lib/events/__tests__/retention.test.ts | 2 +- .../server/src/lib/events/discussion-retry.ts | 108 +++ apps/server/src/lib/events/retention.ts | 4 +- .../github/__tests__/discussion-store.test.ts | 35 + .../lib/github/__tests__/discussions.test.ts | 226 ++++++ .../src/lib/github/__tests__/prompt.test.ts | 13 + .../server/src/lib/github/discussion-store.ts | 175 +++++ apps/server/src/lib/github/discussions.ts | 232 ++++++ apps/server/src/lib/github/dispatch.ts | 19 + apps/server/src/lib/github/prompt.ts | 4 +- apps/server/src/routes/containers/index.ts | 15 +- apps/server/src/routes/webhooks/github.ts | 99 +++ apps/server/wrangler.jsonc | 7 +- 23 files changed, 1765 insertions(+), 9 deletions(-) create mode 100644 apps/server/migrations/0001_conscious_runaways.sql create mode 100644 apps/server/migrations/meta/0001_snapshot.json create mode 100644 apps/server/src/lib/events/__tests__/discussion-retry.test.ts create mode 100644 apps/server/src/lib/events/discussion-retry.ts create mode 100644 apps/server/src/lib/github/__tests__/discussion-store.test.ts create mode 100644 apps/server/src/lib/github/__tests__/discussions.test.ts create mode 100644 apps/server/src/lib/github/discussion-store.ts create mode 100644 apps/server/src/lib/github/discussions.ts diff --git a/apps/server/container/.agents/skills/respond-to-comment/SKILL.md b/apps/server/container/.agents/skills/respond-to-comment/SKILL.md index f8f2af9..154fa88 100644 --- a/apps/server/container/.agents/skills/respond-to-comment/SKILL.md +++ b/apps/server/container/.agents/skills/respond-to-comment/SKILL.md @@ -74,6 +74,25 @@ gh api graphql --paginate -f query='query($o:String!,$r:String!,$n:Int!,$endCurs 4. If not actionable: reply on the thread with the reason and leave it open for a human to resolve. +## Durable discussion inbox + +The webhook prompt can include a **PR discussion inbox**. Those are durable +messages that still need a real response, even if they arrived before the +current event or were compacted out of the conversation history. + +- Inspect the live PR and answer **every** inbox item in its correct GitHub + channel. Do not send generic acknowledgements, status-only comments, or a + canned reply. The response must address that author's actual question. +- Choose the outcome from the work you did: `addressed` after a fix, `explained` + after a considered answer, or `needs-human` after asking the one concrete + decision you cannot safely make. +- After the substantive response, append the exact hidden marker shown for that + item: ``. GitHub delivers your reply + back to Outpost, which verifies this marker and removes only that item from + the inbox. Never add a marker before the visible response exists. +- An inline thread is resolved only after a real code fix. An explanation or + `needs-human` response deliberately leaves it open for the reviewer. + ## Ownership, not advisement For an actionable item on your own PR, carry the work through the workflow diff --git a/apps/server/container/flue/src/agents/instructions.ts b/apps/server/container/flue/src/agents/instructions.ts index 99d89f9..edeb9b5 100644 --- a/apps/server/container/flue/src/agents/instructions.ts +++ b/apps/server/container/flue/src/agents/instructions.ts @@ -127,6 +127,17 @@ from the repository and context, required authority is missing, or the only available action has irreversible or external impact outside the normal PR workflow. Routine implementation choices are yours to make. +### PR discussion closure + +When a webhook prompt includes a **PR discussion inbox**, it is a durable list +of messages that still need your judgment. Inspect the current PR and respond +to **every** listed item in its proper GitHub channel; never substitute a +generic acknowledgement or status-only comment. After each substantive reply, +append that item's exact hidden \`\` +marker, choosing \`addressed\`, \`explained\`, or \`needs-human\` based on what +you actually did. Do not add a marker for work you skipped or before a real +response exists. Resolve inline threads only after an actual code fix. + ### Model tiering — spend the premium model on judgment only Your own model is chosen per event: a premium reasoning model (Opus) for diff --git a/apps/server/container/skills/respond-to-comment/SKILL.md b/apps/server/container/skills/respond-to-comment/SKILL.md index f8f2af9..154fa88 100644 --- a/apps/server/container/skills/respond-to-comment/SKILL.md +++ b/apps/server/container/skills/respond-to-comment/SKILL.md @@ -74,6 +74,25 @@ gh api graphql --paginate -f query='query($o:String!,$r:String!,$n:Int!,$endCurs 4. If not actionable: reply on the thread with the reason and leave it open for a human to resolve. +## Durable discussion inbox + +The webhook prompt can include a **PR discussion inbox**. Those are durable +messages that still need a real response, even if they arrived before the +current event or were compacted out of the conversation history. + +- Inspect the live PR and answer **every** inbox item in its correct GitHub + channel. Do not send generic acknowledgements, status-only comments, or a + canned reply. The response must address that author's actual question. +- Choose the outcome from the work you did: `addressed` after a fix, `explained` + after a considered answer, or `needs-human` after asking the one concrete + decision you cannot safely make. +- After the substantive response, append the exact hidden marker shown for that + item: ``. GitHub delivers your reply + back to Outpost, which verifies this marker and removes only that item from + the inbox. Never add a marker before the visible response exists. +- An inline thread is resolved only after a real code fix. An explanation or + `needs-human` response deliberately leaves it open for the reviewer. + ## Ownership, not advisement For an actionable item on your own PR, carry the work through the workflow diff --git a/apps/server/migrations/0001_conscious_runaways.sql b/apps/server/migrations/0001_conscious_runaways.sql new file mode 100644 index 0000000..b01cd74 --- /dev/null +++ b/apps/server/migrations/0001_conscious_runaways.sql @@ -0,0 +1,24 @@ +CREATE TABLE `github_discussion_obligations` ( + `id` text PRIMARY KEY NOT NULL, + `repo` text NOT NULL, + `pr_number` integer NOT NULL, + `entity_key` text NOT NULL, + `source_kind` text NOT NULL, + `source_comment_id` text NOT NULL, + `reply_to_comment_id` text, + `author` text NOT NULL, + `body` text NOT NULL, + `url` text, + `event_id` text NOT NULL, + `installation_id` integer, + `status` text DEFAULT 'open' NOT NULL, + `outcome` text, + `verified_at` integer, + `reminder_count` integer DEFAULT 0 NOT NULL, + `last_reminded_at` integer, + `created_at` integer NOT NULL, + `updated_at` integer NOT NULL +); +--> statement-breakpoint +CREATE UNIQUE INDEX `github_discussion_obligations_repo_source_unique` ON `github_discussion_obligations` (`repo`,`source_kind`,`source_comment_id`);--> statement-breakpoint +CREATE INDEX `idx_github_discussion_obligations_entity_status` ON `github_discussion_obligations` (`entity_key`,`status`); diff --git a/apps/server/migrations/meta/0001_snapshot.json b/apps/server/migrations/meta/0001_snapshot.json new file mode 100644 index 0000000..46f5f5d --- /dev/null +++ b/apps/server/migrations/meta/0001_snapshot.json @@ -0,0 +1,673 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "e5a75da4-0aea-44e1-9875-8a5fb3ca8a0c", + "prevId": "1cf3575c-be88-4c59-b200-30dc1895b875", + "tables": { + "accounts": { + "name": "accounts", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "provider_id": { + "name": "provider_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "id_token": { + "name": "id_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "access_token_expires_at": { + "name": "access_token_expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "refresh_token_expires_at": { + "name": "refresh_token_expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "password": { + "name": "password", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": { + "accounts_user_id_users_id_fk": { + "name": "accounts_user_id_users_id_fk", + "tableFrom": "accounts", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "agent_sessions": { + "name": "agent_sessions", + "columns": { + "entity_key": { + "name": "entity_key", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session_id": { + "name": "session_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "session_data": { + "name": "session_data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "github_discussion_obligations": { + "name": "github_discussion_obligations", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "repo": { + "name": "repo", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "pr_number": { + "name": "pr_number", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "entity_key": { + "name": "entity_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_kind": { + "name": "source_kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_comment_id": { + "name": "source_comment_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "reply_to_comment_id": { + "name": "reply_to_comment_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "author": { + "name": "author", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "url": { + "name": "url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "event_id": { + "name": "event_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "installation_id": { + "name": "installation_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'open'" + }, + "outcome": { + "name": "outcome", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "verified_at": { + "name": "verified_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "reminder_count": { + "name": "reminder_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "0" + }, + "last_reminded_at": { + "name": "last_reminded_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "github_discussion_obligations_repo_source_unique": { + "name": "github_discussion_obligations_repo_source_unique", + "columns": [ + "repo", + "source_kind", + "source_comment_id" + ], + "isUnique": true + }, + "idx_github_discussion_obligations_entity_status": { + "name": "idx_github_discussion_obligations_entity_status", + "columns": [ + "entity_key", + "status" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "sessions": { + "name": "sessions", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ip_address": { + "name": "ip_address", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "user_agent": { + "name": "user_agent", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "sessions_token_unique": { + "name": "sessions_token_unique", + "columns": [ + "token" + ], + "isUnique": true + } + }, + "foreignKeys": { + "sessions_user_id_users_id_fk": { + "name": "sessions_user_id_users_id_fk", + "tableFrom": "sessions", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "users": { + "name": "users", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "email_verified": { + "name": "email_verified", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "image": { + "name": "image", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "metadata": { + "name": "metadata", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "users_email_unique": { + "name": "users_email_unique", + "columns": [ + "email" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "verifications": { + "name": "verifications", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "identifier": { + "name": "identifier", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "webhook_events": { + "name": "webhook_events", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "entity_key": { + "name": "entity_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "event": { + "name": "event", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "delivery_id": { + "name": "delivery_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "sender": { + "name": "sender", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "repo": { + "name": "repo", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "installation_id": { + "name": "installation_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "payload": { + "name": "payload", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'pending'" + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "dispatched_at": { + "name": "dispatched_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "completed_at": { + "name": "completed_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "webhook_events_delivery_id_unique": { + "name": "webhook_events_delivery_id_unique", + "columns": [ + "delivery_id" + ], + "isUnique": true + }, + "idx_webhook_events_entity_status": { + "name": "idx_webhook_events_entity_status", + "columns": [ + "entity_key", + "status" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/apps/server/migrations/meta/_journal.json b/apps/server/migrations/meta/_journal.json index 00960d5..76fdca3 100644 --- a/apps/server/migrations/meta/_journal.json +++ b/apps/server/migrations/meta/_journal.json @@ -8,6 +8,13 @@ "when": 1780763816531, "tag": "0000_bent_apocalypse", "breakpoints": true + }, + { + "idx": 1, + "version": "6", + "when": 1787753061605, + "tag": "0001_conscious_runaways", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/server/src/agents/instructions.ts b/apps/server/src/agents/instructions.ts index 99d89f9..edeb9b5 100644 --- a/apps/server/src/agents/instructions.ts +++ b/apps/server/src/agents/instructions.ts @@ -127,6 +127,17 @@ from the repository and context, required authority is missing, or the only available action has irreversible or external impact outside the normal PR workflow. Routine implementation choices are yours to make. +### PR discussion closure + +When a webhook prompt includes a **PR discussion inbox**, it is a durable list +of messages that still need your judgment. Inspect the current PR and respond +to **every** listed item in its proper GitHub channel; never substitute a +generic acknowledgement or status-only comment. After each substantive reply, +append that item's exact hidden \`\` +marker, choosing \`addressed\`, \`explained\`, or \`needs-human\` based on what +you actually did. Do not add a marker for work you skipped or before a real +response exists. Resolve inline threads only after an actual code fix. + ### Model tiering — spend the premium model on judgment only Your own model is chosen per event: a premium reasoning model (Opus) for diff --git a/apps/server/src/cloudflare.ts b/apps/server/src/cloudflare.ts index b4fd76c..ef46cd8 100644 --- a/apps/server/src/cloudflare.ts +++ b/apps/server/src/cloudflare.ts @@ -9,6 +9,7 @@ * remain per-conversation via Jared's scheduleFollowUp(). */ +import { retryOpenDiscussionObligations } from "./lib/events/discussion-retry.ts" import { reconcileStuckDispatched } from "./lib/events/reconcile.ts" import { deleteExpiredWebhookEvents } from "./lib/events/retention.ts" import type { BaseEnvBindings } from "./types/env/base.ts" @@ -26,6 +27,15 @@ export default { env: BaseEnvBindings["Bindings"], _ctx: ExecutionContext, ): Promise { + let discussionRetries = { retried: 0, needsHuman: 0 } + try { + discussionRetries = await retryOpenDiscussionObligations(env, controller.scheduledTime) + } catch (err) { + console.warn("github_discussion_obligations.retry.failed", { + error: err instanceof Error ? err.message : String(err), + }) + } + const deleted = await deleteExpiredWebhookEvents(env.DB, controller.scheduledTime) // Intermediate `d:%` sub-statuses (>30m) never reached the agent — a genuine @@ -61,6 +71,8 @@ export default { timedOut: (stuck.meta.changes ?? 0) + reconciled.timedOut, reconciledCompleted: reconciled.completed, reconciledEntities: reconciled.entities, + discussionRetries: discussionRetries.retried, + discussionNeedsHuman: discussionRetries.needsHuman, actionableRetentionHours: 24, skippedRetentionHours: 6, }) diff --git a/apps/server/src/db/schema.ts b/apps/server/src/db/schema.ts index 7a9b523..52285fa 100644 --- a/apps/server/src/db/schema.ts +++ b/apps/server/src/db/schema.ts @@ -1,4 +1,4 @@ -import { index, integer, sqliteTable, text } from "drizzle-orm/sqlite-core" +import { index, integer, sqliteTable, text, uniqueIndex } from "drizzle-orm/sqlite-core" export const users = sqliteTable("users", { id: text("id").primaryKey(), @@ -75,6 +75,39 @@ export const webhookEvents = sqliteTable( (table) => [index("idx_webhook_events_entity_status").on(table.entityKey, table.status)], ) +export const githubDiscussionObligations = sqliteTable( + "github_discussion_obligations", + { + id: text("id").primaryKey(), + repo: text("repo").notNull(), + prNumber: integer("pr_number").notNull(), + entityKey: text("entity_key").notNull(), + sourceKind: text("source_kind").notNull(), + sourceCommentId: text("source_comment_id").notNull(), + replyToCommentId: text("reply_to_comment_id"), + author: text("author").notNull(), + body: text("body").notNull(), + url: text("url"), + eventId: text("event_id").notNull(), + installationId: integer("installation_id"), + status: text("status").notNull().default("open"), + outcome: text("outcome"), + verifiedAt: integer("verified_at", { mode: "timestamp" }), + reminderCount: integer("reminder_count").notNull().default(0), + lastRemindedAt: integer("last_reminded_at", { mode: "timestamp" }), + createdAt: integer("created_at", { mode: "timestamp" }).notNull(), + updatedAt: integer("updated_at", { mode: "timestamp" }).notNull(), + }, + (table) => [ + uniqueIndex("github_discussion_obligations_repo_source_unique").on( + table.repo, + table.sourceKind, + table.sourceCommentId, + ), + index("idx_github_discussion_obligations_entity_status").on(table.entityKey, table.status), + ], +) + export const agentSessions = sqliteTable("agent_sessions", { entityKey: text("entity_key").primaryKey(), sessionId: text("session_id"), diff --git a/apps/server/src/lib/events/__tests__/discussion-retry.test.ts b/apps/server/src/lib/events/__tests__/discussion-retry.test.ts new file mode 100644 index 0000000..55e656d --- /dev/null +++ b/apps/server/src/lib/events/__tests__/discussion-retry.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest" +import { DISCUSSION_RETRY_DELAY_MS, shouldRetryDiscussion } from "../discussion-retry" + +describe("shouldRetryDiscussion", () => { + const now = Date.UTC(2026, 7, 26, 12, 0, 0) + + it("requeues an open obligation Jared has not been reminded about recently", () => { + expect(shouldRetryDiscussion({ status: "open", reminderCount: 0, lastRemindedAt: null }, now)).toBe(true) + expect( + shouldRetryDiscussion( + { status: "open", reminderCount: 1, lastRemindedAt: new Date(now - DISCUSSION_RETRY_DELAY_MS) }, + now, + ), + ).toBe(true) + }) + + it("does not loop closed, recently reminded, or repeatedly missed obligations", () => { + expect(shouldRetryDiscussion({ status: "verified", reminderCount: 0, lastRemindedAt: null }, now)).toBe(false) + expect(shouldRetryDiscussion({ status: "open", reminderCount: 1, lastRemindedAt: new Date(now - 1) }, now)).toBe( + false, + ) + expect(shouldRetryDiscussion({ status: "open", reminderCount: 3, lastRemindedAt: null }, now)).toBe(false) + }) +}) diff --git a/apps/server/src/lib/events/__tests__/retention.test.ts b/apps/server/src/lib/events/__tests__/retention.test.ts index 014234e..ac1047c 100644 --- a/apps/server/src/lib/events/__tests__/retention.test.ts +++ b/apps/server/src/lib/events/__tests__/retention.test.ts @@ -29,7 +29,7 @@ describe("webhook event retention", () => { expect(prepare).toHaveBeenNthCalledWith( 1, - "DELETE FROM webhook_events WHERE status != 'skipped' AND created_at < ?", + "DELETE FROM webhook_events WHERE status != 'skipped' AND created_at < ? AND NOT EXISTS (SELECT 1 FROM github_discussion_obligations WHERE event_id = webhook_events.id AND status = 'open')", ) expect(bind).toHaveBeenNthCalledWith(1, webhookEventCutoffSeconds(now, WEBHOOK_EVENT_RETENTION_MS)) expect(prepare).toHaveBeenNthCalledWith(2, "DELETE FROM webhook_events WHERE status = 'skipped' AND created_at < ?") diff --git a/apps/server/src/lib/events/discussion-retry.ts b/apps/server/src/lib/events/discussion-retry.ts new file mode 100644 index 0000000..b4db2b6 --- /dev/null +++ b/apps/server/src/lib/events/discussion-retry.ts @@ -0,0 +1,108 @@ +import { createLogger } from "@jared/utils" +import { drizzle } from "drizzle-orm/d1" +import * as dbSchema from "@/db/schema" +import { dispatchGitHubEvent } from "@/lib/github/dispatch" +import type { BaseEnvBindings } from "@/types/env/base" + +type Env = BaseEnvBindings["Bindings"] + +/** Give Jared time to finish the current turn before presenting the inbox again. */ +export const DISCUSSION_RETRY_DELAY_MS = 15 * 60 * 1000 +/** Three reminders are enough to recover from a missed turn without self-looping forever. */ +export const MAX_DISCUSSION_REMINDERS = 3 + +type RetryCandidate = { + status: string + reminderCount: number + lastRemindedAt: Date | null +} + +export function shouldRetryDiscussion(candidate: RetryCandidate, now = Date.now()): boolean { + if (candidate.status !== "open" || candidate.reminderCount >= MAX_DISCUSSION_REMINDERS) return false + return candidate.lastRemindedAt === null || candidate.lastRemindedAt.getTime() <= now - DISCUSSION_RETRY_DELAY_MS +} + +type DiscussionRetryRow = { + id: string + entity_key: string + reminder_count: number + last_reminded_at: number | null + status: string + event_id: string + event: string + action: string | null + delivery_id: string + sender: string | null + repo: string | null + installation_id: number | null + payload: string +} + +export type DiscussionRetryResult = { retried: number; needsHuman: number } + +/** + * Wake Jared with the original event after a missed discussion inbox. We reuse + * the event so dispatch renders all still-open obligations for that same PR; + * after three missed reminders the row is explicitly marked needs_human. + */ +export async function retryOpenDiscussionObligations( + env: Env, + scheduledTime: number, + opts: { maxRows?: number } = {}, +): Promise { + const maxRows = opts.maxRows ?? 50 + const rows = await env.DB.prepare( + `SELECT o.id, o.entity_key, o.reminder_count, o.last_reminded_at, o.status, o.event_id, + e.event, e.action, e.delivery_id, e.sender, e.repo, e.installation_id, e.payload + FROM github_discussion_obligations o + JOIN webhook_events e ON e.id = o.event_id + WHERE o.status = 'open' + ORDER BY o.created_at ASC + LIMIT ?`, + ) + .bind(maxRows) + .all() + + const db = drizzle(env.DB, { schema: dbSchema }) + const logger = createLogger({ level: env.ENV === "development" ? "debug" : "info", namespace: "discussion.retry" }) + let retried = 0 + let needsHuman = 0 + + for (const row of rows.results ?? []) { + const candidate: RetryCandidate = { + status: row.status, + reminderCount: row.reminder_count, + lastRemindedAt: row.last_reminded_at === null ? null : new Date(row.last_reminded_at * 1000), + } + if (candidate.reminderCount >= MAX_DISCUSSION_REMINDERS) { + await env.DB.prepare( + "UPDATE github_discussion_obligations SET status = 'needs_human', updated_at = ? WHERE id = ? AND status = 'open'", + ) + .bind(Math.floor(scheduledTime / 1000), row.id) + .run() + needsHuman += 1 + continue + } + if (!shouldRetryDiscussion(candidate, scheduledTime)) continue + + await env.DB.prepare( + "UPDATE github_discussion_obligations SET reminder_count = reminder_count + 1, last_reminded_at = ?, updated_at = ? WHERE id = ? AND status = 'open'", + ) + .bind(Math.floor(scheduledTime / 1000), Math.floor(scheduledTime / 1000), row.id) + .run() + retried += 1 + await dispatchGitHubEvent(env, db, logger, { + eventId: row.event_id, + containerKey: row.entity_key, + event: row.event, + action: row.action, + deliveryId: row.delivery_id, + sender: row.sender, + repo: row.repo, + installationId: row.installation_id, + payload: row.payload, + }) + } + + return { retried, needsHuman } +} diff --git a/apps/server/src/lib/events/retention.ts b/apps/server/src/lib/events/retention.ts index e44ef84..07bb6c6 100644 --- a/apps/server/src/lib/events/retention.ts +++ b/apps/server/src/lib/events/retention.ts @@ -20,7 +20,9 @@ export async function deleteExpiredWebhookEvents(db: D1Database, now = Date.now( const skippedCutoff = webhookEventCutoffSeconds(now, SKIPPED_WEBHOOK_EVENT_RETENTION_MS) const actionable = await db - .prepare("DELETE FROM webhook_events WHERE status != 'skipped' AND created_at < ?") + .prepare( + "DELETE FROM webhook_events WHERE status != 'skipped' AND created_at < ? AND NOT EXISTS (SELECT 1 FROM github_discussion_obligations WHERE event_id = webhook_events.id AND status = 'open')", + ) .bind(actionableCutoff) .run() diff --git a/apps/server/src/lib/github/__tests__/discussion-store.test.ts b/apps/server/src/lib/github/__tests__/discussion-store.test.ts new file mode 100644 index 0000000..f46c210 --- /dev/null +++ b/apps/server/src/lib/github/__tests__/discussion-store.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from "vitest" +import { makeDiscussionRecord } from "../discussion-store" + +describe("makeDiscussionRecord", () => { + it("uses the GitHub comment identity as the deduplication key while retaining the exact obligation", () => { + const record = makeDiscussionRecord({ + eventId: "delivery-1", + entityKey: "getsentry/cli#1482", + repo: "getsentry/cli", + installationId: 42, + obligation: { + kind: "inline", + prNumber: 1484, + sourceCommentId: "102", + replyToCommentId: "99", + author: "MathurAditya724", + body: "Jared, what happened to it?", + url: "https://github.com/getsentry/cli/pull/1484#discussion_r102", + createdAt: "2026-08-26T13:30:32Z", + }, + now: new Date("2026-08-26T13:31:00Z"), + }) + + expect(record).toMatchObject({ + repo: "getsentry/cli", + sourceCommentId: "102", + status: "open", + entityKey: "getsentry/cli#1482", + sourceKind: "inline", + replyToCommentId: "99", + eventId: "delivery-1", + }) + expect(record.id).toMatch(/^[a-f0-9-]{36}$/) + }) +}) diff --git a/apps/server/src/lib/github/__tests__/discussions.test.ts b/apps/server/src/lib/github/__tests__/discussions.test.ts new file mode 100644 index 0000000..d103183 --- /dev/null +++ b/apps/server/src/lib/github/__tests__/discussions.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it } from "vitest" +import { + extractDiscussionObligation, + extractDiscussionPrNumber, + formatDiscussionInbox, + parseDiscussionResponseMarker, + responseEvidenceFromWebhook, + responseMatchesDiscussion, +} from "../discussions" + +describe("extractDiscussionObligation", () => { + it("keeps a human top-level PR request as a reply obligation", () => { + const obligation = extractDiscussionObligation( + "issue_comment", + "created", + { + issue: { number: 1484, pull_request: {} }, + comment: { + id: 101, + body: "Could we rename this flag?", + user: { login: "maintainer", type: "User" }, + html_url: "https://github.com/getsentry/cli/pull/1484#issuecomment-101", + created_at: "2026-08-26T09:59:31Z", + }, + }, + "jared-outpost[bot]", + ) + + expect(obligation).toMatchObject({ + kind: "top_level", + sourceCommentId: "101", + prNumber: 1484, + author: "maintainer", + body: "Could we rename this flag?", + }) + }) + + it("keeps an inline reviewer follow-up as its own obligation", () => { + const obligation = extractDiscussionObligation( + "pull_request_review_comment", + "created", + { + pull_request: { number: 1484 }, + comment: { + id: 102, + in_reply_to_id: 99, + body: "Jared, what happened to it?", + user: { login: "MathurAditya724", type: "User" }, + html_url: "https://github.com/getsentry/cli/pull/1484#discussion_r102", + created_at: "2026-08-26T13:30:32Z", + }, + }, + "jared-outpost[bot]", + ) + + expect(obligation).toMatchObject({ + kind: "inline", + sourceCommentId: "102", + replyToCommentId: "99", + body: "Jared, what happened to it?", + }) + }) + + it("does not create an obligation for Jared's own reply or an integration status message", () => { + const ownReply = extractDiscussionObligation( + "issue_comment", + "created", + { + issue: { number: 1484, pull_request: {} }, + comment: { id: 103, body: "Fixed in abc123.", user: { login: "jared-outpost[bot]", type: "Bot" } }, + }, + "jared-outpost[bot]", + ) + const integration = extractDiscussionObligation( + "issue_comment", + "created", + { + issue: { number: 1484, pull_request: {} }, + comment: { id: 104, body: "Deployment succeeded.", user: { login: "vercel", type: "Bot" } }, + }, + "jared-outpost[bot]", + ) + + expect(ownReply).toBeNull() + expect(integration).toBeNull() + }) + + it("does not turn deleted comments or dismissed reviews into work Jared owes", () => { + expect( + extractDiscussionObligation( + "issue_comment", + "deleted", + { + issue: { number: 1484, pull_request: {} }, + comment: { id: 101, body: "removed", user: { login: "maintainer" } }, + }, + "jared-outpost[bot]", + ), + ).toBeNull() + expect( + extractDiscussionObligation( + "pull_request_review", + "dismissed", + { pull_request: { number: 1484 }, review: { id: 101, body: "dismissed", user: { login: "maintainer" } } }, + "jared-outpost[bot]", + ), + ).toBeNull() + }) +}) + +describe("extractDiscussionPrNumber", () => { + it("uses the actual PR from a CI event rather than the shared linked-issue session", () => { + expect( + extractDiscussionPrNumber("check_suite", { + check_suite: { pull_requests: [{ number: 1484 }] }, + }), + ).toBe(1484) + }) +}) + +describe("parseDiscussionResponseMarker", () => { + it("reads a hidden outcome marker without constraining the visible reply", () => { + expect( + parseDiscussionResponseMarker( + "The defaults command should show it too.\n", + ), + ).toEqual({ obligationId: "abc123", outcome: "needs-human" }) + }) + + it("rejects unrelated comments and unsupported outcomes", () => { + expect(parseDiscussionResponseMarker("I looked into it.")).toBeNull() + expect(parseDiscussionResponseMarker("")).toBeNull() + }) +}) + +describe("formatDiscussionInbox", () => { + it("requires a considered response to every open discussion without prescribing its wording", () => { + const inbox = formatDiscussionInbox([ + { + id: "abc123", + kind: "top_level", + author: "BYK", + body: "Could we rename all sixel flags?", + url: "https://github.com/getsentry/cli/pull/1484#issuecomment-101", + }, + { + id: "def456", + kind: "inline", + author: "MathurAditya724", + body: "Jared, what happened to it?", + url: "https://github.com/getsentry/cli/pull/1484#discussion_r102", + }, + ]) + + expect(inbox).toContain("2 open discussion obligations") + expect(inbox).toContain("Could we rename all sixel flags?") + expect(inbox).toContain("Jared, what happened to it?") + expect(inbox).toContain("") + expect(inbox).toContain("do not send a generic acknowledgement") + }) +}) + +describe("responseEvidenceFromWebhook", () => { + it("accepts a marker only from Jared's actual GitHub reply", () => { + const evidence = responseEvidenceFromWebhook( + "pull_request_review_comment", + { + pull_request: { number: 1484 }, + comment: { + in_reply_to_id: 102, + body: "The UI path needs the matching row too.\n", + }, + }, + "jared-outpost[bot]", + "jared-outpost[bot]", + ) + + expect(evidence).toEqual({ + obligationId: "abc123", + outcome: "explained", + prNumber: 1484, + replyToCommentId: "102", + }) + }) + + it("does not let another user close an obligation by copying a marker", () => { + const evidence = responseEvidenceFromWebhook( + "issue_comment", + { comment: { body: "" } }, + "maintainer", + "jared-outpost[bot]", + ) + + expect(evidence).toBeNull() + }) + + it("requires visible content besides the marker", () => { + const evidence = responseEvidenceFromWebhook( + "issue_comment", + { + issue: { number: 1484, pull_request: {} }, + comment: { body: "" }, + }, + "jared-outpost[bot]", + "jared-outpost[bot]", + ) + + expect(evidence).toBeNull() + }) + + it("requires an inline response to be on the obligation's specific comment", () => { + const matches = responseMatchesDiscussion( + { kind: "inline", prNumber: 1484, sourceCommentId: "102" }, + { obligationId: "abc123", outcome: "addressed", prNumber: 1484, replyToCommentId: "99" }, + ) + + expect(matches).toBe(false) + }) + + it("only accepts a response for the same pull request and exact inline parent", () => { + const response = { obligationId: "abc123", outcome: "addressed" as const, prNumber: 1484, replyToCommentId: "102" } + + expect(responseMatchesDiscussion({ kind: "inline", prNumber: 1484, sourceCommentId: "102" }, response)).toBe(true) + expect(responseMatchesDiscussion({ kind: "inline", prNumber: 1485, sourceCommentId: "102" }, response)).toBe(false) + }) +}) diff --git a/apps/server/src/lib/github/__tests__/prompt.test.ts b/apps/server/src/lib/github/__tests__/prompt.test.ts index 87d17d2..9bfee7b 100644 --- a/apps/server/src/lib/github/__tests__/prompt.test.ts +++ b/apps/server/src/lib/github/__tests__/prompt.test.ts @@ -69,6 +69,19 @@ describe("formatEventPrompt — review guidance", () => { expect(out).toContain("Jared is a requested reviewer") }) + it("preserves the durable discussion inbox alongside the triggering event", () => { + const out = formatEventPrompt({ + ...baseOpts, + event: "issue_comment", + payload: JSON.stringify({ issue: { number: 1108, pull_request: {}, user: { login: "alice" } } }), + discussionInbox: "## PR discussion inbox — 1 open discussion obligations\n\nCould we rename this flag?", + }) + + expect(out).toContain("New webhook event: issue_comment.submitted") + expect(out).toContain("PR discussion inbox — 1 open discussion obligations") + expect(out).toContain("Could we rename this flag?") + }) + it("truncates long issue bodies", () => { const body = "x".repeat(5000) const payload = JSON.stringify({ issue: { number: 1, title: "Big", body, user: { login: "a" } } }) diff --git a/apps/server/src/lib/github/discussion-store.ts b/apps/server/src/lib/github/discussion-store.ts new file mode 100644 index 0000000..b69d7c3 --- /dev/null +++ b/apps/server/src/lib/github/discussion-store.ts @@ -0,0 +1,175 @@ +import { and, asc, eq } from "drizzle-orm" +import type { DrizzleD1Database } from "drizzle-orm/d1" +import * as dbSchema from "@/db/schema" +import { + type DiscussionObligation, + type DiscussionResponseEvidence, + type DiscussionSourceReference, + type OpenDiscussionObligation, + responseMatchesDiscussion, +} from "./discussions" + +type Db = DrizzleD1Database + +export type DiscussionRecordInput = { + eventId: string + entityKey: string + repo: string + installationId: number | null + obligation: DiscussionObligation + now?: Date +} + +export function makeDiscussionRecord(input: DiscussionRecordInput) { + const now = input.now ?? new Date() + const { obligation } = input + return { + id: crypto.randomUUID(), + repo: input.repo, + prNumber: obligation.prNumber, + entityKey: input.entityKey, + sourceKind: obligation.kind, + sourceCommentId: obligation.sourceCommentId, + replyToCommentId: obligation.replyToCommentId, + author: obligation.author, + body: obligation.body, + url: obligation.url, + eventId: input.eventId, + installationId: input.installationId, + status: "open", + outcome: null, + verifiedAt: null, + reminderCount: 0, + lastRemindedAt: null, + createdAt: now, + updatedAt: now, + } +} + +/** + * Persist an inbound discussion before admitting it to the agent. A redelivery + * refreshes the current message but never makes a second inbox item. + */ +export async function recordDiscussionObligation(db: Db, input: DiscussionRecordInput): Promise { + const record = makeDiscussionRecord(input) + await db + .insert(dbSchema.githubDiscussionObligations) + .values(record) + .onConflictDoUpdate({ + target: [ + dbSchema.githubDiscussionObligations.repo, + dbSchema.githubDiscussionObligations.sourceKind, + dbSchema.githubDiscussionObligations.sourceCommentId, + ], + set: { + sourceKind: record.sourceKind, + replyToCommentId: record.replyToCommentId, + author: record.author, + body: record.body, + url: record.url, + eventId: record.eventId, + installationId: record.installationId, + status: "open", + outcome: null, + verifiedAt: null, + reminderCount: 0, + lastRemindedAt: null, + updatedAt: record.updatedAt, + }, + }) +} + +export async function listOpenDiscussionObligations( + db: Db, + repo: string, + prNumber: number, +): Promise { + const rows = await db + .select({ + id: dbSchema.githubDiscussionObligations.id, + kind: dbSchema.githubDiscussionObligations.sourceKind, + author: dbSchema.githubDiscussionObligations.author, + body: dbSchema.githubDiscussionObligations.body, + url: dbSchema.githubDiscussionObligations.url, + }) + .from(dbSchema.githubDiscussionObligations) + .where( + and( + eq(dbSchema.githubDiscussionObligations.repo, repo), + eq(dbSchema.githubDiscussionObligations.prNumber, prNumber), + eq(dbSchema.githubDiscussionObligations.status, "open"), + ), + ) + .orderBy(asc(dbSchema.githubDiscussionObligations.createdAt)) + + return rows.map((row) => ({ + ...row, + kind: row.kind as OpenDiscussionObligation["kind"], + })) +} + +/** A deleted comment or dismissed review must not remain in Jared's inbox. */ +export async function cancelDiscussionObligation( + db: Db, + repo: string, + source: DiscussionSourceReference, + now = new Date(), +): Promise { + await db + .update(dbSchema.githubDiscussionObligations) + .set({ status: "cancelled", updatedAt: now }) + .where( + and( + eq(dbSchema.githubDiscussionObligations.repo, repo), + eq(dbSchema.githubDiscussionObligations.sourceKind, source.kind), + eq(dbSchema.githubDiscussionObligations.sourceCommentId, source.sourceCommentId), + eq(dbSchema.githubDiscussionObligations.status, "open"), + ), + ) +} + +/** Mark a row closed only after GitHub has delivered Jared's marked reply. */ +export async function verifyDiscussionResponse( + db: Db, + repo: string, + response: DiscussionResponseEvidence, + now = new Date(), +): Promise { + const obligation = await db.query.githubDiscussionObligations.findFirst({ + where: and( + eq(dbSchema.githubDiscussionObligations.id, response.obligationId), + eq(dbSchema.githubDiscussionObligations.repo, repo), + eq(dbSchema.githubDiscussionObligations.status, "open"), + ), + columns: { + id: true, + sourceKind: true, + prNumber: true, + sourceCommentId: true, + }, + }) + if (!obligation) return + if ( + !responseMatchesDiscussion( + { + kind: obligation.sourceKind as DiscussionObligation["kind"], + prNumber: obligation.prNumber, + sourceCommentId: obligation.sourceCommentId, + }, + response, + ) + ) { + return + } + + await db + .update(dbSchema.githubDiscussionObligations) + .set({ status: "verified", outcome: response.outcome, verifiedAt: now, updatedAt: now }) + .where( + and( + eq(dbSchema.githubDiscussionObligations.id, response.obligationId), + eq(dbSchema.githubDiscussionObligations.repo, repo), + eq(dbSchema.githubDiscussionObligations.status, "open"), + ), + ) +} diff --git a/apps/server/src/lib/github/discussions.ts b/apps/server/src/lib/github/discussions.ts new file mode 100644 index 0000000..1efc495 --- /dev/null +++ b/apps/server/src/lib/github/discussions.ts @@ -0,0 +1,232 @@ +import { lookup, lookupString } from "./entity" + +export const DISCUSSION_OUTCOMES = ["addressed", "explained", "needs-human"] as const + +export type DiscussionOutcome = (typeof DISCUSSION_OUTCOMES)[number] +export type DiscussionKind = "top_level" | "inline" | "review" + +export type DiscussionObligation = { + kind: DiscussionKind + prNumber: number + sourceCommentId: string + replyToCommentId: string | null + author: string + body: string + url: string | null + createdAt: string | null +} + +export type DiscussionResponseMarker = { + obligationId: string + outcome: DiscussionOutcome +} + +export type DiscussionResponseEvidence = DiscussionResponseMarker & { + prNumber: number + /** For an inline reply GitHub gives us the comment it directly answers. */ + replyToCommentId: string | null +} + +export type OpenDiscussionObligation = Pick & { id: string } + +export type DiscussionSourceReference = Pick + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null +} + +function asId(value: unknown): string | null { + return typeof value === "number" || typeof value === "string" ? String(value) : null +} + +function sameLogin(a: string | null, b: string): boolean { + return !!a && !!b && a.toLowerCase() === b.toLowerCase() +} + +function extractComment(value: unknown): { + id: string + replyToCommentId: string | null + author: string + authorType: string | null + body: string + url: string | null + createdAt: string | null +} | null { + const comment = asRecord(value) + if (!comment) return null + + const id = asId(comment.id) + const author = lookupString(comment, "user.login") + const body = lookupString(comment, "body")?.trim() + if (!id || !author || !body) return null + + return { + id, + replyToCommentId: asId(comment.in_reply_to_id), + author, + authorType: lookupString(comment, "user.type"), + body, + url: lookupString(comment, "html_url"), + createdAt: lookupString(comment, "created_at"), + } +} + +function prNumberFor(event: string, payload: Record): number | null { + const entity = event === "issue_comment" ? asRecord(payload.issue) : asRecord(payload.pull_request) + return typeof entity?.number === "number" ? entity.number : null +} + +/** Return the PR currently being acted on, even when its session is issue-keyed. */ +export function extractDiscussionPrNumber(event: string, payload: Record): number | null { + const direct = prNumberFor(event, payload) + if (direct !== null) return direct + + if (event === "check_suite" || event === "workflow_run") { + const run = asRecord(payload[event]) + const pullRequests = lookup(run ?? {}, "pull_requests") + if (Array.isArray(pullRequests) && typeof asRecord(pullRequests[0])?.number === "number") { + return asRecord(pullRequests[0])?.number as number + } + } + + return null +} + +/** + * Convert a GitHub discussion webhook into a durable reply obligation. + * + * Top-level integration chatter is intentionally excluded. Inline bot reviews + * remain in scope: automated reviewers often raise real, actionable feedback. + */ +export function extractDiscussionObligation( + event: string, + action: string | null, + payload: Record, + botLogin: string, +): DiscussionObligation | null { + const isTopLevel = event === "issue_comment" + const isInline = event === "pull_request_review_comment" + const isReview = event === "pull_request_review" + if (!isTopLevel && !isInline && !isReview) return null + const actionable = isReview + ? action === "submitted" || action === "edited" + : action === "created" || action === "edited" + if (!actionable) return null + + if (isTopLevel && lookup(payload, "issue.pull_request") == null) return null + + const prNumber = prNumberFor(event, payload) + const source = extractComment(isReview ? payload.review : payload.comment) + if (!prNumber || !source || sameLogin(source.author, botLogin)) return null + if (isTopLevel && source.authorType?.toLowerCase() === "bot") return null + + return { + kind: isTopLevel ? "top_level" : isInline ? "inline" : "review", + prNumber, + sourceCommentId: source.id, + replyToCommentId: source.replyToCommentId, + author: source.author, + body: source.body, + url: source.url, + createdAt: source.createdAt, + } +} + +/** Identify a discussion row even when GitHub is telling us it was removed. */ +export function extractDiscussionSourceReference( + event: string, + payload: Record, +): DiscussionSourceReference | null { + const kind = + event === "issue_comment" + ? "top_level" + : event === "pull_request_review_comment" + ? "inline" + : event === "pull_request_review" + ? "review" + : null + if (!kind) return null + const source = asRecord(kind === "review" ? payload.review : payload.comment) + const sourceCommentId = asId(source?.id) + return sourceCommentId ? { kind, sourceCommentId } : null +} + +export function discussionResponseMarker(obligationId: string, outcome: DiscussionOutcome): string { + return `` +} + +/** + * A durable inbox rendered into each admitted turn. The outcome is intentionally + * chosen by Jared after inspecting the live PR; the ledger only guarantees that + * every message gets a direct, attributable response. + */ +export function formatDiscussionInbox(obligations: OpenDiscussionObligation[]): string { + if (obligations.length === 0) return "" + + const items = obligations + .map( + (obligation, index) => `### ${index + 1}. ${obligation.kind.replace("_", " ")} from ${obligation.author} +${obligation.url ? `${obligation.url}\n` : ""} +${obligation.body} + +After your substantive reply, append \`\`, where \`\` is \`addressed\`, \`explained\`, or \`needs-human\`.`, + ) + .join("\n\n") + + return ` + +## PR discussion inbox — ${obligations.length} open discussion obligations + +Before you finish this turn, inspect the current PR and respond to every item below in its correct GitHub channel. Think through the request and give the reviewer a direct answer; do not send a generic acknowledgement, a status-only message, or a fixed template. You may fix code, explain a considered decision, or ask the one specific human decision that is genuinely required. Resolve an inline thread only after an actual fix; a thoughtful explanation or question leaves it open. + +${items}` +} + +export function parseDiscussionResponseMarker(body: string): DiscussionResponseMarker | null { + const match = //.exec(body) + if (!match) return null + return { obligationId: match[1]!, outcome: match[2]! as DiscussionOutcome } +} + +/** + * A signed webhook for Jared's own reply is the immediate completion receipt. + * Do not trust a copied marker authored by a reviewer or another integration. + */ +export function responseEvidenceFromWebhook( + event: string, + payload: Record, + sender: string | null, + botLogin: string, +): DiscussionResponseEvidence | null { + if (!sameLogin(sender, botLogin)) return null + if (event !== "issue_comment" && event !== "pull_request_review_comment" && event !== "pull_request_review") + return null + + const source = event === "pull_request_review" ? asRecord(payload.review) : asRecord(payload.comment) + const body = lookupString(source ?? {}, "body") ?? "" + const marker = parseDiscussionResponseMarker(body) + const prNumber = extractDiscussionPrNumber(event, payload) + // A marker is a receipt, not the reply itself. Requiring text outside it + // stops a broken agent from clearing the inbox without talking to anyone. + const visibleBody = body + .replace(//g, "") + .trim() + if (!marker || !prNumber || !visibleBody) return null + + return { + ...marker, + prNumber, + replyToCommentId: asId(source?.in_reply_to_id), + } +} + +/** A receipt must belong to the same PR and, for review threads, reply to it. */ +export function responseMatchesDiscussion( + obligation: Pick, + response: DiscussionResponseEvidence, +): boolean { + if (obligation.prNumber !== response.prNumber) return false + return obligation.kind !== "inline" || response.replyToCommentId === obligation.sourceCommentId +} diff --git a/apps/server/src/lib/github/dispatch.ts b/apps/server/src/lib/github/dispatch.ts index 8e97f5d..98db5ad 100644 --- a/apps/server/src/lib/github/dispatch.ts +++ b/apps/server/src/lib/github/dispatch.ts @@ -16,6 +16,8 @@ import { dispatchToFlueAgent } from "@/lib/containers/flue-dispatch" import { toAgentInstanceId } from "@/lib/containers/ids" import { SANDBOX_OPTS } from "@/lib/containers/sandbox-opts" import { createGitHubApp } from "@/lib/github/app" +import { listOpenDiscussionObligations } from "@/lib/github/discussion-store" +import { extractDiscussionPrNumber, formatDiscussionInbox } from "@/lib/github/discussions" import { classifyModelTier } from "@/lib/github/model-tier" import { formatEventPrompt } from "@/lib/github/prompt" import type { BaseEnvBindings } from "@/types/env/base" @@ -120,6 +122,22 @@ export async function dispatchGitHubEvent(env: Env, db: Db, logger: Logger, evt: await mark("d:setup_done") logger.info({ entity_key: containerKey, event_id: eventId, sandbox_id: sandboxId }, "dispatch.sandbox_ready.done") + let discussionInbox = "" + try { + const payload = JSON.parse(evt.payload) as Record + const prNumber = extractDiscussionPrNumber(evt.event, payload) + if (evt.repo && prNumber !== null) { + discussionInbox = formatDiscussionInbox(await listOpenDiscussionObligations(db, evt.repo, prNumber)) + } + } catch (err) { + // Discussion tracking must never prevent a normal webhook turn. The next + // admitted event will retry the snapshot query. + logger.warn( + { entity_key: containerKey, event_id: eventId, reason: formatError(err) }, + "discussion inbox load failed", + ) + } + const prompt = formatEventPrompt({ event: evt.event, action: evt.action, @@ -130,6 +148,7 @@ export async function dispatchGitHubEvent(env: Env, db: Db, logger: Logger, evt: payload: evt.payload, botLogin, modelTier: classifyModelTier(evt.event, evt.action, evt.payload), + discussionInbox, }) logger.info({ entity_key: containerKey, event_id: eventId }, "dispatch.prompt.start") diff --git a/apps/server/src/lib/github/prompt.ts b/apps/server/src/lib/github/prompt.ts index 7a7f836..09b2e11 100644 --- a/apps/server/src/lib/github/prompt.ts +++ b/apps/server/src/lib/github/prompt.ts @@ -253,6 +253,8 @@ export function formatEventPrompt(opts: { * reads (see `modelForDelivery`) to pick its model. Omit to leave it heavy. */ modelTier?: "light" | "heavy" + /** Durable, currently-open PR discussion items that also need a response. */ + discussionInbox?: string }): string { const eventLabel = opts.action ? `${opts.event}.${opts.action}` : opts.event const data = parsePayload(opts.payload) @@ -275,5 +277,5 @@ ${routingContext(involvement)} ${reviewGuidance(opts.event, data)} ## Event context -${context}` +${context}${opts.discussionInbox ?? ""}` } diff --git a/apps/server/src/routes/containers/index.ts b/apps/server/src/routes/containers/index.ts index 1e4fd75..9f13885 100644 --- a/apps/server/src/routes/containers/index.ts +++ b/apps/server/src/routes/containers/index.ts @@ -438,6 +438,9 @@ const router = new Hono() await Promise.all([ db.delete(dbSchema.agentSessions).where(eq(dbSchema.agentSessions.entityKey, entityKey)), db.delete(dbSchema.webhookEvents).where(eq(dbSchema.webhookEvents.entityKey, entityKey)), + db + .delete(dbSchema.githubDiscussionObligations) + .where(eq(dbSchema.githubDiscussionObligations.entityKey, entityKey)), ]) } catch { /* best effort */ @@ -1049,6 +1052,9 @@ const router = new Hono() // Also drop stored webhook events so a later re-trigger starts with a // clean "Recent events" list instead of resurrecting the old one. db.delete(dbSchema.webhookEvents).where(eq(dbSchema.webhookEvents.entityKey, entityKey)), + db + .delete(dbSchema.githubDiscussionObligations) + .where(eq(dbSchema.githubDiscussionObligations.entityKey, entityKey)), ]), ) return c.json({ ok: true, mode, deleted: idleKeys.length, destroyed: 0 }) @@ -1066,7 +1072,11 @@ const router = new Hono() if (rows.length > 0) { // Clear both the session snapshots and the stored webhook events so a full // wipe leaves no D1 residue to resurface on the next trigger. - await Promise.all([db.delete(dbSchema.agentSessions), db.delete(dbSchema.webhookEvents)]) + await Promise.all([ + db.delete(dbSchema.agentSessions), + db.delete(dbSchema.webhookEvents), + db.delete(dbSchema.githubDiscussionObligations), + ]) } return c.json({ ok: true, mode, deleted: rows.length, destroyed }) @@ -1079,6 +1089,9 @@ const router = new Hono() await Promise.all([ db.delete(dbSchema.agentSessions).where(eq(dbSchema.agentSessions.entityKey, entityKey)), db.delete(dbSchema.webhookEvents).where(eq(dbSchema.webhookEvents.entityKey, entityKey)), + db + .delete(dbSchema.githubDiscussionObligations) + .where(eq(dbSchema.githubDiscussionObligations.entityKey, entityKey)), ]) return c.json({ ok: true, entityKey }) }) diff --git a/apps/server/src/routes/webhooks/github.ts b/apps/server/src/routes/webhooks/github.ts index 3991c22..f81b457 100644 --- a/apps/server/src/routes/webhooks/github.ts +++ b/apps/server/src/routes/webhooks/github.ts @@ -19,6 +19,16 @@ import { } from "@/lib/github/actionability" import { createGitHubApp, type GitHubApp } from "@/lib/github/app" import { TRIGGER_LABEL } from "@/lib/github/constants" +import { + cancelDiscussionObligation, + recordDiscussionObligation, + verifyDiscussionResponse, +} from "@/lib/github/discussion-store" +import { + extractDiscussionObligation, + extractDiscussionSourceReference, + responseEvidenceFromWebhook, +} from "@/lib/github/discussions" import { dispatchGitHubEvent } from "@/lib/github/dispatch" import { extractEntityKey, lookup, lookupString } from "@/lib/github/entity" import { deriveGitHubInvolvement, shouldAdmitGitHubEvent } from "@/lib/github/involvement" @@ -64,6 +74,36 @@ async function isCiStillRunning(opts: { } } +/** + * A reviewer can reply directly to one of Jared's inline comments without a + * label or another @mention. Fetch that immediate parent so the normal noise + * gate never drops a real follow-up merely because Jared is not the PR author. + */ +async function isDirectReplyToJared(opts: { + event: string + action: string | null + payload: Record + installationId: number | null + repo: string | null + botLogin: string + app: GitHubApp +}): Promise { + const { event, action, payload, installationId, repo, botLogin, app } = opts + if (event !== "pull_request_review_comment" || (action !== "created" && action !== "edited")) return false + const parentId = lookup(payload, "comment.in_reply_to_id") + if ((typeof parentId !== "number" && typeof parentId !== "string") || !installationId || !repo || !botLogin) + return false + const [owner, name] = repo.split("/") + if (!owner || !name) return false + const commentId = Number(parentId) + if (!Number.isSafeInteger(commentId)) return false + + const parent = await app + .getInstallationOctokit(installationId) + .pulls.getReviewComment({ owner, repo: name, comment_id: commentId }) + return parent.data.user?.login?.toLowerCase() === botLogin.toLowerCase() +} + const router = new Hono().post("/", async (c) => { const logger = c.get("logger").child({ ns: "webhook.github" }) const db = c.get("db") @@ -184,6 +224,18 @@ const router = new Hono().post("/", async (c) => { ? "no_label" : null + if (skipReason === "no_label") { + try { + if (await isDirectReplyToJared({ event, action, payload, installationId, repo, botLogin, app })) { + skipReason = null + } + } catch (err) { + // Fail closed here: a parent lookup outage should not turn every inline + // review reply into an unrelated agent wake-up. + logger.warn({ error: formatError(err) }, "inline parent lookup failed") + } + } + const containerKey = entityKey?.key ?? `ephemeral/${deliveryId}` // Second gate: drop noisy CI lifecycle events (requested/in_progress, and @@ -256,6 +308,53 @@ const router = new Hono().post("/", async (c) => { throw err } + // A signed webhook for Jared's own reply is the completion receipt. This is + // intentionally processed before self-event skipping, so no agent turn is + // needed merely to close the durable inbox item. + const responseEvidence = responseEvidenceFromWebhook(event, payload, sender, botLogin) + if (responseEvidence && repo) { + try { + await verifyDiscussionResponse(db, repo, responseEvidence) + } catch (err) { + logger.error({ delivery_id: deliveryId, reason: formatError(err) }, "discussion response verification failed") + Sentry.captureException(err) + } + } + + const removedDiscussion = + (event === "issue_comment" || event === "pull_request_review_comment") && action === "deleted" + ? extractDiscussionSourceReference(event, payload) + : event === "pull_request_review" && action === "dismissed" + ? extractDiscussionSourceReference(event, payload) + : null + if (removedDiscussion && repo) { + try { + await cancelDiscussionObligation(db, repo, removedDiscussion) + } catch (err) { + logger.error({ delivery_id: deliveryId, reason: formatError(err) }, "discussion obligation cancellation failed") + Sentry.captureException(err) + } + } + + // Make the PR discussion durable before handing it to the agent. The live + // conversation may compact or receive later CI activity; the inbox is the + // durable source of every reply Jared still owes. + const discussion = botLogin ? extractDiscussionObligation(event, action, payload, botLogin) : null + if (!isSkipped && discussion && entityKey && repo) { + try { + await recordDiscussionObligation(db, { + eventId, + entityKey: containerKey, + repo, + installationId, + obligation: discussion, + }) + } catch (err) { + logger.error({ delivery_id: deliveryId, reason: formatError(err) }, "discussion obligation persistence failed") + Sentry.captureException(err) + } + } + if (isSkipped) { logger.info({ delivery_id: deliveryId, event, action, reason: skipReason }, "event skipped") return c.json({ diff --git a/apps/server/wrangler.jsonc b/apps/server/wrangler.jsonc index 1cc46b5..0a2e872 100644 --- a/apps/server/wrangler.jsonc +++ b/apps/server/wrangler.jsonc @@ -62,10 +62,9 @@ } ], "triggers": { - // Daily at 03:17 UTC: delete webhook_events older than 24 hours - // (worst-case age ~48h with a once-daily schedule). - // Agent follow-ups remain per-conversation via DO scheduleFollowUp(). - "crons": ["17 3 * * *"] + // The discussion inbox needs bounded follow-ups; retention and stalled-event + // housekeeping are cheap enough to run alongside it every fifteen minutes. + "crons": ["*/15 * * * *"] }, "observability": { "enabled": true, From e4b88747f911bc4cfa074db760c956d40ca2a872 Mon Sep 17 00:00:00 2001 From: mathuraditya724 Date: Thu, 27 Aug 2026 00:54:01 +0530 Subject: [PATCH 2/2] fix(github): Harden discussion inbox retries Delay the first retry, preserve review-thread reply semantics, group retry dispatch by PR, and clear obligations with their source events. --- .../skills/respond-to-comment/SKILL.md | 6 ++- .../container/flue/src/agents/instructions.ts | 6 +++ .../skills/respond-to-comment/SKILL.md | 6 ++- .../migrations/0001_conscious_runaways.sql | 2 + .../server/migrations/meta/0001_snapshot.json | 20 +++++++- apps/server/migrations/meta/_journal.json | 2 +- apps/server/src/agents/instructions.ts | 6 +++ apps/server/src/db/schema.ts | 7 +++ .../events/__tests__/discussion-retry.test.ts | 49 +++++++++++++++++-- .../server/src/lib/events/discussion-retry.ts | 49 ++++++++++++++----- .../github/__tests__/discussion-store.test.ts | 1 + .../lib/github/__tests__/discussions.test.ts | 27 +++++++--- .../server/src/lib/github/discussion-store.ts | 9 +++- apps/server/src/lib/github/discussions.ts | 17 +++++-- apps/server/src/routes/containers/index.ts | 3 ++ apps/server/src/routes/events/index.ts | 4 +- 16 files changed, 177 insertions(+), 37 deletions(-) diff --git a/apps/server/container/.agents/skills/respond-to-comment/SKILL.md b/apps/server/container/.agents/skills/respond-to-comment/SKILL.md index 154fa88..7ea82c2 100644 --- a/apps/server/container/.agents/skills/respond-to-comment/SKILL.md +++ b/apps/server/container/.agents/skills/respond-to-comment/SKILL.md @@ -121,8 +121,10 @@ gh api -X POST \ -f body="Fixed in ." ``` -`` is the review comment's `id` from the event payload (use the -top-level comment of the thread — the one with `in_reply_to_id` unset). +`` must be the top-level comment of the review thread — the one +with `in_reply_to_id` unset; GitHub does not support replies to replies. For a +durable inbox item, use the supplied **top-level comment ID**, not the message +ID shown in parentheses. **Top-level PR comment** (`issue_comment` on a PR, not tied to a line) — reply with: diff --git a/apps/server/container/flue/src/agents/instructions.ts b/apps/server/container/flue/src/agents/instructions.ts index edeb9b5..284c7b8 100644 --- a/apps/server/container/flue/src/agents/instructions.ts +++ b/apps/server/container/flue/src/agents/instructions.ts @@ -55,6 +55,12 @@ FIRST — if any matches, stop with \`SKIPPED: \`. Otherwise, route by t decision table. This routing is deterministic: the same event always maps to the same skill. +**Durable inbox exception:** when the prompt includes a **PR discussion inbox**, +process every listed item with \`respond-to-comment\` even if the triggering +event would otherwise be skipped for being uninvolved or directed at another +user. The inbox was already admitted and is the outstanding work; do not return +\`SKIPPED\` before addressing it. + ### Skip conditions (check first, in order) 1. \`payload.sender.login\` equals \`$ME\` (self-triggered) — skip, EXCEPT for diff --git a/apps/server/container/skills/respond-to-comment/SKILL.md b/apps/server/container/skills/respond-to-comment/SKILL.md index 154fa88..7ea82c2 100644 --- a/apps/server/container/skills/respond-to-comment/SKILL.md +++ b/apps/server/container/skills/respond-to-comment/SKILL.md @@ -121,8 +121,10 @@ gh api -X POST \ -f body="Fixed in ." ``` -`` is the review comment's `id` from the event payload (use the -top-level comment of the thread — the one with `in_reply_to_id` unset). +`` must be the top-level comment of the review thread — the one +with `in_reply_to_id` unset; GitHub does not support replies to replies. For a +durable inbox item, use the supplied **top-level comment ID**, not the message +ID shown in parentheses. **Top-level PR comment** (`issue_comment` on a PR, not tied to a line) — reply with: diff --git a/apps/server/migrations/0001_conscious_runaways.sql b/apps/server/migrations/0001_conscious_runaways.sql index b01cd74..17374fd 100644 --- a/apps/server/migrations/0001_conscious_runaways.sql +++ b/apps/server/migrations/0001_conscious_runaways.sql @@ -21,4 +21,6 @@ CREATE TABLE `github_discussion_obligations` ( ); --> statement-breakpoint CREATE UNIQUE INDEX `github_discussion_obligations_repo_source_unique` ON `github_discussion_obligations` (`repo`,`source_kind`,`source_comment_id`);--> statement-breakpoint +CREATE INDEX `idx_github_discussion_obligations_repo_pr_status_created` ON `github_discussion_obligations` (`repo`,`pr_number`,`status`,`created_at`);--> statement-breakpoint +CREATE INDEX `idx_github_discussion_obligations_status_created` ON `github_discussion_obligations` (`status`,`created_at`);--> statement-breakpoint CREATE INDEX `idx_github_discussion_obligations_entity_status` ON `github_discussion_obligations` (`entity_key`,`status`); diff --git a/apps/server/migrations/meta/0001_snapshot.json b/apps/server/migrations/meta/0001_snapshot.json index 46f5f5d..d74ed4a 100644 --- a/apps/server/migrations/meta/0001_snapshot.json +++ b/apps/server/migrations/meta/0001_snapshot.json @@ -279,7 +279,7 @@ "primaryKey": false, "notNull": true, "autoincrement": false, - "default": "0" + "default": 0 }, "last_reminded_at": { "name": "last_reminded_at", @@ -313,6 +313,24 @@ ], "isUnique": true }, + "idx_github_discussion_obligations_repo_pr_status_created": { + "name": "idx_github_discussion_obligations_repo_pr_status_created", + "columns": [ + "repo", + "pr_number", + "status", + "created_at" + ], + "isUnique": false + }, + "idx_github_discussion_obligations_status_created": { + "name": "idx_github_discussion_obligations_status_created", + "columns": [ + "status", + "created_at" + ], + "isUnique": false + }, "idx_github_discussion_obligations_entity_status": { "name": "idx_github_discussion_obligations_entity_status", "columns": [ diff --git a/apps/server/migrations/meta/_journal.json b/apps/server/migrations/meta/_journal.json index 76fdca3..535af36 100644 --- a/apps/server/migrations/meta/_journal.json +++ b/apps/server/migrations/meta/_journal.json @@ -17,4 +17,4 @@ "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/apps/server/src/agents/instructions.ts b/apps/server/src/agents/instructions.ts index edeb9b5..284c7b8 100644 --- a/apps/server/src/agents/instructions.ts +++ b/apps/server/src/agents/instructions.ts @@ -55,6 +55,12 @@ FIRST — if any matches, stop with \`SKIPPED: \`. Otherwise, route by t decision table. This routing is deterministic: the same event always maps to the same skill. +**Durable inbox exception:** when the prompt includes a **PR discussion inbox**, +process every listed item with \`respond-to-comment\` even if the triggering +event would otherwise be skipped for being uninvolved or directed at another +user. The inbox was already admitted and is the outstanding work; do not return +\`SKIPPED\` before addressing it. + ### Skip conditions (check first, in order) 1. \`payload.sender.login\` equals \`$ME\` (self-triggered) — skip, EXCEPT for diff --git a/apps/server/src/db/schema.ts b/apps/server/src/db/schema.ts index 52285fa..f55cc7b 100644 --- a/apps/server/src/db/schema.ts +++ b/apps/server/src/db/schema.ts @@ -104,6 +104,13 @@ export const githubDiscussionObligations = sqliteTable( table.sourceKind, table.sourceCommentId, ), + index("idx_github_discussion_obligations_repo_pr_status_created").on( + table.repo, + table.prNumber, + table.status, + table.createdAt, + ), + index("idx_github_discussion_obligations_status_created").on(table.status, table.createdAt), index("idx_github_discussion_obligations_entity_status").on(table.entityKey, table.status), ], ) diff --git a/apps/server/src/lib/events/__tests__/discussion-retry.test.ts b/apps/server/src/lib/events/__tests__/discussion-retry.test.ts index 55e656d..8856770 100644 --- a/apps/server/src/lib/events/__tests__/discussion-retry.test.ts +++ b/apps/server/src/lib/events/__tests__/discussion-retry.test.ts @@ -1,11 +1,17 @@ -import { describe, expect, it } from "vitest" -import { DISCUSSION_RETRY_DELAY_MS, shouldRetryDiscussion } from "../discussion-retry" +import { describe, expect, it, vi } from "vitest" + +const { dispatchGitHubEvent } = vi.hoisted(() => ({ dispatchGitHubEvent: vi.fn().mockResolvedValue(undefined) })) +vi.mock("@/lib/github/dispatch", () => ({ dispatchGitHubEvent })) + +import { DISCUSSION_RETRY_DELAY_MS, retryOpenDiscussionObligations, shouldRetryDiscussion } from "../discussion-retry" describe("shouldRetryDiscussion", () => { const now = Date.UTC(2026, 7, 26, 12, 0, 0) - it("requeues an open obligation Jared has not been reminded about recently", () => { - expect(shouldRetryDiscussion({ status: "open", reminderCount: 0, lastRemindedAt: null }, now)).toBe(true) + it("requeues an open obligation only after the initial delivery has had time to complete", () => { + expect(shouldRetryDiscussion({ status: "open", reminderCount: 0, lastRemindedAt: new Date(now - 1) }, now)).toBe( + false, + ) expect( shouldRetryDiscussion( { status: "open", reminderCount: 1, lastRemindedAt: new Date(now - DISCUSSION_RETRY_DELAY_MS) }, @@ -21,4 +27,39 @@ describe("shouldRetryDiscussion", () => { ) expect(shouldRetryDiscussion({ status: "open", reminderCount: 3, lastRemindedAt: null }, now)).toBe(false) }) + + it("retries one representative event per PR and atomically claims its whole inbox batch", async () => { + const all = vi.fn().mockResolvedValue({ + results: [ + { + id: "oldest", + repo: "getsentry/cli", + pr_number: 1484, + entity_key: "getsentry/cli#1484", + reminder_count: 0, + last_reminded_at: Math.floor((now - DISCUSSION_RETRY_DELAY_MS) / 1000), + status: "open", + event_id: "event-1", + event: "issue_comment", + action: "created", + delivery_id: "delivery-1", + sender: "maintainer", + event_repo: "getsentry/cli", + installation_id: 42, + payload: "{}", + }, + ], + }) + const run = vi.fn().mockResolvedValue({ meta: { changes: 2 } }) + const prepare = vi.fn((query: string) => ({ + bind: vi.fn(() => (query.startsWith("SELECT") ? { all } : { run })), + })) + + const result = await retryOpenDiscussionObligations({ ENV: "test", DB: { prepare } } as never, now) + + expect(result).toEqual({ retried: 1, needsHuman: 0 }) + expect(dispatchGitHubEvent).toHaveBeenCalledTimes(1) + expect(prepare.mock.calls[0]?.[0]).toContain("earlier.pr_number = o.pr_number") + expect(prepare.mock.calls[1]?.[0]).toContain("WHERE repo = ? AND pr_number = ?") + }) }) diff --git a/apps/server/src/lib/events/discussion-retry.ts b/apps/server/src/lib/events/discussion-retry.ts index b4db2b6..5fd14b6 100644 --- a/apps/server/src/lib/events/discussion-retry.ts +++ b/apps/server/src/lib/events/discussion-retry.ts @@ -24,6 +24,8 @@ export function shouldRetryDiscussion(candidate: RetryCandidate, now = Date.now( type DiscussionRetryRow = { id: string + repo: string + pr_number: number entity_key: string reminder_count: number last_reminded_at: number | null @@ -33,7 +35,7 @@ type DiscussionRetryRow = { action: string | null delivery_id: string sender: string | null - repo: string | null + event_repo: string | null installation_id: number | null payload: string } @@ -41,22 +43,33 @@ type DiscussionRetryRow = { export type DiscussionRetryResult = { retried: number; needsHuman: number } /** - * Wake Jared with the original event after a missed discussion inbox. We reuse - * the event so dispatch renders all still-open obligations for that same PR; - * after three missed reminders the row is explicitly marked needs_human. + * Wake Jared with one original event per PR after a missed discussion inbox. + * Reusing one event renders every still-open obligation for that PR, avoiding + * a burst of duplicate prompts when several reviewers wrote at once. After + * three missed reminders, the entire outstanding PR batch becomes needs_human. */ export async function retryOpenDiscussionObligations( env: Env, scheduledTime: number, opts: { maxRows?: number } = {}, ): Promise { - const maxRows = opts.maxRows ?? 50 + // A retry can cold-start a sandbox. Keep one cron invocation bounded so it + // cannot monopolize the Worker when several old discussions need attention. + const maxRows = opts.maxRows ?? 10 const rows = await env.DB.prepare( - `SELECT o.id, o.entity_key, o.reminder_count, o.last_reminded_at, o.status, o.event_id, - e.event, e.action, e.delivery_id, e.sender, e.repo, e.installation_id, e.payload + `SELECT o.id, o.repo, o.pr_number, o.entity_key, o.reminder_count, o.last_reminded_at, o.status, o.event_id, + e.event, e.action, e.delivery_id, e.sender, e.repo AS event_repo, e.installation_id, e.payload FROM github_discussion_obligations o JOIN webhook_events e ON e.id = o.event_id WHERE o.status = 'open' + AND NOT EXISTS ( + SELECT 1 + FROM github_discussion_obligations earlier + WHERE earlier.status = 'open' + AND earlier.repo = o.repo + AND earlier.pr_number = o.pr_number + AND (earlier.created_at < o.created_at OR (earlier.created_at = o.created_at AND earlier.id < o.id)) + ) ORDER BY o.created_at ASC LIMIT ?`, ) @@ -76,20 +89,30 @@ export async function retryOpenDiscussionObligations( } if (candidate.reminderCount >= MAX_DISCUSSION_REMINDERS) { await env.DB.prepare( - "UPDATE github_discussion_obligations SET status = 'needs_human', updated_at = ? WHERE id = ? AND status = 'open'", + "UPDATE github_discussion_obligations SET status = 'needs_human', updated_at = ? WHERE repo = ? AND pr_number = ? AND status = 'open'", ) - .bind(Math.floor(scheduledTime / 1000), row.id) + .bind(Math.floor(scheduledTime / 1000), row.repo, row.pr_number) .run() needsHuman += 1 continue } if (!shouldRetryDiscussion(candidate, scheduledTime)) continue - await env.DB.prepare( - "UPDATE github_discussion_obligations SET reminder_count = reminder_count + 1, last_reminded_at = ?, updated_at = ? WHERE id = ? AND status = 'open'", + const retriedRow = await env.DB.prepare( + "UPDATE github_discussion_obligations SET reminder_count = reminder_count + 1, last_reminded_at = ?, updated_at = ? WHERE repo = ? AND pr_number = ? AND status = 'open' AND reminder_count < ? AND (last_reminded_at IS NULL OR last_reminded_at <= ?)", ) - .bind(Math.floor(scheduledTime / 1000), Math.floor(scheduledTime / 1000), row.id) + .bind( + Math.floor(scheduledTime / 1000), + Math.floor(scheduledTime / 1000), + row.repo, + row.pr_number, + MAX_DISCUSSION_REMINDERS, + Math.floor((scheduledTime - DISCUSSION_RETRY_DELAY_MS) / 1000), + ) .run() + // Another overlapping cron may have reclaimed this row while we were + // reading it. Only the execution that won the guarded update dispatches. + if ((retriedRow.meta.changes ?? 0) === 0) continue retried += 1 await dispatchGitHubEvent(env, db, logger, { eventId: row.event_id, @@ -98,7 +121,7 @@ export async function retryOpenDiscussionObligations( action: row.action, deliveryId: row.delivery_id, sender: row.sender, - repo: row.repo, + repo: row.event_repo, installationId: row.installation_id, payload: row.payload, }) diff --git a/apps/server/src/lib/github/__tests__/discussion-store.test.ts b/apps/server/src/lib/github/__tests__/discussion-store.test.ts index f46c210..2502675 100644 --- a/apps/server/src/lib/github/__tests__/discussion-store.test.ts +++ b/apps/server/src/lib/github/__tests__/discussion-store.test.ts @@ -31,5 +31,6 @@ describe("makeDiscussionRecord", () => { eventId: "delivery-1", }) expect(record.id).toMatch(/^[a-f0-9-]{36}$/) + expect(record.lastRemindedAt).toEqual(new Date("2026-08-26T13:31:00Z")) }) }) diff --git a/apps/server/src/lib/github/__tests__/discussions.test.ts b/apps/server/src/lib/github/__tests__/discussions.test.ts index d103183..e1a9631 100644 --- a/apps/server/src/lib/github/__tests__/discussions.test.ts +++ b/apps/server/src/lib/github/__tests__/discussions.test.ts @@ -139,6 +139,8 @@ describe("formatDiscussionInbox", () => { { id: "abc123", kind: "top_level", + sourceCommentId: "101", + replyToCommentId: null, author: "BYK", body: "Could we rename all sixel flags?", url: "https://github.com/getsentry/cli/pull/1484#issuecomment-101", @@ -146,6 +148,8 @@ describe("formatDiscussionInbox", () => { { id: "def456", kind: "inline", + sourceCommentId: "102", + replyToCommentId: "99", author: "MathurAditya724", body: "Jared, what happened to it?", url: "https://github.com/getsentry/cli/pull/1484#discussion_r102", @@ -157,6 +161,7 @@ describe("formatDiscussionInbox", () => { expect(inbox).toContain("Jared, what happened to it?") expect(inbox).toContain("") expect(inbox).toContain("do not send a generic acknowledgement") + expect(inbox).toContain("Reply in this review thread via top-level comment ID: 99") }) }) @@ -208,19 +213,29 @@ describe("responseEvidenceFromWebhook", () => { expect(evidence).toBeNull() }) - it("requires an inline response to be on the obligation's specific comment", () => { + it("requires an inline response to be on the obligation's specific review thread", () => { const matches = responseMatchesDiscussion( - { kind: "inline", prNumber: 1484, sourceCommentId: "102" }, - { obligationId: "abc123", outcome: "addressed", prNumber: 1484, replyToCommentId: "99" }, + { kind: "inline", prNumber: 1484, sourceCommentId: "102", replyToCommentId: "99" }, + { obligationId: "abc123", outcome: "addressed", prNumber: 1484, replyToCommentId: "100" }, ) expect(matches).toBe(false) }) - it("only accepts a response for the same pull request and exact inline parent", () => { + it("only accepts a response for the same pull request and exact inline thread", () => { const response = { obligationId: "abc123", outcome: "addressed" as const, prNumber: 1484, replyToCommentId: "102" } - expect(responseMatchesDiscussion({ kind: "inline", prNumber: 1484, sourceCommentId: "102" }, response)).toBe(true) - expect(responseMatchesDiscussion({ kind: "inline", prNumber: 1485, sourceCommentId: "102" }, response)).toBe(false) + expect( + responseMatchesDiscussion( + { kind: "inline", prNumber: 1484, sourceCommentId: "103", replyToCommentId: "102" }, + response, + ), + ).toBe(true) + expect( + responseMatchesDiscussion( + { kind: "inline", prNumber: 1485, sourceCommentId: "103", replyToCommentId: "102" }, + response, + ), + ).toBe(false) }) }) diff --git a/apps/server/src/lib/github/discussion-store.ts b/apps/server/src/lib/github/discussion-store.ts index b69d7c3..869c566 100644 --- a/apps/server/src/lib/github/discussion-store.ts +++ b/apps/server/src/lib/github/discussion-store.ts @@ -40,7 +40,9 @@ export function makeDiscussionRecord(input: DiscussionRecordInput) { outcome: null, verifiedAt: null, reminderCount: 0, - lastRemindedAt: null, + // The initial webhook already prompted Jared. Do not turn the first cron + // after a comment into an immediate duplicate delivery. + lastRemindedAt: now, createdAt: now, updatedAt: now, } @@ -73,7 +75,7 @@ export async function recordDiscussionObligation(db: Db, input: DiscussionRecord outcome: null, verifiedAt: null, reminderCount: 0, - lastRemindedAt: null, + lastRemindedAt: record.lastRemindedAt, updatedAt: record.updatedAt, }, }) @@ -88,6 +90,8 @@ export async function listOpenDiscussionObligations( .select({ id: dbSchema.githubDiscussionObligations.id, kind: dbSchema.githubDiscussionObligations.sourceKind, + sourceCommentId: dbSchema.githubDiscussionObligations.sourceCommentId, + replyToCommentId: dbSchema.githubDiscussionObligations.replyToCommentId, author: dbSchema.githubDiscussionObligations.author, body: dbSchema.githubDiscussionObligations.body, url: dbSchema.githubDiscussionObligations.url, @@ -155,6 +159,7 @@ export async function verifyDiscussionResponse( kind: obligation.sourceKind as DiscussionObligation["kind"], prNumber: obligation.prNumber, sourceCommentId: obligation.sourceCommentId, + replyToCommentId: obligation.replyToCommentId, }, response, ) diff --git a/apps/server/src/lib/github/discussions.ts b/apps/server/src/lib/github/discussions.ts index 1efc495..85a6ba9 100644 --- a/apps/server/src/lib/github/discussions.ts +++ b/apps/server/src/lib/github/discussions.ts @@ -27,7 +27,10 @@ export type DiscussionResponseEvidence = DiscussionResponseMarker & { replyToCommentId: string | null } -export type OpenDiscussionObligation = Pick & { id: string } +export type OpenDiscussionObligation = Pick< + DiscussionObligation, + "kind" | "sourceCommentId" | "replyToCommentId" | "author" | "body" | "url" +> & { id: string } export type DiscussionSourceReference = Pick @@ -169,6 +172,11 @@ export function formatDiscussionInbox(obligations: OpenDiscussionObligation[]): .map( (obligation, index) => `### ${index + 1}. ${obligation.kind.replace("_", " ")} from ${obligation.author} ${obligation.url ? `${obligation.url}\n` : ""} +${ + obligation.kind === "inline" + ? `Reply in this review thread via top-level comment ID: ${obligation.replyToCommentId ?? obligation.sourceCommentId} (the message above is comment ID: ${obligation.sourceCommentId})\n` + : "" +} ${obligation.body} After your substantive reply, append \`\`, where \`\` is \`addressed\`, \`explained\`, or \`needs-human\`.`, @@ -222,11 +230,12 @@ export function responseEvidenceFromWebhook( } } -/** A receipt must belong to the same PR and, for review threads, reply to it. */ +/** A receipt must belong to the same PR and, for review threads, reply to its root. */ export function responseMatchesDiscussion( - obligation: Pick, + obligation: Pick, response: DiscussionResponseEvidence, ): boolean { if (obligation.prNumber !== response.prNumber) return false - return obligation.kind !== "inline" || response.replyToCommentId === obligation.sourceCommentId + const threadRoot = obligation.replyToCommentId ?? obligation.sourceCommentId + return obligation.kind !== "inline" || response.replyToCommentId === threadRoot } diff --git a/apps/server/src/routes/containers/index.ts b/apps/server/src/routes/containers/index.ts index 9f13885..1fed008 100644 --- a/apps/server/src/routes/containers/index.ts +++ b/apps/server/src/routes/containers/index.ts @@ -1163,6 +1163,9 @@ const router = new Hono() await Promise.all([ db.delete(dbSchema.agentSessions).where(eq(dbSchema.agentSessions.entityKey, entityKey)), db.delete(dbSchema.webhookEvents).where(eq(dbSchema.webhookEvents.entityKey, entityKey)), + db + .delete(dbSchema.githubDiscussionObligations) + .where(eq(dbSchema.githubDiscussionObligations.entityKey, entityKey)), ]) } catch { /* best effort */ diff --git a/apps/server/src/routes/events/index.ts b/apps/server/src/routes/events/index.ts index ca333cb..9197bcd 100644 --- a/apps/server/src/routes/events/index.ts +++ b/apps/server/src/routes/events/index.ts @@ -1,6 +1,6 @@ import { and, desc, eq, like, sql } from "drizzle-orm" import { Hono } from "hono" -import { webhookEvents } from "@/db/schema" +import { githubDiscussionObligations, webhookEvents } from "@/db/schema" import { dispatchGitHubEvent } from "@/lib/github/dispatch" import { isAuthenticated } from "@/middlewares" import type { AuthEnv } from "@/types" @@ -9,7 +9,7 @@ const router = new Hono() .use(isAuthenticated()) .delete("/", async (c) => { const db = c.get("db") - await db.delete(webhookEvents) + await Promise.all([db.delete(webhookEvents), db.delete(githubDiscussionObligations)]) return c.json({ ok: true }) }) .get("/grouped", async (c) => {