diff --git a/.env.example b/.env.example index 533e4d789..de52a46b7 100644 --- a/.env.example +++ b/.env.example @@ -124,6 +124,8 @@ GITHUB_REPOS_TO_MONITOR=owner/repo1,owner/repo2 # ProPR pulls events from the GitHub API on this interval using any usable # GitHub auth (relay or your own App). POLLING_INTERVAL_MS is the poll period in # milliseconds (default: 60000). +# NOTE: `/split` requires issue_comment event delivery and is not accepted in +# polling mode. Use routing_websocket or direct_webhook for `/split` commands. POLLING_INTERVAL_MS=60000 # Config Repository (for dynamic repository management) @@ -142,6 +144,12 @@ GITHUB_ISSUE_QUEUE_NAME=github-issue-processor WORKER_CONCURRENCY=2 COMMENT_BATCH_DELAY_MS=3000 +# Webhook `/split` commands are recognized, durably recorded, and receive a +# disabled response even while this is false. Set true only when a deployed +# split-operation worker is ready to consume the queued operations it enables. +# `/split` remains unavailable to polling intake in either state. +# PR_SPLIT_EXECUTION_ENABLED=false + # Summarization fallback/cooldown controls # Promote the configured summarization fallback to primary after this many # primary quota failures for the same agent/model. Defaults to 3. @@ -322,6 +330,8 @@ PROPR_DEMO_MODE=false # --- SQLite Database --- # Path to SQLite database file (will be created if it doesn't exist) DB_FILENAME=./data/propr.sqlite +# How long each SQLite connection waits for another process's writer lock. +# SQLITE_BUSY_TIMEOUT_MS=30000 # --- PR Preview Environment --- # docker-compose.yml and scripts/deploy-pr.sh consume these local names. diff --git a/docs/docs/operations/configuration-reference.md b/docs/docs/operations/configuration-reference.md index b0a8df665..6659d7adb 100644 --- a/docs/docs/operations/configuration-reference.md +++ b/docs/docs/operations/configuration-reference.md @@ -33,6 +33,7 @@ The backend authenticates to GitHub in one of three modes — `demo`, `relay`, o | `LOG_LEVEL` | `info` | Log verbosity across services. | Optional. | | `NODE_ENV` | `development` | Node environment; use `production` on servers. | Optional. | | `DB_FILENAME` | `./data/propr.sqlite` | Path to the SQLite database file (created if it doesn't exist). | Optional. | +| `SQLITE_BUSY_TIMEOUT_MS` | `30000` | Maximum time each SQLite connection waits for a writer in another process before reporting contention. | Optional. | ## Event Intake @@ -46,6 +47,7 @@ How ProPR receives GitHub events, plus what it watches for once they arrive. All | `PROPR_ROUTING_WS_PONG_TIMEOUT_MS` | `30000` (30 seconds) | Maximum wait for a transport pong before the stale socket is terminated and reconnected. | Optional. | | `POLLING_INTERVAL_MS` | `60000` | Poll period when pulling events from the GitHub API. | Polling mode only. | | `GH_WEBHOOK_SECRET` | Unset | Shared secret GitHub signs webhook deliveries with. | Direct webhook mode. | +| `PR_SPLIT_EXECUTION_ENABLED` | `false` | Allows eligible `/split` commands to create queued operations. Even when false, webhook intake recognizes and stores the command and posts a disabled response. Accepted true values are `1` and case-insensitive `true`. | Only with `routing_websocket` or `direct_webhook` intake and a deployed split-operation worker. | | `GITHUB_REPOS_TO_MONITOR` | Placeholder (`owner/repo1,owner/repo2`) | Comma-separated repositories the daemon watches. | Always. | | `CONFIG_REPO` | Example config repo URL | Git repository for dynamic repository management; when set, processing labels and repo config load from it. | Optional. | | `PRIMARY_PROCESSING_LABELS` | Shipped `AI,propr` / code falls back to `AI` | Issue labels that trigger processing. | Optional. | @@ -56,6 +58,8 @@ How ProPR receives GitHub events, plus what it watches for once they arrive. All | `PR_FOLLOWUP_TRIGGER_KEYWORDS` | `!propr` | Keywords in PR comments that trigger follow-up work. See [PR Follow-up](../features/pr-followup.md). | Optional. | | `LABEL_APPLIER_TIMELINE_MAX_PAGES` | `5` | With a whitelist set, polling resolves who applied the trigger label from the issue timeline (page 1 + the most recent N pages). Raise it if long-lived issues are skipped with "Could not determine label applier". | Optional. | +> **`/split` intake limitation:** `/split` commands require an `issue_comment` event and are recognized only in `routing_websocket` or `direct_webhook` mode. Polling intentionally ignores them and does not post a response. With execution disabled (the default), webhook commands still create durable receipts and a disabled comment but never create an operation. Enable execution only alongside a deployed split-operation worker. Intake durably suppresses responses after five commands from the same repository/user identity in ten minutes. + ## Agents & Timeouts Unified image selection, per-agent credential paths, and execution limits. Coding-agent task executions default to 24 hours; analysis calls keep their separate, shorter timeouts. diff --git a/docs/docs/operations/deployment.md b/docs/docs/operations/deployment.md index 816aced45..985052cc0 100644 --- a/docs/docs/operations/deployment.md +++ b/docs/docs/operations/deployment.md @@ -124,6 +124,8 @@ in [ProPR Connect](./propr-connect.md). **Polling** (`GITHUB_EVENT_INTAKE_MODE=polling`) suits installs that prefer to pull rather than maintain a streaming connection; it needs no inbound endpoint but adds latency and consumes the API budget continuously. The interval is `POLLING_INTERVAL_MS` (default `60000`). +> **Command limitation:** polling does not accept `/split` PR comments and does not post a refusal. `/split` requires `issue_comment` event delivery, so use `routing_websocket` or `direct_webhook` for that command. + **Direct webhook** (`GITHUB_EVENT_INTAKE_MODE=direct_webhook`) is for running your own GitHub App with GitHub delivering events to a public endpoint: ```bash diff --git a/package.json b/package.json index a74d2497e..5a0520855 100644 --- a/package.json +++ b/package.json @@ -17,16 +17,16 @@ "lint": "eslint src/", "typecheck": "tsc --noEmit", "test": "node --test", - "test:unit": "NODE_ENV=test npx tsx --test test/minimal.test.ts test/modelName.test.ts test/daemonEventIntake.test.ts test/githubEventIntakeMode.test.ts test/intakeModePrerequisites.test.ts test/validateRoutingUrl.test.ts test/routingWebSocketProtocol.test.ts test/routingWebSocketIntakeService.test.ts test/routingStatusPublisher.test.ts packages/api/test/statusRoutes.test.ts packages/api/test/agentRuntimeRoutes.test.ts packages/api/test/instanceAuthorization.test.ts packages/api/test/routeAuthorization.test.ts", + "test:unit": "NODE_ENV=test npx tsx --test test/minimal.test.ts test/modelName.test.ts test/daemonEventIntake.test.ts test/githubEventIntakeMode.test.ts test/intakeModePrerequisites.test.ts test/validateRoutingUrl.test.ts test/routingWebSocketProtocol.test.ts test/routingWebSocketIntakeService.test.ts test/routingStatusPublisher.test.ts test/prSplit/commandAuthorization.test.ts test/prSplit/operationStore.test.ts test/prSplit/intake.test.ts test/prSplit/interception.test.ts packages/api/test/statusRoutes.test.ts packages/api/test/agentRuntimeRoutes.test.ts packages/api/test/instanceAuthorization.test.ts packages/api/test/routeAuthorization.test.ts", "test:e2e": "npx tsx --test test/e2e.test.ts", - "test:docker": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker npx tsx --test test/*.test.ts", + "test:docker": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker npx tsx --test test/*.test.ts test/prSplit/*.test.ts", "test:docker:single": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker npx tsx --test", - "test:docker:quick": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'npx tsx --test test/*.test.ts --test-exclude=\"**/agentRegistry.test.ts\"'", - "test:docker:timeout": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker timeout 30 npx tsx --test test/*.test.ts", + "test:docker:quick": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'npx tsx --test test/*.test.ts test/prSplit/*.test.ts --test-exclude=\"**/agentRegistry.test.ts\"'", + "test:docker:timeout": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker timeout 30 npx tsx --test test/*.test.ts test/prSplit/*.test.ts", "test:docker:clean": "docker ps --filter 'name=propr-worker-run' -q | xargs -r docker stop; npm run test:docker", - "test:docker:verbose": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts; do echo \"=== Testing: $f ===\"; npx tsx --test \"$f\" || echo \"FAILED: $f\"; done'", - "test:docker:find-hanging": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts; do echo \"=== Testing: $f ===\"; timeout 5 npx tsx --test \"$f\" > /dev/null 2>&1 && echo \"✓ PASSED\" || echo \"✗ FAILED or TIMEOUT\"; done'", - "test:docker:stable": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts; do [[ \"$f\" == *\"agentRegistry\"* ]] && continue; echo \"=== Testing: $f ===\"; timeout 10 npx tsx --test \"$f\" > /dev/null 2>&1 && echo \"✓ PASSED\" || echo \"✗ FAILED or TIMEOUT\"; done'", + "test:docker:verbose": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts test/prSplit/*.test.ts; do echo \"=== Testing: $f ===\"; npx tsx --test \"$f\" || echo \"FAILED: $f\"; done'", + "test:docker:find-hanging": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts test/prSplit/*.test.ts; do echo \"=== Testing: $f ===\"; timeout 5 npx tsx --test \"$f\" > /dev/null 2>&1 && echo \"✓ PASSED\" || echo \"✗ FAILED or TIMEOUT\"; done'", + "test:docker:stable": "docker-compose run --rm -e REDIS_HOST=redis -e NODE_ENV=test worker sh -c 'for f in test/*.test.ts test/prSplit/*.test.ts; do [[ \"$f\" == *\"agentRegistry\"* ]] && continue; echo \"=== Testing: $f ===\"; timeout 10 npx tsx --test \"$f\" > /dev/null 2>&1 && echo \"✓ PASSED\" || echo \"✗ FAILED or TIMEOUT\"; done'", "start": "node dist/src/index.js", "dev": "tsx watch src/index.ts", "daemon": "node dist/src/daemon.js", diff --git a/packages/core/src/agents/impl/OpenCodeAgent.ts b/packages/core/src/agents/impl/OpenCodeAgent.ts index 7d7faa5b2..7c87ea819 100644 --- a/packages/core/src/agents/impl/OpenCodeAgent.ts +++ b/packages/core/src/agents/impl/OpenCodeAgent.ts @@ -3,7 +3,7 @@ import path from 'path'; import { execSync } from 'child_process'; import logger from '../../utils/logger.js'; import { Agent, AgentConfig, AgentTaskOptions, AgentExecutionResult, AnalysisResult, AnalyzeOptions } from '../types.js'; -import { executeDockerCommand } from '../../claude/docker/dockerExecutor.js'; +import { executeDockerCommand, type ExecutionResult } from '../../claude/docker/dockerExecutor.js'; import { verifyWorktreeStructure, verifyWorktreePostExecution, setWorktreeOwnership, UsageLimitError } from '../../claude/claudeHelpers.js'; import { resolveConfigPath } from '../../config/configManager.js'; import { persistLlmLog, createLlmLogFromAnalysis, createLlmLogFromAgentExecution, buildTaskWorkRef, buildAnalysisWorkRef, formatUsageMetrics } from '../../utils/llmLogger.js'; @@ -18,6 +18,14 @@ export { UsageLimitError }; const DEFAULT_OPENCODE_ANALYSIS_ROOT = '/tmp/git-processor/opencode-analysis'; +function resolveOpenCodeExecutionOutcome( + result: ExecutionResult, + parsedOutput: ParsedOpenCodeOutput +): Pick { + const terminationReason = resolveAgentTerminationReason({ timedOut: result.timedOut, error: parsedOutput.error || result.stderr }); + return { success: result.exitCode === 0 && !parsedOutput.error && !terminationReason, terminationReason }; +} + export class OpenCodeAgent implements Agent { readonly config: AgentConfig; private readonly timeoutMs: number; @@ -69,8 +77,7 @@ export class OpenCodeAgent implements Agent { const executionTime = Date.now() - startTime; const parsedOutput = this.parseOpenCodeJsonl(result.stdout); const modelUsed = parsedOutput.modelUsed || effectiveModel || 'unknown'; - const terminationReason = resolveAgentTerminationReason({ timedOut: result.timedOut, error: parsedOutput.error || result.stderr }); - const success = result.exitCode === 0 && !parsedOutput.error && !terminationReason; + const { success, terminationReason } = resolveOpenCodeExecutionOutcome(result, parsedOutput); const errorText = success ? undefined : (parsedOutput.error || result.stderr || `OpenCode exited with code ${result.exitCode ?? 'unknown'}`); const response: AgentExecutionResult = { success, diff --git a/packages/core/src/agents/impl/VibeAgent.ts b/packages/core/src/agents/impl/VibeAgent.ts index 793633779..26b48a9b0 100644 --- a/packages/core/src/agents/impl/VibeAgent.ts +++ b/packages/core/src/agents/impl/VibeAgent.ts @@ -1,7 +1,7 @@ import fs from 'fs'; import logger from '../../utils/logger.js'; import { Agent, AgentConfig, AgentTaskOptions, AgentExecutionResult, AnalysisResult, AnalyzeOptions } from '../types.js'; -import { executeDockerCommand } from '../../claude/docker/dockerExecutor.js'; +import { executeDockerCommand, type ExecutionResult } from '../../claude/docker/dockerExecutor.js'; import { wrapDockerRunArgsWithRepoSetup } from '../../claude/docker/repoSetupWrapper.js'; import { verifyWorktreeStructure, verifyWorktreePostExecution, setWorktreeOwnership, UsageLimitError } from '../../claude/claudeHelpers.js'; import { resolveConfigPath, loadSettings } from '../../config/configManager.js'; @@ -21,6 +21,14 @@ export { getMistralApiKeyFromSettings, readLatestVibeSessionTokenUsage } from '. const DEFAULT_VIBE_MAX_TURNS = 1000; const CONTAINER_CONFIG_PATH = '/home/node/.vibe'; +function resolveVibeExecutionOutcome( + result: ExecutionResult, + parsedOutput: ReturnType +): Pick { + const terminationReason = resolveAgentTerminationReason({ timedOut: result.timedOut, error: parsedOutput.error || result.stderr }); + return { success: isSuccessfulVibeResult(result.exitCode, parsedOutput) && !terminationReason, terminationReason }; +} + interface VibeDockerArgsParams { worktreePath: string; githubToken: string; @@ -106,8 +114,7 @@ export class VibeAgent implements Agent { const conversationLog = parseVibeConversationLog(result.stdout); const tokenUsage = parsedOutput.tokenUsage || readLatestVibeSessionTokenUsage(runtimeHomePath); const modelUsed = parsedOutput.model || effectiveModel || 'unknown'; - const terminationReason = resolveAgentTerminationReason({ timedOut: result.timedOut, error: parsedOutput.error || result.stderr }); - const success = isSuccessfulVibeResult(result.exitCode, parsedOutput) && !terminationReason; + const { success, terminationReason } = resolveVibeExecutionOutcome(result, parsedOutput); const error = success ? undefined : buildVibeFailureMessage(result, parsedOutput); if (parsedOutput.sessionId && onSessionId) onSessionId(parsedOutput.sessionId); diff --git a/packages/core/src/claude/claudeHelpers.ts b/packages/core/src/claude/claudeHelpers.ts index 8e4851335..ec6fac5cc 100644 --- a/packages/core/src/claude/claudeHelpers.ts +++ b/packages/core/src/claude/claudeHelpers.ts @@ -460,7 +460,6 @@ export async function storePromptInRedis(options: StorePromptOptions): Promise { + table.text('id').primary(); + table.bigInteger('repository_id').notNullable(); + table.text('repository').notNullable(); + table.integer('source_pr_number').notNullable(); + table.text('base_ref').notNullable(); + table.text('base_sha').notNullable(); + table.text('head_sha').notNullable(); + table.bigInteger('requester_id').notNullable(); + table.text('requester').notNullable(); + table.bigInteger('original_comment_id').notNullable(); + table.text('instruction').notNullable().defaultTo(''); + table.text('event_key').notNullable(); + table.text('dedupe_key').notNullable(); + table.text('status').notNullable().defaultTo('queued').checkIn([ + 'queued', + 'running', + 'completed', + 'failed', + ]); + table.text('error_message').nullable(); + table.timestamp('started_at').nullable(); + table.timestamp('heartbeat_at').nullable(); + table.timestamp('lease_expires_at').nullable(); + table.text('lease_token').nullable(); + table.timestamp('finished_at').nullable(); + table.timestamp('created_at').notNullable().defaultTo(knex.fn.now()); + table.timestamp('updated_at').notNullable().defaultTo(knex.fn.now()); + + table.index(['repository_id', 'source_pr_number']); + table.index('status'); + table.index('original_comment_id'); + table.index('dedupe_key'); + }); + + await knex.raw(` + CREATE UNIQUE INDEX pr_split_operations_event_key_unique + ON pr_split_operations (event_key) + `); + + await knex.raw(` + CREATE UNIQUE INDEX pr_split_operations_semantic_dedupe + ON pr_split_operations (dedupe_key) + WHERE status != 'failed' + `); + + await knex.raw(` + CREATE UNIQUE INDEX pr_split_operations_one_active_per_pr + ON pr_split_operations (repository_id, source_pr_number) + WHERE status IN ('queued', 'running') + `); + + await knex.schema.createTable('pr_split_command_receipts', (table) => { + table.text('event_key').primary(); + table.bigInteger('repository_id').notNullable(); + table.text('repository').notNullable(); + table.integer('source_pr_number').notNullable(); + table.bigInteger('requester_id').notNullable(); + table.text('requester').notNullable(); + table.bigInteger('original_comment_id').notNullable(); + table.text('instruction').notNullable().defaultTo(''); + table.text('outcome').notNullable().checkIn([ + 'processing', + 'disabled', + 'unauthorized', + 'closed', + 'invalid', + 'rate_limited', + 'queued', + 'duplicate', + 'active', + ]); + table.text('duplicate_kind').nullable().checkIn(['event', 'semantic']); + table.text('operation_id') + .nullable() + .references('id') + .inTable('pr_split_operations') + .onDelete('SET NULL'); + table.text('response_state').notNullable().defaultTo('pending').checkIn([ + 'pending', + 'claimed', + 'posted', + 'suppressed', + ]); + table.text('response_claim_token').nullable(); + table.timestamp('response_claimed_at').nullable(); + table.bigInteger('response_comment_id').nullable(); + table.timestamp('response_posted_at').nullable(); + table.timestamp('created_at').notNullable().defaultTo(knex.fn.now()); + table.timestamp('updated_at').notNullable().defaultTo(knex.fn.now()); + + table.unique(['repository_id', 'original_comment_id']); + table.index(['repository_id', 'source_pr_number']); + table.index(['repository_id', 'requester_id', 'created_at']); + table.index('operation_id'); + table.index('response_state'); + }); +} + +export async function down(knex) { + await knex.schema.dropTableIfExists('pr_split_command_receipts'); + await knex.schema.dropTableIfExists('pr_split_operations'); +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index b496822c0..db687048d 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -122,7 +122,7 @@ export { areAllChecksPassing, buildRedisRuntimeConfig, closeUltrafixStateRedis, export type { CheckRunsStatus, ActivePRWork, ActivePRTask, ActivePRQueuedJob } from './webhook/checkRunHelpers.js'; export { handleCheckRunEvent, handleStatusEvent, reevaluatePRAutoMerge, setUltrafixCheckRunHook, type StatusEventPayload } from './webhook/checkRunHandler.js'; export { processWebhookEvent, initializeWebhookHandler, SUPPORTED_WEBHOOK_EVENTS } from './webhook/webhookHandler.js'; -export type { WebhookEventType, DetectedIssue, IssueProcessor, CommentProcessor, CommentDeletedHandler, CommentEditedHandler, CheckRunProcessor, WebhookHandlerOptions } from './webhook/webhookHandler.js'; +export type { WebhookEventType, DetectedIssue, IssueProcessor, CommentProcessor, CommentDeletedHandler, CommentEditedHandler, CheckRunProcessor, SplitCommentHandler, WebhookHandlerOptions } from './webhook/webhookHandler.js'; export { RoutingWebSocketIntakeService } from './intake/RoutingWebSocketIntakeService.js'; export type { RoutingWebSocketIntakeServiceOptions, RoutingWebSocketStatus, MinimalWebSocket, RawData, WebSocketCtor, FetchLike, DeliveryAckBilling, DeliveryAckEvidence, DeliveryAckStatus, DeliveryDisposition } from './intake/RoutingWebSocketIntakeService.js'; // The routing wire-protocol primitives (BoundedDeliverySet, BoundedTokenCache, @@ -147,6 +147,7 @@ export type { PlanIssueStatus as StatusMachinePlanIssueStatus } from './webhook/ export { getExecutionAnalysis } from './services/analysisService.js'; export { getModelPricing } from './services/pricingService.js'; +export * from './services/prSplit/index.js'; export { getWorktreeChanges, storeFileChanges, getStoredFileChanges, clearFileChanges, updateFileChangesFromWorktree, getCommitChanges, isValidCommitHash } from './services/worktreeMonitorService.js'; export type { FileChange, FileChangesData } from './services/worktreeMonitorService.js'; export { generateContext, generateAdditionalContext, SecurityException } from './services/context/index.js'; diff --git a/packages/core/src/services/prSplit/authorization.ts b/packages/core/src/services/prSplit/authorization.ts new file mode 100644 index 000000000..6954adfe6 --- /dev/null +++ b/packages/core/src/services/prSplit/authorization.ts @@ -0,0 +1,87 @@ +import { normalizeGitHubId } from './keys.js'; + +export const SPLIT_AUTHORIZED_PERMISSIONS = ['write', 'maintain', 'admin'] as const; + +export type SplitAuthorizedPermission = (typeof SPLIT_AUTHORIZED_PERMISSIONS)[number]; + +export interface PrSplitGitHubResponse { + data: unknown; + headers?: Record; +} + +/** The only GitHub client capability used by `/split` intake. */ +export interface PrSplitRequestClient { + request( + route: string, + parameters: Record, + ): Promise; +} + +export interface SplitAuthorizationRequest { + owner: string; + repo: string; + username: string; + requesterId: number; +} + +export type SplitAuthorizationResult = + | { authorized: true; permission: SplitAuthorizedPermission } + | { authorized: false; permission: string | null }; + +type UnknownRecord = Record; + +function isRecord(value: unknown): value is UnknownRecord { + return typeof value === 'object' && value !== null; +} + +/** Map GitHub's collaborator permission levels to `/split` authorization. */ +export function isSplitPermissionAuthorized( + permission: string | null | undefined, +): permission is SplitAuthorizedPermission { + return SPLIT_AUTHORIZED_PERMISSIONS.includes(permission as SplitAuthorizedPermission); +} + +function githubStatus(error: unknown): number | undefined { + if (!isRecord(error)) return undefined; + return typeof error.status === 'number' ? error.status : undefined; +} + +/** + * Fail-closed repository authorization for a split requester. + * + * A collaborator 404 is terminal only after the same credential proves that + * it can still read the repository. Repository/installation failures remain + * retryable instead of becoming an immutable authorization refusal. + */ +export async function authorizeSplitRequester( + octokit: PrSplitRequestClient, + request: SplitAuthorizationRequest, +): Promise { + const requesterId = normalizeGitHubId(request.requesterId, 'requesterId'); + try { + const { data } = await octokit.request( + 'GET /repos/{owner}/{repo}/collaborators/{username}/permission', + { owner: request.owner, repo: request.repo, username: request.username }, + ); + const permission = isRecord(data) && typeof data.permission === 'string' + ? data.permission + : null; + const responseUserId = isRecord(data) && isRecord(data.user) && typeof data.user.id === 'number' + ? data.user.id + : null; + + return responseUserId === requesterId && isSplitPermissionAuthorized(permission) + ? { authorized: true, permission } + : { authorized: false, permission }; + } catch (error) { + const status = githubStatus(error); + if (status === 404) { + await octokit.request( + 'GET /repos/{owner}/{repo}', + { owner: request.owner, repo: request.repo }, + ); + return { authorized: false, permission: null }; + } + throw error; + } +} diff --git a/packages/core/src/services/prSplit/command.ts b/packages/core/src/services/prSplit/command.ts new file mode 100644 index 000000000..62a88b2a2 --- /dev/null +++ b/packages/core/src/services/prSplit/command.ts @@ -0,0 +1,39 @@ +export interface ParsedSplitCommand { + /** Whitespace-normalized natural-language guidance after `/split`. */ + instruction: string; + validationError?: 'instruction_too_long'; +} + +export const MAX_SPLIT_INSTRUCTION_LENGTH = 8_000; +const SPLIT_COMMAND_PATTERN = /^\/split(?=$|[\t\r\n ])[\t\r\n ]*([\s\S]*)$/; + +/** + * Normalize guidance for persistence and idempotency comparisons. + * + * Case is preserved because it may be meaningful in identifiers or paths; + * leading/trailing whitespace is removed and internal whitespace is collapsed. + */ +export function normalizeSplitInstruction(instruction: string): string { + return instruction.trim().replace(/\s+/g, ' '); +} + +/** + * Parse a `/split` issue-comment command. + * + * The command must be the first bytes in the comment. This deliberately rejects + * leading prose/whitespace, mentions such as `/splitter`, and a later `/split` + * token so an ordinary discussion comment cannot accidentally start an + * operation. + */ +export function parseSplitCommand(body: string | null | undefined): ParsedSplitCommand | null { + if (!body) return null; + + const match = SPLIT_COMMAND_PATTERN.exec(body); + if (!match) return null; + + const instruction = normalizeSplitInstruction(match[1] ?? ''); + if (instruction.length > MAX_SPLIT_INSTRUCTION_LENGTH) { + return { instruction: '', validationError: 'instruction_too_long' }; + } + return { instruction }; +} diff --git a/packages/core/src/services/prSplit/commandStore.ts b/packages/core/src/services/prSplit/commandStore.ts new file mode 100644 index 000000000..17a9ff82c --- /dev/null +++ b/packages/core/src/services/prSplit/commandStore.ts @@ -0,0 +1,436 @@ +import { randomUUID } from 'node:crypto'; +import type { Knex } from 'knex'; +import { normalizeSplitInstruction } from './command.js'; +import { + buildSplitOperationDedupeKey, + buildSplitOperationEventKey, + normalizeGitHubId, + normalizePositiveInteger, + type SplitEventKeyInput, +} from './keys.js'; +import { + createPrSplitOperationDecision, + getPrSplitOperation, + isPrSplitUniqueConstraintError, + resolvePrSplitDb, + type CreatePrSplitOperationInput, + type PrSplitOperation, + type PrSplitOperationDecision, +} from './operationStore.js'; + +export const TERMINAL_PR_SPLIT_COMMAND_OUTCOMES = [ + 'disabled', + 'unauthorized', + 'closed', + 'invalid', + 'rate_limited', + 'queued', + 'duplicate', + 'active', +] as const; + +export type PrSplitCommandOutcome = (typeof TERMINAL_PR_SPLIT_COMMAND_OUTCOMES)[number]; +export type PrSplitResponseState = 'pending' | 'claimed' | 'posted' | 'suppressed'; + +export const DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS = 5 * 60 * 1000; +export const DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT = 5; +export const DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT_WINDOW_MS = 10 * 60 * 1000; + +export interface PrSplitResponseClaim { + token: string; + needsReconciliation: boolean; +} + +export interface PrSplitCommandRateLimitOptions { + now?: Date; + maxCommands?: number; + windowMs?: number; +} + +export interface PrSplitCommandReceipt { + event_key: string; + repository_id: number; + repository: string; + source_pr_number: number; + requester_id: number; + requester: string; + original_comment_id: number; + instruction: string; + outcome: PrSplitCommandOutcome | 'processing'; + duplicate_kind: 'event' | 'semantic' | null; + operation_id: string | null; + response_state: PrSplitResponseState; + response_claim_token: string | null; + response_claimed_at: string | null; + response_comment_id: number | null; + response_posted_at: string | null; + created_at: string; + updated_at: string; +} + +export interface PrSplitCommandInput { + repositoryId: number; + repository: string; + sourcePrNumber: number; + requesterId: number; + requester: string; + originalCommentId: number; + instruction: string; +} + +export interface RecordPrSplitCommandOutcomeInput extends PrSplitCommandInput { + outcome: Extract< + PrSplitCommandOutcome, + 'disabled' | 'unauthorized' | 'closed' | 'invalid' | 'rate_limited' + >; +} + +export interface PrSplitCommandRecord { + receipt: PrSplitCommandReceipt; + operation: PrSplitOperation | null; + replayed: boolean; +} + +let localCommandWriteTail = Promise.resolve(); + +async function withLocalCommandWriteLock(action: () => Promise): Promise { + const previous = localCommandWriteTail; + let release = (): void => undefined; + localCommandWriteTail = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + return await action(); + } finally { + release(); + } +} + +function timestamp(date: Date): string { + return date.toISOString(); +} + +function isSqliteBusyError(error: unknown): boolean { + return typeof error === 'object' + && error !== null + && 'code' in error + && typeof error.code === 'string' + && error.code.startsWith('SQLITE_BUSY'); +} + +async function withSqliteBusyRetry(action: () => Promise): Promise { + for (let attempt = 0; attempt < 5; attempt += 1) { + try { + return await action(); + } catch (error) { + if (!isSqliteBusyError(error) || attempt === 4) throw error; + await new Promise((resolve) => setTimeout(resolve, 10 * (attempt + 1))); + } + } + throw new Error('SQLite write retry limit exhausted'); +} + +async function withCommandWrite(action: () => Promise): Promise { + return withLocalCommandWriteLock(() => withSqliteBusyRetry(action)); +} + +function receiptInsert(input: PrSplitCommandInput, outcome: string, now: Date) { + const repositoryId = normalizeGitHubId(input.repositoryId, 'repositoryId'); + const requesterId = normalizeGitHubId(input.requesterId, 'requesterId'); + const sourcePrNumber = normalizePositiveInteger(input.sourcePrNumber, 'sourcePrNumber'); + const eventKey = buildSplitOperationEventKey({ + repositoryId, + originalCommentId: input.originalCommentId, + }); + const currentTimestamp = timestamp(now); + + return { + event_key: eventKey, + repository_id: repositoryId, + repository: input.repository.trim(), + source_pr_number: sourcePrNumber, + requester_id: requesterId, + requester: input.requester, + original_comment_id: normalizeGitHubId(input.originalCommentId, 'originalCommentId'), + instruction: normalizeSplitInstruction(input.instruction), + outcome, + duplicate_kind: null, + operation_id: null, + response_state: outcome === 'rate_limited' ? 'suppressed' : 'pending', + response_claim_token: null, + response_claimed_at: null, + response_comment_id: null, + response_posted_at: null, + created_at: currentTimestamp, + updated_at: currentTimestamp, + }; +} + +async function findReceipt( + client: Knex, + eventKey: string, +): Promise { + return client('pr_split_command_receipts') + .where({ event_key: eventKey }) + .first(); +} + +async function hydrateRecord( + client: Knex, + receipt: PrSplitCommandReceipt, + replayed: boolean, +): Promise { + const operation = receipt.operation_id + ? await getPrSplitOperation(receipt.operation_id, client) + : null; + return { receipt, operation, replayed }; +} + +async function findRecord( + client: Knex, + eventKey: string, + replayed: boolean, +): Promise { + const receipt = await findReceipt(client, eventKey); + return receipt ? hydrateRecord(client, receipt, replayed) : null; +} + +/** Return the immutable disposition already assigned to a source comment. */ +export async function getPrSplitCommandRecord( + input: SplitEventKeyInput, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + return findRecord(client, buildSplitOperationEventKey(input), true); +} + +function rateLimitOptions(options: PrSplitCommandRateLimitOptions): { + now: Date; + maxCommands: number; + windowMs: number; +} { + const now = options.now ?? new Date(); + const maxCommands = options.maxCommands ?? DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT; + const windowMs = options.windowMs ?? DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT_WINDOW_MS; + if (!Number.isSafeInteger(maxCommands) || maxCommands <= 0) { + throw new RangeError('PR split command rate limit must be a positive safe integer'); + } + if (!Number.isFinite(windowMs) || windowMs <= 0) { + throw new RangeError('PR split command rate limit window must be positive'); + } + return { now, maxCommands, windowMs }; +} + +/** + * Atomically reserve an event and its per-user rate-limit slot. Inserting the + * processing receipt is deliberately the transaction's first database action: + * SQLite writers are serialized before any process counts the active window. + */ +export async function reservePrSplitCommand( + input: PrSplitCommandInput, + dbClient?: Knex, + options: PrSplitCommandRateLimitOptions = {}, +): Promise { + const { now, maxCommands, windowMs } = rateLimitOptions(options); + const client = await resolvePrSplitDb(dbClient); + const pendingReceipt = receiptInsert(input, 'processing', now); + const existing = await findRecord(client, pendingReceipt.event_key, true); + if (existing) return existing; + + return withCommandWrite(async () => { + const raced = await findRecord(client, pendingReceipt.event_key, true); + if (raced) return raced; + try { + return await client.transaction(async (transaction) => { + await transaction('pr_split_command_receipts').insert(pendingReceipt); + const row = await transaction('pr_split_command_receipts') + .where({ + repository_id: pendingReceipt.repository_id, + requester_id: pendingReceipt.requester_id, + }) + .andWhere('created_at', '>=', timestamp(new Date(now.getTime() - windowMs))) + .count({ count: '*' }) + .first(); + if (Number(row?.count ?? 0) > maxCommands) { + await transaction('pr_split_command_receipts') + .where({ event_key: pendingReceipt.event_key, outcome: 'processing' }) + .update({ outcome: 'rate_limited', response_state: 'suppressed' }); + } + const receipt = await findReceipt(transaction, pendingReceipt.event_key); + if (!receipt) throw new Error('Reserved PR split command receipt could not be read back'); + return hydrateRecord(transaction, receipt, false); + }); + } catch (error) { + if (isPrSplitUniqueConstraintError(error)) { + const concurrent = await findRecord(client, pendingReceipt.event_key, true); + if (concurrent) return concurrent; + } + throw error; + } + }); +} + +/** Persist a non-executable command disposition before attempting its response. */ +export async function recordPrSplitCommandOutcome( + input: RecordPrSplitCommandOutcomeInput, + dbClient?: Knex, +): Promise { + const reservation = await reservePrSplitCommand(input, dbClient); + if (reservation.receipt.outcome !== 'processing') return reservation; + const client = await resolvePrSplitDb(dbClient); + const eventKey = reservation.receipt.event_key; + + return withCommandWrite(async () => { + const now = new Date(); + const updated = await client('pr_split_command_receipts') + .where({ event_key: eventKey, outcome: 'processing' }) + .update({ + outcome: input.outcome, + response_state: input.outcome === 'rate_limited' ? 'suppressed' : 'pending', + updated_at: timestamp(now), + }); + const receipt = await findReceipt(client, eventKey); + if (!receipt) throw new Error('Completed PR split command receipt could not be read back'); + return hydrateRecord(client, receipt, updated === 0); + }); +} + +function receiptOutcome(decision: PrSplitOperationDecision): { + outcome: Extract; + duplicateKind: 'event' | 'semantic' | null; +} { + if (decision.outcome === 'created') return { outcome: 'queued', duplicateKind: null }; + if (decision.outcome === 'active') return { outcome: 'active', duplicateKind: null }; + return { outcome: 'duplicate', duplicateKind: decision.duplicateKind }; +} + +/** + * Atomically assign a terminal command disposition and create/dedupe/lock its + * executable operation. A competing condition change can never overwrite the + * first disposition committed for the source comment. + */ +export async function createOrGetPrSplitOperation( + input: CreatePrSplitOperationInput, + dbClient?: Knex, +): Promise { + buildSplitOperationDedupeKey(input); + const reservation = await reservePrSplitCommand(input, dbClient); + if (reservation.receipt.outcome !== 'processing') return reservation; + const client = await resolvePrSplitDb(dbClient); + const eventKey = reservation.receipt.event_key; + + return withCommandWrite(() => client.transaction(async (transaction) => { + const now = new Date(); + const claimed = await transaction('pr_split_command_receipts') + .where({ event_key: eventKey, outcome: 'processing' }) + .update({ updated_at: timestamp(now) }); + if (claimed === 0) { + const completed = await findRecord(transaction, eventKey, true); + if (!completed) throw new Error('Reserved PR split command receipt disappeared'); + return completed; + } + + const decision = await createPrSplitOperationDecision(input, transaction, now); + const terminal = receiptOutcome(decision); + await transaction('pr_split_command_receipts') + .where({ event_key: eventKey, outcome: 'processing' }) + .update({ + outcome: terminal.outcome, + duplicate_kind: terminal.duplicateKind, + operation_id: decision.operation.id, + updated_at: timestamp(now), + }); + const receipt = await findReceipt(transaction, eventKey); + if (!receipt) throw new Error('Created PR split command receipt could not be read back'); + return { receipt, operation: decision.operation, replayed: false }; + })); +} + +/** Claim a response attempt, reclaiming abandoned attempts after their lease. */ +export async function claimPrSplitCommandResponse( + eventKey: string, + dbClient?: Knex, + now = new Date(), + leaseDurationMs = DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS, +): Promise { + if (!Number.isFinite(leaseDurationMs) || leaseDurationMs <= 0) { + throw new RangeError('PR split response claim lease must be positive'); + } + const client = await resolvePrSplitDb(dbClient); + const claimToken = randomUUID(); + const currentTimestamp = timestamp(now); + const claimedPending = await client('pr_split_command_receipts') + .where({ event_key: eventKey, response_state: 'pending' }) + .whereNot({ outcome: 'processing' }) + .update({ + response_state: 'claimed', + response_claim_token: claimToken, + response_claimed_at: currentTimestamp, + updated_at: currentTimestamp, + }); + if (claimedPending === 1) { + return { token: claimToken, needsReconciliation: false }; + } + + const staleBefore = timestamp(new Date(now.getTime() - leaseDurationMs)); + const reclaimed = await client('pr_split_command_receipts') + .where({ event_key: eventKey, response_state: 'claimed' }) + .andWhere((builder) => { + builder.whereNull('response_claimed_at').orWhere('response_claimed_at', '<=', staleBefore); + }) + .update({ + response_claim_token: claimToken, + response_claimed_at: currentTimestamp, + updated_at: currentTimestamp, + }); + return reclaimed === 1 ? { token: claimToken, needsReconciliation: true } : null; +} + +/** Release a claim only when no GitHub response could have been created. */ +export async function releasePrSplitCommandResponseClaim( + eventKey: string, + claimToken: string, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const currentTimestamp = timestamp(new Date()); + const released = await client('pr_split_command_receipts') + .where({ + event_key: eventKey, + response_state: 'claimed', + response_claim_token: claimToken, + }) + .update({ + response_state: 'pending', + response_claim_token: null, + response_claimed_at: null, + updated_at: currentTimestamp, + }); + return released === 1; +} + +/** Mark the claimed response as posted without allowing another claimant. */ +export async function markPrSplitCommandResponsePosted( + eventKey: string, + claimToken: string, + responseCommentId: number | null, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const currentTimestamp = timestamp(new Date()); + const updated = await client('pr_split_command_receipts') + .where({ + event_key: eventKey, + response_state: 'claimed', + response_claim_token: claimToken, + }) + .update({ + response_state: 'posted', + response_claim_token: null, + response_comment_id: responseCommentId, + response_posted_at: currentTimestamp, + updated_at: currentTimestamp, + }); + return updated === 1; +} diff --git a/packages/core/src/services/prSplit/index.ts b/packages/core/src/services/prSplit/index.ts new file mode 100644 index 000000000..5b3e77b6f --- /dev/null +++ b/packages/core/src/services/prSplit/index.ts @@ -0,0 +1,87 @@ +export { + MAX_SPLIT_INSTRUCTION_LENGTH, + parseSplitCommand, + normalizeSplitInstruction, +} from './command.js'; +export type { ParsedSplitCommand } from './command.js'; + +export { + SPLIT_AUTHORIZED_PERMISSIONS, + isSplitPermissionAuthorized, + authorizeSplitRequester, +} from './authorization.js'; +export type { + PrSplitGitHubResponse, + PrSplitRequestClient, + SplitAuthorizedPermission, + SplitAuthorizationRequest, + SplitAuthorizationResult, +} from './authorization.js'; + +export { + buildSplitOperationEventKey, + buildSplitOperationDedupeKey, + normalizePositiveInteger, + normalizeRef, + normalizeSha, +} from './keys.js'; +export type { + SplitEventKeyInput, + SplitDedupeKeyInput, +} from './keys.js'; + +export { + TERMINAL_PR_SPLIT_COMMAND_OUTCOMES, + DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT, + DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT_WINDOW_MS, + DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS, + claimPrSplitCommandResponse, + createOrGetPrSplitOperation, + getPrSplitCommandRecord, + markPrSplitCommandResponsePosted, + recordPrSplitCommandOutcome, + releasePrSplitCommandResponseClaim, + reservePrSplitCommand, +} from './commandStore.js'; +export type { + PrSplitCommandInput, + PrSplitCommandOutcome, + PrSplitCommandRateLimitOptions, + PrSplitCommandReceipt, + PrSplitCommandRecord, + PrSplitResponseClaim, + PrSplitResponseState, + RecordPrSplitCommandOutcomeInput, +} from './commandStore.js'; + +export { + ACTIVE_SPLIT_OPERATION_STATUSES, + CANCELLED_QUEUED_SPLIT_OPERATION_ERROR, + DEFAULT_SPLIT_OPERATION_LEASE_MS, + STALE_SPLIT_OPERATION_ERROR, + TERMINAL_SPLIT_OPERATION_STATUSES, + SPLIT_OPERATION_STATUSES, + assertPrSplitOperationLease, + cancelQueuedPrSplitOperation, + isActiveSplitOperationStatus, + isTerminalSplitOperationStatus, + getActivePrSplitOperation, + getPrSplitOperation, + heartbeatPrSplitOperation, + recoverStalePrSplitOperations, + updatePrSplitOperationStatus, +} from './operationStore.js'; +export type { + SplitOperationStatus, + PrSplitOperation, + CreatePrSplitOperationInput, + CancelQueuedPrSplitOperationOptions, + HeartbeatPrSplitOperationOptions, + UpdatePrSplitOperationStatusOptions, +} from './operationStore.js'; + +export { handlePrSplitComment, isPrSplitExecutionEnabled } from './intake.js'; +export type { + PrSplitIntakeDependencies, + PrSplitIntakeResult, +} from './intake.js'; diff --git a/packages/core/src/services/prSplit/intake.ts b/packages/core/src/services/prSplit/intake.ts new file mode 100644 index 000000000..5c7fd7cf7 --- /dev/null +++ b/packages/core/src/services/prSplit/intake.ts @@ -0,0 +1,432 @@ +import type { IssueCommentEvent } from '@octokit/webhooks-types'; +import type { Knex } from 'knex'; +import { getAuthenticatedOctokit } from '../../auth/githubAuth.js'; +import type { DeliveryDisposition } from '../../intake/routingWebSocketProtocol.js'; +import logger from '../../utils/logger.js'; +import { + authorizeSplitRequester, + type PrSplitRequestClient, + type SplitAuthorizationRequest, + type SplitAuthorizationResult, +} from './authorization.js'; +import { + claimPrSplitCommandResponse, + createOrGetPrSplitOperation, + getPrSplitCommandRecord, + markPrSplitCommandResponsePosted, + recordPrSplitCommandOutcome, + releasePrSplitCommandResponseClaim, + reservePrSplitCommand, + type PrSplitCommandInput, + type PrSplitCommandOutcome, + type PrSplitCommandRecord, +} from './commandStore.js'; +import { MAX_SPLIT_INSTRUCTION_LENGTH, parseSplitCommand } from './command.js'; +import type { PrSplitOperation } from './operationStore.js'; + +export interface PrSplitIntakeDependencies { + getOctokit: () => Promise; + authorizeRequester: ( + client: PrSplitRequestClient, + request: SplitAuthorizationRequest, + ) => Promise; + isExecutionEnabled: () => boolean; + getResponseAuthorLogin: () => string | undefined; + db?: Knex; +} + +export type PrSplitIntakeResult = + | { handled: false } + | { + handled: true; + disposition: DeliveryDisposition; + outcome: PrSplitCommandOutcome; + operation?: PrSplitOperation; + }; + +export function isPrSplitExecutionEnabled( + value = process.env.PR_SPLIT_EXECUTION_ENABLED, +): boolean { + return value === '1' || value?.toLowerCase() === 'true'; +} + +async function getDefaultOctokit(): Promise { + return getAuthenticatedOctokit(); +} + +const DEFAULT_DEPENDENCIES: PrSplitIntakeDependencies = { + getOctokit: getDefaultOctokit, + authorizeRequester: authorizeSplitRequester, + isExecutionEnabled: isPrSplitExecutionEnabled, + getResponseAuthorLogin: () => process.env.GITHUB_BOT_USERNAME?.trim() || undefined, +}; + +const BLOCKED_OUTCOME_REASONS: Partial> = { + disabled: 'split_execution_not_enabled', + unauthorized: 'insufficient_repository_permission', + active: 'split_operation_already_active', + closed: 'split_pull_request_closed', + invalid: 'split_instruction_too_long', + rate_limited: 'split_request_rate_limited', +}; + +function acceptedDisposition(commentId: number): DeliveryDisposition { + return { + status: 'accepted', + billing: { seatConsumed: false }, + evidence: { triggerCommentIds: [commentId] }, + }; +} + +function blockedDisposition(reason: string): DeliveryDisposition { + return { + status: 'blocked', + reason, + billing: { seatConsumed: false }, + }; +} + +function dispositionFor(record: PrSplitCommandRecord): DeliveryDisposition { + const reason = BLOCKED_OUTCOME_REASONS[record.receipt.outcome as PrSplitCommandOutcome]; + return reason + ? blockedDisposition(reason) + : acceptedDisposition(record.receipt.original_comment_id); +} + +type UnknownRecord = Record; + +function isRecord(value: unknown): value is UnknownRecord { + return typeof value === 'object' && value !== null; +} + +function shortOperationId(record: PrSplitCommandRecord): string { + return record.receipt.operation_id?.slice(0, 8) ?? 'unknown'; +} + +function responseBody(record: PrSplitCommandRecord): string { + const shortId = shortOperationId(record); + + switch (record.receipt.outcome) { + case 'disabled': + return '⏸️ `/split` is not available because split execution workers are not enabled for this deployment.'; + case 'unauthorized': + return '⛔ `/split` requires `write`, `maintain`, or `admin` permission on this repository.'; + case 'closed': + return '⛔ `/split` can only run on an open, unmerged pull request.'; + case 'invalid': + return `⛔ \`/split\` instructions are limited to ${MAX_SPLIT_INSTRUCTION_LENGTH.toLocaleString('en-US')} characters.`; + case 'rate_limited': + return '⏳ Too many `/split` commands were received from this account. Try again later.'; + case 'queued': + return `✅ Split operation \`${shortId}\` queued. The source PR branch will not be modified.`; + case 'active': + return `⏳ Split operation \`${shortId}\` already owned this PR when the command was recorded. Wait for it to finish before requesting another split.`; + case 'duplicate': + return `ℹ️ Equivalent split operation \`${shortId}\` was already recorded for this PR.`; + default: + throw new Error(`Unsupported PR split command outcome: ${record.receipt.outcome}`); + } +} + +interface PostResponseContext { + owner: string; + repo: string; + responseAuthorLogin?: string; + getOctokit: () => Promise; + octokit?: PrSplitRequestClient; + db?: Knex; +} + +const RESPONSE_RECONCILIATION_PAGE_SIZE = 100; +const RESPONSE_RECONCILIATION_SAFETY_MARGIN_MS = 60_000; + +function responseMarker(eventKey: string): string { + return ``; +} + +function githubErrorStatus(error: unknown): number | null { + return isRecord(error) && typeof error.status === 'number' ? error.status : null; +} + +function isDefinitiveGitHubRejection(error: unknown): boolean { + const status = githubErrorStatus(error); + return status !== null && status >= 400 && status < 500; +} + +async function findPostedResponse( + record: PrSplitCommandRecord, + context: PostResponseContext, + octokit: PrSplitRequestClient, +): Promise { + const marker = responseMarker(record.receipt.event_key); + let responseAuthorLogin = context.responseAuthorLogin; + if (!responseAuthorLogin) { + const { data } = await octokit.request('GET /installation', {}); + if (!isRecord(data) || typeof data.app_slug !== 'string') { + throw new Error('GitHub installation response did not include an App slug'); + } + responseAuthorLogin = `${data.app_slug}[bot]`; + } + const since = new Date( + new Date(record.receipt.created_at).getTime() - RESPONSE_RECONCILIATION_SAFETY_MARGIN_MS, + ).toISOString(); + for (let page = 1; ; page += 1) { + const { data } = await octokit.request( + 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + { + owner: context.owner, + repo: context.repo, + issue_number: record.receipt.source_pr_number, + since, + per_page: RESPONSE_RECONCILIATION_PAGE_SIZE, + page, + }, + ); + if (!Array.isArray(data)) { + throw new Error('GitHub issue comments response was not an array'); + } + const response = data.find((comment) => { + if (!isRecord(comment) || typeof comment.body !== 'string') return false; + const user = comment.user; + return comment.body.includes(marker) + && isRecord(user) + && user.type === 'Bot' + && typeof user.login === 'string' + && user.login.toLowerCase() === responseAuthorLogin.toLowerCase(); + }); + if (isRecord(response)) return typeof response.id === 'number' ? response.id : null; + if (data.length < RESPONSE_RECONCILIATION_PAGE_SIZE) return undefined; + } +} + +async function markResponsePosted( + record: PrSplitCommandRecord, + claimToken: string, + responseCommentId: number | null, + db?: Knex, +): Promise { + const marked = await markPrSplitCommandResponsePosted( + record.receipt.event_key, + claimToken, + responseCommentId, + db, + ); + if (!marked) throw new Error(`Lost PR split response claim for ${record.receipt.event_key}`); +} + +async function postResponseOnce( + record: PrSplitCommandRecord, + context: PostResponseContext, +): Promise { + const eventKey = record.receipt.event_key; + if (record.receipt.response_state === 'posted' + || record.receipt.response_state === 'suppressed') return false; + const claim = await claimPrSplitCommandResponse(eventKey, context.db); + if (!claim) { + const refreshed = await getPrSplitCommandRecord({ + repositoryId: record.receipt.repository_id, + originalCommentId: record.receipt.original_comment_id, + }, context.db); + if (refreshed?.receipt.response_state === 'posted' + || refreshed?.receipt.response_state === 'suppressed') return false; + throw new Error(`PR split response claim for ${eventKey} is still live; retry delivery`); + } + + let octokit = context.octokit; + try { + octokit ??= await context.getOctokit(); + } catch (error) { + if (!claim.needsReconciliation) { + await releasePrSplitCommandResponseClaim(eventKey, claim.token, context.db); + } + throw error; + } + + if (claim.needsReconciliation) { + const responseCommentId = await findPostedResponse(record, context, octokit); + if (responseCommentId !== undefined) { + await markResponsePosted(record, claim.token, responseCommentId, context.db); + return true; + } + } + + let data: unknown; + try { + ({ data } = await octokit.request( + 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + { + owner: context.owner, + repo: context.repo, + issue_number: record.receipt.source_pr_number, + body: `${responseBody(record)}\n\n${responseMarker(eventKey)}`, + }, + )); + } catch (error) { + if (isDefinitiveGitHubRejection(error)) { + await releasePrSplitCommandResponseClaim(eventKey, claim.token, context.db); + } + throw error; + } + const responseCommentId = isRecord(data) && typeof data.id === 'number' ? data.id : null; + await markResponsePosted(record, claim.token, responseCommentId, context.db); + return true; +} + +interface PullRequestSnapshot { + baseRef: string; + baseSha: string; + headSha: string; + state: 'open' | 'closed'; + merged: boolean; +} + +function parsePullRequestSnapshot(data: unknown): PullRequestSnapshot { + if (!isRecord(data) || !isRecord(data.base) || !isRecord(data.head)) { + throw new Error('GitHub pull request response did not include base/head metadata'); + } + const { ref, sha: baseSha } = data.base; + const headSha = data.head.sha; + const state = data.state; + const merged = data.merged; + if ( + typeof ref !== 'string' + || typeof baseSha !== 'string' + || typeof headSha !== 'string' + || (state !== 'open' && state !== 'closed') + || typeof merged !== 'boolean' + ) { + throw new Error('GitHub pull request response contained invalid snapshot metadata'); + } + return { baseRef: ref, baseSha, headSha, state, merged }; +} + +interface IntakeContext { + owner: string; + repo: string; + baseInput: PrSplitCommandInput; + dependencies: PrSplitIntakeDependencies; + correlationId: string; +} + +async function finishIntake( + record: PrSplitCommandRecord, + context: IntakeContext, + existingOctokit?: PrSplitRequestClient, +): Promise { + if (record.receipt.outcome === 'processing') { + throw new Error(`PR split command ${record.receipt.event_key} is still processing`); + } + const responsePosted = await postResponseOnce(record, { + owner: context.owner, + repo: context.repo, + responseAuthorLogin: context.dependencies.getResponseAuthorLogin()?.trim() || undefined, + getOctokit: context.dependencies.getOctokit, + ...(existingOctokit ? { octokit: existingOctokit } : {}), + db: context.dependencies.db, + }); + logger.withCorrelation(context.correlationId).info({ + repositoryId: record.receipt.repository_id, + repository: record.receipt.repository, + sourcePrNumber: record.receipt.source_pr_number, + requesterId: record.receipt.requester_id, + requester: record.receipt.requester, + commentId: record.receipt.original_comment_id, + operationId: record.receipt.operation_id, + outcome: record.receipt.outcome, + replayed: record.replayed, + responsePosted, + }, 'Handled /split command'); + + return { + handled: true, + disposition: dispositionFor(record), + outcome: record.receipt.outcome, + ...(record.operation ? { operation: record.operation } : {}), + }; +} + +/** + * Durable intake boundary for `issue_comment.created` `/split` commands. + * Every recognized command receives one immutable disposition before response. + */ +export async function handlePrSplitComment( + payload: IssueCommentEvent, + correlationId: string, + dependencyOverrides: Partial = {}, +): Promise { + if (payload.action !== 'created' || !payload.issue.pull_request) return { handled: false }; + + const command = parseSplitCommand(payload.comment.body); + if (!command) return { handled: false }; + if (!payload.comment.user) throw new Error('GitHub split comment did not include a user'); + + const dependencies = { ...DEFAULT_DEPENDENCIES, ...dependencyOverrides }; + const owner = payload.repository.owner.login; + const repo = payload.repository.name; + const baseInput: PrSplitCommandInput = { + repositoryId: payload.repository.id, + repository: payload.repository.full_name, + sourcePrNumber: payload.issue.number, + requesterId: payload.comment.user.id, + requester: payload.comment.user.login, + originalCommentId: payload.comment.id, + instruction: command.instruction, + }; + const context: IntakeContext = { owner, repo, baseInput, dependencies, correlationId }; + const reservation = await reservePrSplitCommand(baseInput, dependencies.db); + if (reservation.receipt.outcome !== 'processing') { + return finishIntake(reservation, context); + } + + if (command.validationError) { + const record = await recordPrSplitCommandOutcome( + { ...baseInput, outcome: 'invalid' }, + dependencies.db, + ); + return finishIntake(record, context); + } + + if (!dependencies.isExecutionEnabled()) { + const record = await recordPrSplitCommandOutcome( + { ...baseInput, outcome: 'disabled' }, + dependencies.db, + ); + return finishIntake(record, context); + } + + const octokit = await dependencies.getOctokit(); + const authorization = await dependencies.authorizeRequester(octokit, { + owner, + repo, + username: baseInput.requester, + requesterId: baseInput.requesterId, + }); + if (!authorization.authorized) { + const record = await recordPrSplitCommandOutcome( + { ...baseInput, outcome: 'unauthorized' }, + dependencies.db, + ); + return finishIntake(record, context, octokit); + } + + const { data } = await octokit.request( + 'GET /repos/{owner}/{repo}/pulls/{pull_number}', + { owner, repo, pull_number: baseInput.sourcePrNumber }, + ); + const pullRequest = parsePullRequestSnapshot(data); + if (pullRequest.state !== 'open' || pullRequest.merged) { + const record = await recordPrSplitCommandOutcome( + { ...baseInput, outcome: 'closed' }, + dependencies.db, + ); + return finishIntake(record, context, octokit); + } + + const record = await createOrGetPrSplitOperation({ + ...baseInput, + baseRef: pullRequest.baseRef, + baseSha: pullRequest.baseSha, + headSha: pullRequest.headSha, + }, dependencies.db); + return finishIntake(record, context, octokit); +} diff --git a/packages/core/src/services/prSplit/keys.ts b/packages/core/src/services/prSplit/keys.ts new file mode 100644 index 000000000..94a91543d --- /dev/null +++ b/packages/core/src/services/prSplit/keys.ts @@ -0,0 +1,65 @@ +import { createHash } from 'node:crypto'; +import { normalizeSplitInstruction } from './command.js'; + +export interface SplitEventKeyInput { + repositoryId: number; + originalCommentId: number; +} + +export interface SplitDedupeKeyInput { + repositoryId: number; + sourcePrNumber: number; + baseRef: string; + baseSha: string; + headSha: string; + instruction: string; +} + +export function normalizePositiveInteger(value: number, field: string): number { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new RangeError(`${field} must be a positive safe integer`); + } + return value; +} + +export function normalizeGitHubId(value: number, field: string): number { + return normalizePositiveInteger(value, field); +} + +function normalizeNonEmptyString(value: string, field: string): string { + const normalized = value.trim(); + if (!normalized) throw new RangeError(`${field} must not be empty`); + return normalized; +} + +export function normalizeSha(sha: string): string { + return normalizeNonEmptyString(sha, 'sha').toLowerCase(); +} + +export function normalizeRef(ref: string): string { + return normalizeNonEmptyString(ref, 'ref'); +} + +function hashCanonicalInput(parts: readonly (string | number)[]): string { + return createHash('sha256').update(JSON.stringify(parts)).digest('hex'); +} + +/** Stable identity for one GitHub issue comment across webhook retries and repository renames. */ +export function buildSplitOperationEventKey(input: SplitEventKeyInput): string { + return hashCanonicalInput([ + normalizeGitHubId(input.repositoryId, 'repositoryId'), + normalizeGitHubId(input.originalCommentId, 'originalCommentId'), + ]); +} + +/** Stable semantic key for equivalent split inputs, independent of event identity. */ +export function buildSplitOperationDedupeKey(input: SplitDedupeKeyInput): string { + return hashCanonicalInput([ + normalizeGitHubId(input.repositoryId, 'repositoryId'), + normalizePositiveInteger(input.sourcePrNumber, 'sourcePrNumber'), + normalizeRef(input.baseRef), + normalizeSha(input.baseSha), + normalizeSha(input.headSha), + normalizeSplitInstruction(input.instruction), + ]); +} diff --git a/packages/core/src/services/prSplit/operationStore.ts b/packages/core/src/services/prSplit/operationStore.ts new file mode 100644 index 000000000..37d5f8f4f --- /dev/null +++ b/packages/core/src/services/prSplit/operationStore.ts @@ -0,0 +1,392 @@ +import { randomUUID } from 'node:crypto'; +import type { Knex } from 'knex'; +import { normalizeSplitInstruction } from './command.js'; +import { + buildSplitOperationDedupeKey, + buildSplitOperationEventKey, + normalizeGitHubId, + normalizePositiveInteger, + normalizeRef, + normalizeSha, +} from './keys.js'; + +export const ACTIVE_SPLIT_OPERATION_STATUSES = ['queued', 'running'] as const; +export const TERMINAL_SPLIT_OPERATION_STATUSES = ['completed', 'failed'] as const; +export const SPLIT_OPERATION_STATUSES = [ + ...ACTIVE_SPLIT_OPERATION_STATUSES, + ...TERMINAL_SPLIT_OPERATION_STATUSES, +] as const; +export const DEFAULT_SPLIT_OPERATION_LEASE_MS = 15 * 60 * 1000; +export const STALE_SPLIT_OPERATION_ERROR = 'Split operation lease expired before completion'; +export const CANCELLED_QUEUED_SPLIT_OPERATION_ERROR = 'Queued split operation cancelled'; + +export type SplitOperationStatus = (typeof SPLIT_OPERATION_STATUSES)[number]; + +export interface PrSplitOperation { + id: string; + repository_id: number; + repository: string; + source_pr_number: number; + base_ref: string; + base_sha: string; + head_sha: string; + requester_id: number; + requester: string; + original_comment_id: number; + instruction: string; + event_key: string; + dedupe_key: string; + status: SplitOperationStatus; + error_message: string | null; + started_at: string | null; + heartbeat_at: string | null; + lease_expires_at: string | null; + lease_token: string | null; + finished_at: string | null; + created_at: string; + updated_at: string; +} + +export interface CreatePrSplitOperationInput { + repositoryId: number; + repository: string; + sourcePrNumber: number; + baseRef: string; + baseSha: string; + headSha: string; + requesterId: number; + requester: string; + originalCommentId: number; + instruction: string; +} + +export type PrSplitOperationDecision = + | { outcome: 'created'; operation: PrSplitOperation } + | { + outcome: 'duplicate'; + duplicateKind: 'event' | 'semantic'; + operation: PrSplitOperation; + } + | { outcome: 'active'; operation: PrSplitOperation }; + +export interface UpdatePrSplitOperationStatusOptions { + errorMessage?: string | null; + leaseDurationMs?: number; + leaseToken?: string; + now?: Date; +} + +export interface HeartbeatPrSplitOperationOptions { + leaseToken: string; + leaseDurationMs?: number; + now?: Date; +} + +export interface CancelQueuedPrSplitOperationOptions { + reason?: string; + now?: Date; +} + +export function isActiveSplitOperationStatus(status: SplitOperationStatus): boolean { + return ACTIVE_SPLIT_OPERATION_STATUSES.includes( + status as (typeof ACTIVE_SPLIT_OPERATION_STATUSES)[number], + ); +} + +export function isTerminalSplitOperationStatus(status: SplitOperationStatus): boolean { + return TERMINAL_SPLIT_OPERATION_STATUSES.includes( + status as (typeof TERMINAL_SPLIT_OPERATION_STATUSES)[number], + ); +} + +export async function resolvePrSplitDb(client?: Knex): Promise { + if (client) return client; + return (await import('../../db/connection.js')).db; +} + +export function isPrSplitUniqueConstraintError(error: unknown): boolean { + if (typeof error !== 'object' || error === null || !('code' in error)) return false; + const code = typeof error.code === 'string' ? error.code : ''; + return code === 'SQLITE_CONSTRAINT_UNIQUE' || code === 'SQLITE_CONSTRAINT_PRIMARYKEY'; +} + +function timestamp(date: Date): string { + return date.toISOString(); +} + +function leaseExpiry(now: Date, leaseDurationMs = DEFAULT_SPLIT_OPERATION_LEASE_MS): string { + if (!Number.isFinite(leaseDurationMs) || leaseDurationMs <= 0) { + throw new RangeError('Split operation lease duration must be a positive number'); + } + return timestamp(new Date(now.getTime() + leaseDurationMs)); +} + +async function findByEventKey(client: Knex, eventKey: string): Promise { + return client('pr_split_operations') + .where({ event_key: eventKey }) + .first(); +} + +async function findSemanticDuplicate( + client: Knex, + dedupeKey: string, +): Promise { + return client('pr_split_operations') + .where({ dedupe_key: dedupeKey }) + .whereNot({ status: 'failed' }) + .orderBy('created_at', 'desc') + .first(); +} + +/** Find the queued/running operation that currently owns a source PR lock. */ +export async function getActivePrSplitOperation( + repositoryId: number, + sourcePrNumber: number, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const operation = await client('pr_split_operations') + .where({ + repository_id: normalizeGitHubId(repositoryId, 'repositoryId'), + source_pr_number: normalizePositiveInteger(sourcePrNumber, 'sourcePrNumber'), + }) + .whereIn('status', ACTIVE_SPLIT_OPERATION_STATUSES) + .first(); + + return operation ?? null; +} + +/** Fail running operations whose worker lease has elapsed and release their PR lock. */ +export async function recoverStalePrSplitOperations( + repositoryId: number, + sourcePrNumber: number, + dbClient?: Knex, + now = new Date(), +): Promise { + const client = await resolvePrSplitDb(dbClient); + const currentTimestamp = timestamp(now); + + return client('pr_split_operations') + .where({ + repository_id: normalizeGitHubId(repositoryId, 'repositoryId'), + source_pr_number: normalizePositiveInteger(sourcePrNumber, 'sourcePrNumber'), + status: 'running', + }) + .andWhere((builder) => { + builder.whereNull('lease_expires_at').orWhere('lease_expires_at', '<=', currentTimestamp); + }) + .update({ + status: 'failed', + error_message: STALE_SPLIT_OPERATION_ERROR, + lease_expires_at: null, + lease_token: null, + finished_at: currentTimestamp, + updated_at: currentTimestamp, + }); +} + +/** Administratively fail abandoned queued work and release its per-PR lock. */ +export async function cancelQueuedPrSplitOperation( + operationId: string, + options: CancelQueuedPrSplitOperationOptions = {}, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const currentTimestamp = timestamp(options.now ?? new Date()); + const reason = options.reason?.trim() || CANCELLED_QUEUED_SPLIT_OPERATION_ERROR; + const updated = await client('pr_split_operations') + .where({ id: operationId, status: 'queued' }) + .update({ + status: 'failed', + error_message: reason, + finished_at: currentTimestamp, + updated_at: currentTimestamp, + }); + return updated === 1 ? getPrSplitOperation(operationId, client) : null; +} + +/** Resolve the executable operation decision inside the caller's transaction. */ +export async function createPrSplitOperationDecision( + input: CreatePrSplitOperationInput, + dbClient: Knex, + now = new Date(), +): Promise { + const normalizedInstruction = normalizeSplitInstruction(input.instruction); + const repositoryId = normalizeGitHubId(input.repositoryId, 'repositoryId'); + const sourcePrNumber = normalizePositiveInteger(input.sourcePrNumber, 'sourcePrNumber'); + const requesterId = normalizeGitHubId(input.requesterId, 'requesterId'); + const originalCommentId = normalizeGitHubId(input.originalCommentId, 'originalCommentId'); + const eventKey = buildSplitOperationEventKey({ + repositoryId, + originalCommentId, + }); + const dedupeKey = buildSplitOperationDedupeKey({ + repositoryId, + sourcePrNumber, + baseRef: input.baseRef, + baseSha: input.baseSha, + headSha: input.headSha, + instruction: normalizedInstruction, + }); + + await recoverStalePrSplitOperations(repositoryId, sourcePrNumber, dbClient, now); + + const currentTimestamp = timestamp(now); + const record = { + id: randomUUID(), + repository_id: repositoryId, + repository: input.repository.trim(), + source_pr_number: sourcePrNumber, + base_ref: normalizeRef(input.baseRef), + base_sha: normalizeSha(input.baseSha), + head_sha: normalizeSha(input.headSha), + requester_id: requesterId, + requester: input.requester, + original_comment_id: originalCommentId, + instruction: normalizedInstruction, + event_key: eventKey, + dedupe_key: dedupeKey, + status: 'queued' as const, + error_message: null, + started_at: null, + heartbeat_at: null, + lease_expires_at: null, + lease_token: null, + finished_at: null, + created_at: currentTimestamp, + updated_at: currentTimestamp, + }; + + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + await dbClient('pr_split_operations').insert(record); + const operation = await findByEventKey(dbClient, eventKey); + if (!operation) throw new Error('Created PR split operation could not be read back'); + return { outcome: 'created', operation }; + } catch (error) { + if (!isPrSplitUniqueConstraintError(error)) throw error; + + const eventDuplicate = await findByEventKey(dbClient, eventKey); + if (eventDuplicate) { + return { outcome: 'duplicate', duplicateKind: 'event', operation: eventDuplicate }; + } + + const semanticDuplicate = await findSemanticDuplicate(dbClient, dedupeKey); + if (semanticDuplicate) { + return { outcome: 'duplicate', duplicateKind: 'semantic', operation: semanticDuplicate }; + } + + const active = await getActivePrSplitOperation( + repositoryId, + sourcePrNumber, + dbClient, + ); + if (active) return { outcome: 'active', operation: active }; + if (attempt === 1) throw error; + } + } + + throw new Error('Unable to create PR split operation'); +} + +export async function getPrSplitOperation( + operationId: string, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const operation = await client('pr_split_operations') + .where({ id: operationId }) + .first(); + return operation ?? null; +} + +/** + * Apply a fenced lifecycle transition. A worker lease starts when queued work + * is claimed; running workers must present their token and own a live lease. + */ +export async function updatePrSplitOperationStatus( + operationId: string, + status: SplitOperationStatus, + options: UpdatePrSplitOperationStatusOptions = {}, + dbClient?: Knex, +): Promise { + if (status === 'queued') return null; + + const client = await resolvePrSplitDb(dbClient); + const now = options.now ?? new Date(); + const currentTimestamp = timestamp(now); + const updates: Record = { status, updated_at: currentTimestamp }; + const query = client('pr_split_operations').where({ id: operationId }); + let requiresLiveLease = false; + + if (status === 'running') { + query.andWhere({ status: 'queued' }); + updates.started_at = currentTimestamp; + updates.heartbeat_at = currentTimestamp; + updates.lease_expires_at = leaseExpiry(now, options.leaseDurationMs); + updates.lease_token = randomUUID(); + updates.finished_at = null; + updates.error_message = null; + } else if (options.leaseToken) { + query.andWhere({ status: 'running', lease_token: options.leaseToken }); + requiresLiveLease = true; + updates.finished_at = currentTimestamp; + updates.lease_expires_at = null; + updates.lease_token = null; + updates.error_message = status === 'completed' + ? null + : options.errorMessage?.trim() || 'Split operation failed'; + } else if (status === 'failed') { + query.andWhere({ status: 'queued' }); + updates.finished_at = currentTimestamp; + updates.lease_expires_at = null; + updates.lease_token = null; + updates.error_message = options.errorMessage?.trim() || 'Split operation failed'; + } else { + return null; + } + + if (requiresLiveLease) query.andWhere('lease_expires_at', '>', currentTimestamp); + const updated = await query.update(updates); + if (updated === 0) return null; + return getPrSplitOperation(operationId, client); +} + +/** Extend a running operation's lease while proving ownership with its claim token. */ +export async function heartbeatPrSplitOperation( + operationId: string, + options: HeartbeatPrSplitOperationOptions, + dbClient?: Knex, +): Promise { + const client = await resolvePrSplitDb(dbClient); + const now = options.now ?? new Date(); + const currentTimestamp = timestamp(now); + const updated = await client('pr_split_operations') + .where({ id: operationId, status: 'running', lease_token: options.leaseToken }) + .andWhere('lease_expires_at', '>', currentTimestamp) + .update({ + heartbeat_at: currentTimestamp, + lease_expires_at: leaseExpiry(now, options.leaseDurationMs), + updated_at: currentTimestamp, + }); + if (updated === 0) return null; + return getPrSplitOperation(operationId, client); +} + +/** + * Fence external GitHub side effects. A future worker must call this immediately + * before each side effect and stop if the token no longer owns a live lease. + */ +export async function assertPrSplitOperationLease( + operationId: string, + leaseToken: string, + dbClient?: Knex, + now = new Date(), +): Promise { + const client = await resolvePrSplitDb(dbClient); + const operation = await client('pr_split_operations') + .where({ id: operationId, status: 'running', lease_token: leaseToken }) + .andWhere('lease_expires_at', '>', timestamp(now)) + .first(); + return operation ?? null; +} diff --git a/packages/core/src/webhook/commentEventHandler.ts b/packages/core/src/webhook/commentEventHandler.ts index 7184d9c1b..23e8fcd45 100644 --- a/packages/core/src/webhook/commentEventHandler.ts +++ b/packages/core/src/webhook/commentEventHandler.ts @@ -20,6 +20,7 @@ import { MODEL_INFO_MAP } from '../config/modelDefinitions.js'; import { getBotUsername } from '../daemon/configLoader.js'; import { AgentRegistry } from '../agents/AgentRegistry.js'; import type { DeliveryDisposition } from '../intake/routingWebSocketProtocol.js'; +import { parseSplitCommand } from '../services/prSplit/command.js'; export interface UltrafixDeps { loadUltrafixRatingGoal: () => Promise; @@ -112,6 +113,13 @@ function getCommentEventDetails( return null; } + // Webhook intake handles `/split` before this generic processor. Keep + // polling/synthetic callers from turning it into an implementation job. + if (parseSplitCommand(issuePayload.comment.body)) { + correlatedLogger.debug({ repository: repoFullName, commentId: issuePayload.comment.id }, 'Skipping /split outside webhook intake'); + return null; + } + return { prNumber: issuePayload.issue.number, comment: issuePayload.comment, @@ -130,6 +138,18 @@ function getCommentEventDetails( return null; } +function ignoredCommentReason( + payload: IssueCommentEvent | PullRequestReviewCommentEvent, + eventType: CommentEventType, +): string { + if (eventType !== 'issue_comment') return 'not_pull_request_comment'; + const issuePayload = payload as IssueCommentEvent; + if (!issuePayload.issue.pull_request) return 'not_pull_request_comment'; + return parseSplitCommand(issuePayload.comment.body) + ? 'split_requires_webhook_intake' + : 'not_pull_request_comment'; +} + export async function handleCommentDeleted(payload: IssueCommentEvent | PullRequestReviewCommentEvent, eventType: CommentEventType, correlationId: string, config: CommentEventConfig): Promise { const { redisClient } = config; const correlatedLogger = logger.withCorrelation(correlationId); @@ -527,7 +547,7 @@ export async function processCommentEvent(payload: IssueCommentEvent | PullReque const repoFullName = `${owner}/${repo}`; const eventDetails = getCommentEventDetails(payload, eventType, repoFullName, correlatedLogger); - if (!eventDetails) return { status: 'ignored', reason: 'not_pull_request_comment' }; + if (!eventDetails) return { status: 'ignored', reason: ignoredCommentReason(payload, eventType) }; const { prNumber, comment } = eventDetails; diff --git a/packages/core/src/webhook/webhookHandler.ts b/packages/core/src/webhook/webhookHandler.ts index 1df1d404c..909b9d59c 100644 --- a/packages/core/src/webhook/webhookHandler.ts +++ b/packages/core/src/webhook/webhookHandler.ts @@ -27,6 +27,10 @@ import type { } from '@octokit/webhooks-types'; import type { Redis } from 'ioredis'; import { ACCEPTED_NO_SEAT_DISPOSITION, normalizeDisposition, type DeliveryDisposition } from '../intake/routingWebSocketProtocol.js'; +import { + handlePrSplitComment, + type PrSplitIntakeResult, +} from '../services/prSplit/intake.js'; /** Runtime-accessible list of supported webhook event types — single source of truth. */ export const SUPPORTED_WEBHOOK_EVENTS = [ @@ -65,6 +69,10 @@ export type CommentDeletedHandler = (payload: IssueCommentEvent | PullRequestRev export type CommentEditedHandler = (payload: IssueCommentEvent | PullRequestReviewCommentEvent, eventType: CommentEventType, correlationId: string) => Promise; export type PullRequestProcessor = (payload: PullRequestEvent, correlationId: string) => Promise; export type CheckRunProcessor = (payload: CheckRunEvent, correlationId: string) => Promise; +export type SplitCommentHandler = ( + payload: IssueCommentEvent, + correlationId: string, +) => Promise; let processDetectedIssue: IssueProcessor | null = null; let processCommentEvent: CommentProcessor | null = null; @@ -73,6 +81,7 @@ let handleCommentEdited: CommentEditedHandler | null = null; let processPullRequest: PullRequestProcessor | null = null; let processCheckRun: CheckRunProcessor | null = null; let webhookRedisClient: Redis | null = null; +let processSplitComment: SplitCommentHandler = handlePrSplitComment; export interface WebhookHandlerOptions { issueProcessor: IssueProcessor; @@ -81,6 +90,7 @@ export interface WebhookHandlerOptions { commentEditedHandler: CommentEditedHandler; pullRequestProcessor?: PullRequestProcessor; checkRunProcessor?: CheckRunProcessor; + splitCommentHandler?: SplitCommentHandler; redisClient?: Redis; } @@ -91,6 +101,7 @@ export async function initializeWebhookHandler(options: WebhookHandlerOptions): handleCommentEdited = options.commentEditedHandler; processPullRequest = options.pullRequestProcessor || null; processCheckRun = options.checkRunProcessor || null; + processSplitComment = options.splitCommentHandler ?? handlePrSplitComment; webhookRedisClient = options.redisClient || null; logger.info('Webhook handler initialized'); } @@ -334,6 +345,14 @@ export async function processWebhookEvent( ): Promise { const correlatedLogger = logger.withCorrelation(correlationId); + // `/split` owns a separate authorization and durable-operation boundary. + // Intercept it before plan tracking or generic PR follow-up processing so a + // command can never be mistaken for an implementation request. + if (eventType === 'issue_comment' && isIssueCommentEvent(payload)) { + const splitResult = await processSplitComment(payload, correlationId); + if (splitResult.handled) return splitResult.disposition; + } + await handleUltrafixLabelRemoval(payload, eventType, correlationId); // Plan Issue Tracking (runs before standard processing to update status) diff --git a/src/polling/prCommentPolling.ts b/src/polling/prCommentPolling.ts index 0faa5ab92..1e68e54e3 100644 --- a/src/polling/prCommentPolling.ts +++ b/src/polling/prCommentPolling.ts @@ -5,6 +5,7 @@ import { getIssueQueue, COMMENT_BATCH_DELAY_MS, type CommentJobData, type Unproc import { filterCommentByAuthor, checkCommentTrigger } from '@propr/core'; import { extractLlmFromLabels, resolveModelAlias } from '@propr/core'; import { loadPrimaryProcessingLabels } from '@propr/core'; +import { parseSplitCommand } from '@propr/core'; import type { Redis } from 'ioredis'; type Octokit = { @@ -225,6 +226,15 @@ async function collectUnprocessedComments( for (const comment of commentsByTime) { const commentAuthor = comment.user.login; + + // `/split` is accepted only through issue_comment.created intake, where + // repository authorization and DB locking are enforced. Polling must not + // reinterpret it as a generic follow-up job. + if (parseSplitCommand(comment.body)) { + correlatedLogger.debug({ pullRequestNumber: pr.number, commentId: comment.id }, 'Skipping /split command in generic PR comment polling'); + continue; + } + const filterResult = filterCommentByAuthor(commentAuthor, correlationId); if (filterResult.shouldFilter) continue; diff --git a/test/prSplit/commandAuthorization.test.ts b/test/prSplit/commandAuthorization.test.ts new file mode 100644 index 000000000..a2ba45fd4 --- /dev/null +++ b/test/prSplit/commandAuthorization.test.ts @@ -0,0 +1,149 @@ +import assert from 'node:assert/strict'; +import { describe, mock, test } from 'node:test'; +import { + MAX_SPLIT_INSTRUCTION_LENGTH, + normalizeSplitInstruction, + parseSplitCommand, +} from '../../packages/core/src/services/prSplit/command.js'; +import { + authorizeSplitRequester, + isSplitPermissionAuthorized, + type PrSplitRequestClient, +} from '../../packages/core/src/services/prSplit/authorization.js'; + +describe('/split command parsing', () => { + test('accepts empty or natural-language guidance and normalizes whitespace', () => { + assert.deepEqual(parseSplitCommand('/split'), { instruction: '' }); + assert.deepEqual(parseSplitCommand('/split extract auth changes'), { + instruction: 'extract auth changes', + }); + assert.equal(normalizeSplitInstruction(' extract\n\tauth changes '), 'extract auth changes'); + assert.deepEqual(parseSplitCommand('/split\n extract\n\tauth changes'), { + instruction: 'extract auth changes', + }); + }); + + test('requires /split to be the exact first command token', () => { + assert.equal(parseSplitCommand('please /split this PR'), null); + assert.equal(parseSplitCommand(' /split this PR'), null); + assert.equal(parseSplitCommand('/review\n/split this PR'), null); + assert.equal(parseSplitCommand('/splitter this PR'), null); + assert.equal(parseSplitCommand('/SPLIT this PR'), null); + }); + + test('recognizes but rejects instructions beyond the execution limit', () => { + assert.deepEqual(parseSplitCommand(`/split ${'x'.repeat(MAX_SPLIT_INSTRUCTION_LENGTH + 1)}`), { + instruction: '', + validationError: 'instruction_too_long', + }); + }); +}); + +describe('/split repository authorization', () => { + const authorizationRequest = { + owner: 'integry', + repo: 'propr', + username: 'maintainer', + requesterId: 7654321, + }; + + test('maps only write-like GitHub permissions to authorized', () => { + for (const permission of ['write', 'maintain', 'admin']) { + assert.equal(isSplitPermissionAuthorized(permission), true, permission); + } + for (const permission of ['read', 'triage', 'none', '', null, undefined]) { + assert.equal(isSplitPermissionAuthorized(permission), false, String(permission)); + } + }); + + test('verifies repository access before treating a collaborator 404 as definitive', async () => { + const requestedRoutes: string[] = []; + const octokit: PrSplitRequestClient = { + request: mock.fn(async (route: string) => { + requestedRoutes.push(route); + return { data: { permission: 'maintain', user: { id: 7654321 } } }; + }), + }; + assert.deepEqual(await authorizeSplitRequester(octokit, authorizationRequest), { + authorized: true, + permission: 'maintain', + }); + assert.deepEqual(requestedRoutes, [ + 'GET /repos/{owner}/{repo}/collaborators/{username}/permission', + ]); + + const notFoundRoutes: string[] = []; + const notFoundClient: PrSplitRequestClient = { + request: mock.fn(async (route: string) => { + notFoundRoutes.push(route); + if (route === 'GET /repos/{owner}/{repo}') return { data: { id: 123456 } }; + throw Object.assign(new Error('Collaborator not found'), { status: 404 }); + }), + }; + assert.deepEqual(await authorizeSplitRequester(notFoundClient, authorizationRequest), { + authorized: false, + permission: null, + }); + assert.deepEqual(notFoundRoutes, [ + 'GET /repos/{owner}/{repo}/collaborators/{username}/permission', + 'GET /repos/{owner}/{repo}', + ]); + + const repositoryError = Object.assign(new Error('Repository not found'), { status: 404 }); + const inaccessibleClient: PrSplitRequestClient = { + request: mock.fn(async (route: string) => { + if (route === 'GET /repos/{owner}/{repo}') throw repositoryError; + throw Object.assign(new Error('Collaborator not found'), { status: 404 }); + }), + }; + await assert.rejects( + authorizeSplitRequester(inaccessibleClient, authorizationRequest), + error => error === repositoryError, + ); + }); + + test('rejects a renamed or recycled login whose numeric identity differs', async () => { + const octokit: PrSplitRequestClient = { + request: mock.fn(async () => ({ + data: { permission: 'admin', user: { id: 9999999 } }, + })), + }; + + assert.deepEqual(await authorizeSplitRequester(octokit, authorizationRequest), { + authorized: false, + permission: 'admin', + }); + }); + + test('rethrows credential, rate-limit, and ambiguous 403 responses', async () => { + const retryableErrors = [ + Object.assign(new Error('Resource not accessible by integration'), { + status: 403, + response: { + data: { message: 'Resource not accessible by integration' }, + headers: {}, + }, + }), + Object.assign(new Error('API rate limit exceeded'), { + status: 403, + response: { + data: { message: 'API rate limit exceeded' }, + headers: { 'x-ratelimit-remaining': '0' }, + }, + }), + Object.assign(new Error('ambiguous GitHub 403'), { status: 403 }), + Object.assign(new Error('Too Many Requests'), { status: 429 }), + Object.assign(new Error('Service Unavailable'), { status: 503 }), + ]; + + for (const error of retryableErrors) { + const client: PrSplitRequestClient = { + request: mock.fn(async () => { throw error; }), + }; + await assert.rejects( + authorizeSplitRequester(client, authorizationRequest), + (caught) => caught === error, + ); + } + }); +}); diff --git a/test/prSplit/helpers.ts b/test/prSplit/helpers.ts new file mode 100644 index 000000000..fe302bc05 --- /dev/null +++ b/test/prSplit/helpers.ts @@ -0,0 +1,79 @@ +import assert from 'node:assert/strict'; +import knex, { type Knex } from 'knex'; +import type { IssueCommentEvent } from '@octokit/webhooks-types'; +import { up as createPrSplitTables } from '../../packages/core/src/db/migrations/20260804000000_create_pr_split_operations.js'; +import type { PrSplitCommandRecord } from '../../packages/core/src/services/prSplit/commandStore.js'; +import type { + CreatePrSplitOperationInput, + PrSplitOperation, +} from '../../packages/core/src/services/prSplit/operationStore.js'; + +export const BASE_SPLIT_INPUT: CreatePrSplitOperationInput = { + repositoryId: 123456, + repository: 'Integry/ProPR', + sourcePrNumber: 1735, + baseRef: '1735-epic-pr-split-rjb', + baseSha: 'AAA111', + headSha: 'BBB222', + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', +}; + +export async function createSplitTestDatabase(filename = ':memory:'): Promise { + const database = knex({ + client: 'better-sqlite3', + connection: { filename }, + useNullAsDefault: true, + pool: { min: 1, max: 1 }, + }); + await createPrSplitTables(database); + return database; +} + +export function requiredOperation(record: PrSplitCommandRecord): PrSplitOperation { + assert.ok(record.operation, `receipt ${record.receipt.event_key} should reference an operation`); + return record.operation; +} + +export function issueCommentPayload( + body: string, + options: { + commentId?: number; + isPullRequest?: boolean; + repository?: string; + } = {}, +): IssueCommentEvent { + const repository = options.repository ?? 'integry/propr'; + const [owner, repo] = repository.split('/'); + return { + action: 'created', + issue: { + number: 1735, + ...(options.isPullRequest === false + ? {} + : { pull_request: { url: 'https://api.github.test/pulls/1735' } }), + }, + comment: { + id: options.commentId ?? 9001, + body, + user: { id: 7654321, login: 'maintainer' }, + }, + repository: { + id: 123456, + name: repo, + full_name: repository, + owner: { login: owner }, + }, + } as unknown as IssueCommentEvent; +} + +export function openPullRequestData() { + return { + base: { ref: '1735-epic-pr-split-rjb', sha: 'aaa111' }, + head: { sha: 'bbb222' }, + state: 'open', + merged: false, + }; +} diff --git a/test/prSplit/intake.test.ts b/test/prSplit/intake.test.ts new file mode 100644 index 000000000..e07ea98e4 --- /dev/null +++ b/test/prSplit/intake.test.ts @@ -0,0 +1,642 @@ +import assert from 'node:assert/strict'; +import { afterEach, beforeEach, describe, mock, test } from 'node:test'; +import type { Knex } from 'knex'; +import type { PrSplitRequestClient } from '../../packages/core/src/services/prSplit/authorization.js'; +import { + DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT, + DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS, + claimPrSplitCommandResponse, + getPrSplitCommandRecord, + markPrSplitCommandResponsePosted, + recordPrSplitCommandOutcome, +} from '../../packages/core/src/services/prSplit/commandStore.js'; +import { MAX_SPLIT_INSTRUCTION_LENGTH } from '../../packages/core/src/services/prSplit/command.js'; +import { + handlePrSplitComment, + isPrSplitExecutionEnabled, + type PrSplitIntakeDependencies, +} from '../../packages/core/src/services/prSplit/intake.js'; +import { + updatePrSplitOperationStatus, + type PrSplitOperation, +} from '../../packages/core/src/services/prSplit/operationStore.js'; +import { + createSplitTestDatabase, + issueCommentPayload, + openPullRequestData, +} from './helpers.js'; + +describe('/split issue-comment intake', () => { + let database: Knex; + + beforeEach(async () => { + database = await createSplitTestDatabase(); + }); + + afterEach(async () => { + await database.destroy(); + }); + + function dependencies( + client: PrSplitRequestClient, + overrides: Partial = {}, + ): PrSplitIntakeDependencies { + return { + getOctokit: mock.fn(async () => client), + authorizeRequester: mock.fn(async () => ({ + authorized: true as const, + permission: 'write' as const, + })), + isExecutionEnabled: () => true, + getResponseAuthorLogin: () => 'propr-dev[bot]', + db: database, + ...overrides, + }; + } + + test('keeps staged intake disabled unless explicitly enabled', () => { + assert.equal(isPrSplitExecutionEnabled(undefined), false); + assert.equal(isPrSplitExecutionEnabled('false'), false); + assert.equal(isPrSplitExecutionEnabled('1'), true); + assert.equal(isPrSplitExecutionEnabled('TRUE'), true); + }); + + test('ignores /split on a normal issue before loading GitHub dependencies', async () => { + const client: PrSplitRequestClient = { + request: mock.fn(async () => ({ data: [] })), + }; + const getOctokit = mock.fn(async () => client); + const result = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes', { isPullRequest: false }), + 'correlation-id', + { getOctokit, db: database }, + ); + assert.deepEqual(result, { handled: false }); + assert.equal(getOctokit.mock.callCount(), 0); + }); + + test('durably preserves a disabled disposition after execution is enabled', async () => { + const request = mock.fn(async () => ({ data: { id: 42 } })); + const client: PrSplitRequestClient = { request }; + const authorizeRequester = mock.fn(async () => ({ + authorized: true as const, + permission: 'write' as const, + })); + let enabled = false; + const intakeDependencies = dependencies(client, { + authorizeRequester, + isExecutionEnabled: () => enabled, + }); + const payload = issueCommentPayload('/split extract auth changes'); + + const first = await handlePrSplitComment(payload, 'first-delivery', intakeDependencies); + enabled = true; + const replay = await handlePrSplitComment(payload, 'redelivery', intakeDependencies); + + for (const result of [first, replay]) { + assert.equal(result.handled && result.outcome, 'disabled'); + assert.equal(result.handled && result.disposition.reason, 'split_execution_not_enabled'); + } + assert.equal(authorizeRequester.mock.callCount(), 0); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 0); + assert.equal(request.mock.calls.filter( + call => call.arguments[0] === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ).length, 1); + const receipt = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001, + }, database); + assert.equal(receipt?.receipt.outcome, 'disabled'); + assert.equal(receipt?.receipt.response_state, 'posted'); + assert.equal(receipt?.receipt.requester_id, 7654321); + }); + + test('durably preserves an authorization refusal after permission changes', async () => { + const request = mock.fn(async () => ({ data: { id: 42 } })); + const client: PrSplitRequestClient = { request }; + let authorized = false; + const authorizeRequester = mock.fn(async () => authorized + ? { authorized: true as const, permission: 'write' as const } + : { authorized: false as const, permission: 'read' }); + const intakeDependencies = dependencies(client, { authorizeRequester }); + const payload = issueCommentPayload('/split extract auth changes'); + + const first = await handlePrSplitComment(payload, 'first-delivery', intakeDependencies); + authorized = true; + const replay = await handlePrSplitComment(payload, 'redelivery', intakeDependencies); + + assert.equal(first.handled && first.outcome, 'unauthorized'); + assert.equal(replay.handled && replay.outcome, 'unauthorized'); + assert.equal(authorizeRequester.mock.callCount(), 1); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 0); + assert.match(String(request.mock.calls[0]?.arguments[1].body), + /requires.*write.*maintain.*admin/i); + }); + + test('snapshots an open PR and queues normalized guidance with immutable identities', async () => { + const request = mock.fn(async ( + route: string, + _parameters: Record, + ): Promise<{ data: unknown }> => { + if (route === 'GET /repos/{owner}/{repo}/pulls/{pull_number}') { + return { data: openPullRequestData() }; + } + return { data: { id: 42 } }; + }); + const client: PrSplitRequestClient = { request }; + const result = await handlePrSplitComment( + issueCommentPayload('/split extract\n auth changes'), + 'correlation-id', + dependencies(client), + ); + + assert.equal(result.handled && result.outcome, 'queued'); + assert.equal(result.handled && result.disposition.billing?.seatConsumed, false); + assert.deepEqual(result.handled && result.disposition.evidence?.triggerCommentIds, [9001]); + const operation = await database('pr_split_operations').first(); + assert.equal(operation?.repository_id, 123456); + assert.equal(operation?.repository, 'integry/propr'); + assert.equal(operation?.requester_id, 7654321); + assert.equal(operation?.requester, 'maintainer'); + assert.equal(operation?.instruction, 'extract auth changes'); + assert.equal(operation?.base_ref, '1735-epic-pr-split-rjb'); + assert.equal(operation?.base_sha, 'aaa111'); + assert.equal(operation?.head_sha, 'bbb222'); + const postCall = request.mock.calls.find( + call => call.arguments[0] === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ); + assert.match(String(postCall?.arguments[1].body), /queued/i); + assert.match(String(postCall?.arguments[1].body), /propr:pr-split-response/); + assert.equal(request.mock.calls.some( + call => call.arguments[0] === 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + ), false); + }); + + test('keeps an active-lock refusal terminal after the owner finishes', async () => { + const request = mock.fn(async ( + route: string, + _parameters: Record, + ): Promise<{ data: unknown }> => { + if (route === 'GET /repos/{owner}/{repo}/pulls/{pull_number}') { + return { data: openPullRequestData() }; + } + return { data: { id: 42 } }; + }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client); + const queued = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes', { commentId: 9000 }), + 'first-command', + intakeDependencies, + ); + const blockedPayload = issueCommentPayload('/split extract API changes', { + commentId: 9001, + repository: 'integry/propr-renamed', + }); + const blocked = await handlePrSplitComment( + blockedPayload, + 'blocked-command', + intakeDependencies, + ); + assert.equal(blocked.handled && blocked.outcome, 'active'); + assert.equal(blocked.handled && blocked.disposition.reason, 'split_operation_already_active'); + if (!queued.handled || !queued.operation) assert.fail('first command did not queue'); + + const claimed = await updatePrSplitOperationStatus(queued.operation.id, 'running', {}, database); + assert.ok(claimed?.lease_token); + assert.ok(await updatePrSplitOperationStatus( + queued.operation.id, + 'completed', + { leaseToken: claimed.lease_token }, + database, + )); + const replay = await handlePrSplitComment( + blockedPayload, + 'blocked-redelivery', + intakeDependencies, + ); + assert.equal(replay.handled && replay.outcome, 'active'); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 1); + assert.equal(request.mock.calls.filter( + call => call.arguments[0] === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ).length, 2); + }); + + test('rejects closed and merged pull requests without creating operations', async () => { + const snapshots = [ + { ...openPullRequestData(), state: 'closed' }, + { ...openPullRequestData(), merged: true }, + ]; + const request = mock.fn(async (route: string): Promise<{ data: unknown }> => { + if (route === 'GET /repos/{owner}/{repo}/pulls/{pull_number}') { + return { data: snapshots.shift() }; + } + return { data: { id: 42 } }; + }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client); + + for (const commentId of [9001, 9002]) { + const result = await handlePrSplitComment( + issueCommentPayload('/split historical work', { commentId }), + `closed-${commentId}`, + intakeDependencies, + ); + assert.equal(result.handled && result.outcome, 'closed'); + assert.equal(result.handled && result.disposition.reason, 'split_pull_request_closed'); + } + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 0); + }); + + test('rejects oversized instructions durably before authorization', async () => { + const request = mock.fn(async () => ({ data: { id: 42 } })); + const client: PrSplitRequestClient = { request }; + const authorizeRequester = mock.fn(async () => ({ + authorized: true as const, + permission: 'write' as const, + })); + const result = await handlePrSplitComment( + issueCommentPayload(`/split ${'x'.repeat(MAX_SPLIT_INSTRUCTION_LENGTH + 1)}`), + 'correlation-id', + dependencies(client, { authorizeRequester }), + ); + assert.equal(result.handled && result.outcome, 'invalid'); + assert.equal(authorizeRequester.mock.callCount(), 0); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 0); + }); + + test('keeps a concurrent redelivery retryable while a response claim is live', async () => { + await recordPrSplitCommandOutcome({ + repositoryId: 123456, + repository: 'integry/propr', + sourcePrNumber: 1735, + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', + outcome: 'disabled', + }, database); + let signalPostStarted = (): void => undefined; + const postStarted = new Promise((resolve) => { signalPostStarted = resolve; }); + let releasePost = (): void => undefined; + const postCanFinish = new Promise((resolve) => { releasePost = resolve; }); + const request = mock.fn(async () => { + signalPostStarted(); + await postCanFinish; + return { data: { id: 42 } }; + }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client, { isExecutionEnabled: () => false }); + const payload = issueCommentPayload('/split extract auth changes'); + const firstDelivery = handlePrSplitComment(payload, 'first-delivery', intakeDependencies); + await postStarted; + try { + await assert.rejects( + handlePrSplitComment(payload, 'second-delivery', intakeDependencies), + /response claim.*still live/i, + ); + } finally { + releasePost(); + } + const result = await firstDelivery; + + assert.equal(result.handled && result.outcome, 'disabled'); + assert.equal(request.mock.calls.filter( + call => call.arguments[0] === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ).length, 1); + assert.equal(request.mock.calls.some( + call => call.arguments[0] === 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + ), false); + }); + + test('does not retry an ambiguous response POST while its claim lease is live', async () => { + const postError = new Error('socket closed after request write'); + const request = mock.fn(async () => { throw postError; }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client, { isExecutionEnabled: () => false }); + const payload = issueCommentPayload('/split extract auth changes'); + + await assert.rejects( + handlePrSplitComment(payload, 'first-delivery', intakeDependencies), + error => error === postError, + ); + await assert.rejects( + handlePrSplitComment(payload, 'redelivery', intakeDependencies), + /response claim.*still live/i, + ); + assert.equal(request.mock.callCount(), 1); + const receipt = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001, + }, database); + assert.equal(receipt?.receipt.response_state, 'claimed'); + }); + + test('reconciles the marker after an ambiguous POST instead of duplicating it', async () => { + const postError = new Error('socket closed after request write'); + let postedBody = ''; + let postedAt = ''; + const request = mock.fn(async ( + route: string, + parameters: Record, + ): Promise<{ data: unknown }> => { + if (route === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments') { + postedBody = String(parameters.body); + throw postError; + } + const comments = [{ + id: 42, + body: postedBody, + created_at: postedAt, + user: { login: 'propr-dev[bot]', type: 'Bot' }, + }]; + return { + data: comments.filter(comment => comment.created_at >= String(parameters.since)), + }; + }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client, { isExecutionEnabled: () => false }); + const payload = issueCommentPayload('/split extract auth changes'); + + await assert.rejects( + handlePrSplitComment(payload, 'first-delivery', intakeDependencies), + error => error === postError, + ); + const localCreatedAt = new Date(); + localCreatedAt.setMilliseconds(900); + const githubCreatedAt = new Date(localCreatedAt); + githubCreatedAt.setMilliseconds(0); + postedAt = githubCreatedAt.toISOString(); + await database('pr_split_command_receipts').update({ + created_at: localCreatedAt.toISOString(), + response_claimed_at: new Date( + Date.now() - DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS - 1_000, + ).toISOString(), + }); + + const replay = await handlePrSplitComment(payload, 'stale-redelivery', intakeDependencies); + assert.equal(replay.handled && replay.outcome, 'disabled'); + assert.deepEqual(request.mock.calls.map(call => call.arguments[0]), [ + 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + ]); + const reconciliationCall = request.mock.calls[1]; + assert.ok(String(reconciliationCall?.arguments[1].since) < postedAt); + const receipt = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001, + }, database); + assert.equal(receipt?.receipt.response_state, 'posted'); + assert.equal(receipt?.receipt.response_comment_id, 42); + }); + + test('ignores copied markers from other authors during reconciliation', async () => { + const record = await recordPrSplitCommandOutcome({ + repositoryId: 123456, + repository: 'integry/propr', + sourcePrNumber: 1735, + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', + outcome: 'disabled', + }, database); + await claimPrSplitCommandResponse( + record.receipt.event_key, + database, + new Date(Date.now() - DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS - 1_000), + ); + const marker = ``; + const request = mock.fn(async (route: string): Promise<{ data: unknown }> => { + if (route === 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments') { + return { + data: [{ + id: 41, + body: marker, + user: { login: 'maintainer', type: 'User' }, + }], + }; + } + return { data: { id: 42 } }; + }); + + const result = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes'), + 'copied-marker-redelivery', + dependencies({ request }, { isExecutionEnabled: () => false }), + ); + assert.equal(result.handled && result.outcome, 'disabled'); + assert.deepEqual(request.mock.calls.map(call => call.arguments[0]), [ + 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ]); + }); + + test('reconciles beyond one thousand newer comments without abandoning the claim', async () => { + const record = await recordPrSplitCommandOutcome({ + repositoryId: 123456, + repository: 'integry/propr', + sourcePrNumber: 1735, + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', + outcome: 'disabled', + }, database); + await claimPrSplitCommandResponse( + record.receipt.event_key, + database, + new Date(Date.now() - DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS - 1_000), + ); + const marker = ``; + const request = mock.fn(async ( + route: string, + parameters: Record, + ): Promise<{ data: unknown }> => { + assert.equal(route, 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments'); + const page = Number(parameters.page); + if (page <= 10) { + return { + data: Array.from({ length: 100 }, (_, index) => ({ + id: page * 100 + index, + body: 'unrelated comment', + user: { login: 'propr-dev[bot]', type: 'Bot' }, + })), + }; + } + return { + data: [{ + id: 4242, + body: marker, + user: { login: 'propr-dev[bot]', type: 'Bot' }, + }], + }; + }); + + const result = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes'), + 'active-pr-redelivery', + dependencies({ request }, { isExecutionEnabled: () => false }), + ); + assert.equal(result.handled && result.outcome, 'disabled'); + assert.equal(request.mock.callCount(), 11); + const receipt = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001, + }, database); + assert.equal(receipt?.receipt.response_comment_id, 4242); + }); + + test('recovers a stale claim left by a crash before the response request', async () => { + const record = await recordPrSplitCommandOutcome({ + repositoryId: 123456, + repository: 'integry/propr', + sourcePrNumber: 1735, + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', + outcome: 'disabled', + }, database); + const staleClaim = await claimPrSplitCommandResponse( + record.receipt.event_key, + database, + new Date(Date.now() - DEFAULT_PR_SPLIT_RESPONSE_CLAIM_LEASE_MS - 1_000), + ); + assert.equal(staleClaim?.needsReconciliation, false); + + const request = mock.fn(async (route: string): Promise<{ data: unknown }> => ( + route === 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments' + ? { data: [] } + : { data: { id: 42 } } + )); + const client: PrSplitRequestClient = { request }; + const result = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes'), + 'crash-redelivery', + dependencies(client, { isExecutionEnabled: () => false }), + ); + + assert.equal(result.handled && result.outcome, 'disabled'); + assert.deepEqual(request.mock.calls.map(call => call.arguments[0]), [ + 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments', + 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ]); + }); + + test('releases definitive GitHub rejections so a redelivery can retry', async () => { + const rejectionStatuses = [401, 403, 422, 429]; + let rejectionStatus: number | null = null; + const request = mock.fn(async (route: string): Promise<{ data: unknown }> => { + if ( + route === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments' + && rejectionStatus !== null + ) { + throw Object.assign(new Error(`GitHub rejected response with ${rejectionStatus}`), { + status: rejectionStatus, + }); + } + return { data: { id: 42 } }; + }); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client, { isExecutionEnabled: () => false }); + + for (const [index, status] of rejectionStatuses.entries()) { + const payload = issueCommentPayload('/split extract auth changes', { + commentId: 9001 + index, + }); + rejectionStatus = status; + await assert.rejects( + handlePrSplitComment(payload, `rejected-${status}`, intakeDependencies), + (error: unknown) => typeof error === 'object' + && error !== null + && 'status' in error + && error.status === status, + ); + const pending = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001 + index, + }, database); + assert.equal(pending?.receipt.response_state, 'pending'); + + rejectionStatus = null; + const replay = await handlePrSplitComment(payload, `retry-${status}`, intakeDependencies); + assert.equal(replay.handled && replay.outcome, 'disabled'); + } + assert.equal(request.mock.calls.filter( + call => call.arguments[0] === 'POST /repos/{owner}/{repo}/issues/{issue_number}/comments', + ).length, rejectionStatuses.length * 2); + }); + + test('returns a posted durable replay during a GitHub authentication outage', async () => { + const record = await recordPrSplitCommandOutcome({ + repositoryId: 123456, + repository: 'integry/propr', + sourcePrNumber: 1735, + requesterId: 7654321, + requester: 'maintainer', + originalCommentId: 9001, + instruction: 'extract auth changes', + outcome: 'disabled', + }, database); + const claim = await claimPrSplitCommandResponse(record.receipt.event_key, database); + assert.ok(claim); + assert.equal(await markPrSplitCommandResponsePosted( + record.receipt.event_key, + claim.token, + 42, + database, + ), true); + const authError = new Error('relay unavailable'); + const getOctokit = mock.fn(async (): Promise => { throw authError; }); + + const replay = await handlePrSplitComment( + issueCommentPayload('/split extract auth changes'), + 'auth-outage-redelivery', + { getOctokit, isExecutionEnabled: () => false, db: database }, + ); + assert.equal(replay.handled && replay.outcome, 'disabled'); + assert.equal(getOctokit.mock.callCount(), 0); + }); + + test('durably suppresses API responses after the per-user command limit', async () => { + const request = mock.fn(async () => ({ data: { id: 42 } })); + const client: PrSplitRequestClient = { request }; + const intakeDependencies = dependencies(client, { isExecutionEnabled: () => false }); + let limitedPayload = issueCommentPayload('/split extract auth changes'); + + for (let index = 0; index <= DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT; index += 1) { + limitedPayload = issueCommentPayload('/split extract auth changes', { + commentId: 9001 + index, + }); + const result = await handlePrSplitComment( + limitedPayload, + `rate-limit-${index}`, + intakeDependencies, + ); + assert.equal( + result.handled && result.outcome, + index < DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT ? 'disabled' : 'rate_limited', + ); + } + + const replay = await handlePrSplitComment( + limitedPayload, + 'rate-limited-redelivery', + intakeDependencies, + ); + assert.equal(replay.handled && replay.outcome, 'rate_limited'); + assert.equal(replay.handled && replay.disposition.reason, 'split_request_rate_limited'); + assert.equal(request.mock.callCount(), DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT); + const receipt = await getPrSplitCommandRecord({ + repositoryId: 123456, + originalCommentId: 9001 + DEFAULT_PR_SPLIT_COMMAND_RATE_LIMIT, + }, database); + assert.equal(receipt?.receipt.response_state, 'suppressed'); + }); +}); diff --git a/test/prSplit/interception.test.ts b/test/prSplit/interception.test.ts new file mode 100644 index 000000000..58691d4b1 --- /dev/null +++ b/test/prSplit/interception.test.ts @@ -0,0 +1,100 @@ +import assert from 'node:assert/strict'; +import { after, describe, mock, test } from 'node:test'; +import type { Redis } from 'ioredis'; +import { closeConnection } from '../../packages/core/src/db/connection.js'; +import { processCommentEvent } from '../../packages/core/src/webhook/commentEventHandler.js'; +import { + initializeWebhookHandler, + processWebhookEvent, +} from '../../packages/core/src/webhook/webhookHandler.js'; +import { pollForPullRequestComments } from '../../src/polling/prCommentPolling.js'; +import { issueCommentPayload } from './helpers.js'; + +after(async () => { + await closeConnection(); +}); + +describe('/split interception boundaries', () => { + test('intercepts webhook /split before the generic comment processor', async () => { + const commentProcessor = mock.fn(async () => ({ status: 'accepted' as const })); + const splitCommentHandler = mock.fn(async () => ({ + handled: true as const, + disposition: { + status: 'blocked' as const, + reason: 'split_execution_not_enabled', + billing: { seatConsumed: false }, + }, + outcome: 'disabled' as const, + })); + await initializeWebhookHandler({ + issueProcessor: mock.fn(async () => undefined), + commentProcessor, + commentDeletedHandler: mock.fn(async () => undefined), + commentEditedHandler: mock.fn(async () => undefined), + splitCommentHandler, + }); + + const disposition = await processWebhookEvent( + issueCommentPayload('/split extract auth changes'), + 'issue_comment', + 'correlation-id', + ); + assert.equal(disposition.reason, 'split_execution_not_enabled'); + assert.equal(splitCommentHandler.mock.callCount(), 1); + assert.equal(commentProcessor.mock.callCount(), 0); + }); + + test('the generic synthetic-comment path skips /split without touching Redis', async () => { + const redisGet = mock.fn(async () => null); + const result = await processCommentEvent( + issueCommentPayload('/split extract auth changes'), + 'issue_comment', + 'synthetic-correlation-id', + { + redisClient: { get: redisGet } as unknown as Redis, + PR_FOLLOWUP_TRIGGER_KEYWORDS: [], + }, + ); + assert.deepEqual(result, { + status: 'ignored', + reason: 'split_requires_webhook_intake', + }); + assert.equal(redisGet.mock.callCount(), 0); + }); + + test('polling skips /split without claiming or enqueueing generic follow-up work', async () => { + const redisGet = mock.fn(async () => null); + const paginate = mock.fn(async (endpoint: string): Promise => { + if (endpoint === 'GET /repos/{owner}/{repo}/pulls') { + return [{ + number: 1735, + title: 'Split this PR', + labels: [{ name: 'AI' }], + head: { ref: 'feature' }, + }]; + } + if (endpoint === 'GET /repos/{owner}/{repo}/issues/{issue_number}/comments') { + return [{ + id: 9001, + body: '/split extract auth changes', + user: { login: 'maintainer' }, + created_at: '2026-08-04T00:00:00.000Z', + }]; + } + return []; + }); + const octokit = { + paginate: async (endpoint: string, _options: Record): Promise => ( + await paginate(endpoint) + ) as T[], + }; + + await pollForPullRequestComments(octokit, 'integry/propr', 'poll-correlation-id', { + redisClient: { get: redisGet } as unknown as Redis, + PR_FOLLOWUP_TRIGGER_KEYWORDS: [], + MODEL_LABEL_PATTERN: '^llm-(.+)$', + }); + assert.equal(paginate.mock.callCount(), 3); + assert.equal(redisGet.mock.callCount(), 0); + }); +}); diff --git a/test/prSplit/operationRaceChild.ts b/test/prSplit/operationRaceChild.ts new file mode 100644 index 000000000..09b718c35 --- /dev/null +++ b/test/prSplit/operationRaceChild.ts @@ -0,0 +1,38 @@ +import knex from 'knex'; +import { + createOrGetPrSplitOperation, + reservePrSplitCommand, +} from '../../packages/core/src/services/prSplit/commandStore.js'; +import type { + CreatePrSplitOperationInput, +} from '../../packages/core/src/services/prSplit/operationStore.js'; + +const [filename, serializedInput, startAtValue, mode = 'operation'] = process.argv.slice(2); +if (!filename || !serializedInput || !startAtValue) { + throw new Error('Operation race child requires database, input, and start time arguments'); +} + +const input = JSON.parse(serializedInput) as CreatePrSplitOperationInput; +const startAt = Number(startAtValue); +const delay = Math.max(0, startAt - Date.now()); +await new Promise(resolve => setTimeout(resolve, delay)); + +const database = knex({ + client: 'better-sqlite3', + connection: { filename }, + useNullAsDefault: true, + pool: { min: 1, max: 1 }, +}); + +try { + await database.raw('PRAGMA busy_timeout = 1000'); + const result = mode === 'reserve' + ? await reservePrSplitCommand(input, database) + : await createOrGetPrSplitOperation(input, database); + process.stdout.write(JSON.stringify({ + outcome: result.receipt.outcome, + processId: process.pid, + })); +} finally { + await database.destroy(); +} diff --git a/test/prSplit/operationStore.test.ts b/test/prSplit/operationStore.test.ts new file mode 100644 index 000000000..34cd0f2a7 --- /dev/null +++ b/test/prSplit/operationStore.test.ts @@ -0,0 +1,463 @@ +import assert from 'node:assert/strict'; +import { execFile } from 'node:child_process'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, test } from 'node:test'; +import { promisify } from 'node:util'; +import knex, { type Knex } from 'knex'; +import { up as createPrSplitTables } from '../../packages/core/src/db/migrations/20260804000000_create_pr_split_operations.js'; +import { + createOrGetPrSplitOperation, + getPrSplitCommandRecord, + recordPrSplitCommandOutcome, +} from '../../packages/core/src/services/prSplit/commandStore.js'; +import { + buildSplitOperationDedupeKey, + buildSplitOperationEventKey, +} from '../../packages/core/src/services/prSplit/keys.js'; +import { + CANCELLED_QUEUED_SPLIT_OPERATION_ERROR, + DEFAULT_SPLIT_OPERATION_LEASE_MS, + STALE_SPLIT_OPERATION_ERROR, + assertPrSplitOperationLease, + cancelQueuedPrSplitOperation, + getActivePrSplitOperation, + getPrSplitOperation, + heartbeatPrSplitOperation, + recoverStalePrSplitOperations, + updatePrSplitOperationStatus, +} from '../../packages/core/src/services/prSplit/operationStore.js'; +import { + BASE_SPLIT_INPUT, + createSplitTestDatabase, + requiredOperation, +} from './helpers.js'; + +const execFileAsync = promisify(execFile); + +describe('PR split command and operation persistence', () => { + let database: Knex; + + beforeEach(async () => { + database = await createSplitTestDatabase(); + }); + + afterEach(async () => { + await database.destroy(); + }); + + test('keys use immutable repository identity and all snapshot inputs', () => { + const eventKey = buildSplitOperationEventKey(BASE_SPLIT_INPUT); + assert.equal(eventKey, buildSplitOperationEventKey({ + repositoryId: BASE_SPLIT_INPUT.repositoryId, + originalCommentId: BASE_SPLIT_INPUT.originalCommentId, + })); + assert.notEqual(eventKey, buildSplitOperationEventKey({ + repositoryId: BASE_SPLIT_INPUT.repositoryId, + originalCommentId: 9002, + })); + + const first = buildSplitOperationDedupeKey(BASE_SPLIT_INPUT); + assert.equal(first, buildSplitOperationDedupeKey({ + ...BASE_SPLIT_INPUT, + repository: 'integry/renamed', + headSha: 'bbb222', + instruction: ' extract\n auth changes ', + })); + for (const changed of [ + { ...BASE_SPLIT_INPUT, repositoryId: 999999 }, + { ...BASE_SPLIT_INPUT, instruction: 'extract database changes' }, + { ...BASE_SPLIT_INPUT, headSha: 'ccc333' }, + { ...BASE_SPLIT_INPUT, baseRef: 'main' }, + { ...BASE_SPLIT_INPUT, baseSha: 'ddd444' }, + ]) { + assert.notEqual(first, buildSplitOperationDedupeKey(changed)); + } + }); + + test('rejects invalid PR numbers and empty refs or SHAs at persistence boundaries', async () => { + for (const invalidInput of [ + { ...BASE_SPLIT_INPUT, sourcePrNumber: 0 }, + { ...BASE_SPLIT_INPUT, baseRef: ' ' }, + { ...BASE_SPLIT_INPUT, baseSha: '' }, + { ...BASE_SPLIT_INPUT, headSha: '\n' }, + ]) { + assert.throws(() => buildSplitOperationDedupeKey(invalidInput), RangeError); + await assert.rejects( + createOrGetPrSplitOperation(invalidInput, database), + RangeError, + ); + } + }); + + test('preserves the first terminal disposition for non-executable commands', async () => { + const first = await recordPrSplitCommandOutcome({ + ...BASE_SPLIT_INPUT, + instruction: ' extract\n auth changes ', + outcome: 'disabled', + }, database); + const replay = await recordPrSplitCommandOutcome({ + ...BASE_SPLIT_INPUT, + outcome: 'unauthorized', + }, database); + + assert.equal(first.receipt.outcome, 'disabled'); + assert.equal(first.replayed, false); + assert.equal(replay.receipt.outcome, 'disabled'); + assert.equal(replay.replayed, true); + assert.equal(replay.receipt.requester_id, BASE_SPLIT_INPUT.requesterId); + assert.equal(first.receipt.instruction, 'extract auth changes'); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 0); + }); + + test('semantically deduplicates distinct comments and stores a receipt for each', async () => { + const first = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const repeated = await createOrGetPrSplitOperation({ + ...BASE_SPLIT_INPUT, + repository: 'integry/propr-renamed', + originalCommentId: 9002, + instruction: ' extract auth\nchanges ', + }, database); + + const firstOperation = requiredOperation(first); + const repeatedOperation = requiredOperation(repeated); + assert.equal(first.receipt.outcome, 'queued'); + assert.equal(repeated.receipt.outcome, 'duplicate'); + assert.equal(repeated.receipt.duplicate_kind, 'semantic'); + assert.equal(repeatedOperation.id, firstOperation.id); + assert.equal(firstOperation.repository_id, BASE_SPLIT_INPUT.repositoryId); + assert.equal(firstOperation.requester_id, BASE_SPLIT_INPUT.requesterId); + assert.equal(firstOperation.instruction, 'extract auth changes'); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 1); + assert.equal(await database('pr_split_command_receipts').count('* as count').first() + .then(row => Number(row?.count)), 2); + }); + + test('redelivery resolves to its original queued disposition after terminal changes', async () => { + const first = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const operation = requiredOperation(first); + const claimed = await updatePrSplitOperationStatus(operation.id, 'running', {}, database); + assert.ok(claimed?.lease_token); + assert.ok(await updatePrSplitOperationStatus( + operation.id, + 'completed', + { leaseToken: claimed.lease_token }, + database, + )); + + const replay = await createOrGetPrSplitOperation({ + ...BASE_SPLIT_INPUT, + baseRef: 'main', + baseSha: 'ccc333', + headSha: 'ddd444', + instruction: 'different instruction after webhook retry', + }, database); + assert.equal(replay.receipt.outcome, 'queued'); + assert.equal(replay.replayed, true); + assert.equal(requiredOperation(replay).id, operation.id); + assert.equal(await database('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 1); + }); + + test('allows a distinct comment to retry an equivalent failed operation', async () => { + const first = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const firstOperation = requiredOperation(first); + assert.ok(await updatePrSplitOperationStatus( + firstOperation.id, + 'failed', + { errorMessage: 'worker failed' }, + database, + )); + + const retry = await createOrGetPrSplitOperation({ + ...BASE_SPLIT_INPUT, + originalCommentId: 9002, + instruction: ' extract auth\nchanges ', + }, database); + assert.equal(retry.receipt.outcome, 'queued'); + assert.notEqual(requiredOperation(retry).id, firstOperation.id); + }); + + test('keys the active mutex by repository ID and keeps active refusals terminal', async () => { + const first = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const firstOperation = requiredOperation(first); + const blockedInput = { + ...BASE_SPLIT_INPUT, + repository: 'integry/propr-renamed', + originalCommentId: 9002, + instruction: 'extract API changes', + }; + const blocked = await createOrGetPrSplitOperation(blockedInput, database); + assert.equal(blocked.receipt.outcome, 'active'); + assert.equal(blocked.receipt.repository, 'integry/propr-renamed'); + assert.equal(requiredOperation(blocked).id, firstOperation.id); + assert.equal( + (await getActivePrSplitOperation(BASE_SPLIT_INPUT.repositoryId, 1735, database))?.id, + firstOperation.id, + ); + + const claimed = await updatePrSplitOperationStatus(firstOperation.id, 'running', {}, database); + assert.ok(claimed?.lease_token); + assert.ok(await updatePrSplitOperationStatus( + firstOperation.id, + 'completed', + { leaseToken: claimed.lease_token }, + database, + )); + const replay = await createOrGetPrSplitOperation(blockedInput, database); + assert.equal(replay.receipt.outcome, 'active'); + assert.equal(replay.replayed, true); + assert.equal(requiredOperation(replay).id, firstOperation.id); + + const next = await createOrGetPrSplitOperation({ + ...blockedInput, + originalCommentId: 9003, + }, database); + assert.equal(next.receipt.outcome, 'queued'); + assert.notEqual(requiredOperation(next).id, firstOperation.id); + }); + + test('fences claims, heartbeats, side effects, and terminal transitions', async () => { + const created = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const operation = requiredOperation(created); + const startedAt = new Date(new Date(operation.created_at).getTime() + 1_000); + const claimed = await updatePrSplitOperationStatus( + operation.id, + 'running', + { now: startedAt, leaseDurationMs: 60_000 }, + database, + ); + assert.ok(claimed?.lease_token); + assert.equal(claimed.started_at, startedAt.toISOString()); + assert.equal(await updatePrSplitOperationStatus( + operation.id, + 'running', + { now: new Date(startedAt.getTime() + 1_000) }, + database, + ), null); + assert.equal(await heartbeatPrSplitOperation( + operation.id, + { leaseToken: 'stale-token', now: new Date(startedAt.getTime() + 2_000) }, + database, + ), null); + + const heartbeatAt = new Date(startedAt.getTime() + 5_000); + const heartbeat = await heartbeatPrSplitOperation( + operation.id, + { leaseToken: claimed.lease_token, now: heartbeatAt, leaseDurationMs: 60_000 }, + database, + ); + assert.ok(heartbeat); + assert.ok(await assertPrSplitOperationLease( + operation.id, + claimed.lease_token, + database, + new Date(heartbeatAt.getTime() + 1_000), + )); + assert.equal(await updatePrSplitOperationStatus( + operation.id, + 'completed', + { leaseToken: 'stale-token', now: new Date(heartbeatAt.getTime() + 2_000) }, + database, + ), null); + + const finishedAt = new Date(heartbeatAt.getTime() + 3_000); + const completed = await updatePrSplitOperationStatus( + operation.id, + 'completed', + { leaseToken: claimed.lease_token, now: finishedAt }, + database, + ); + assert.equal(completed?.finished_at, finishedAt.toISOString()); + assert.equal(completed?.lease_expires_at, null); + assert.equal(completed?.lease_token, null); + }); + + test('keeps queued backlogs claimable and recovers only expired running leases', async () => { + const queued = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const queuedOperation = requiredOperation(queued); + assert.equal(queuedOperation.heartbeat_at, null); + assert.equal(queuedOperation.lease_expires_at, null); + + const afterLongBacklog = new Date( + new Date(queuedOperation.created_at).getTime() + DEFAULT_SPLIT_OPERATION_LEASE_MS + 1, + ); + assert.equal(await recoverStalePrSplitOperations( + BASE_SPLIT_INPUT.repositoryId, + BASE_SPLIT_INPUT.sourcePrNumber, + database, + afterLongBacklog, + ), 0); + const claimed = await updatePrSplitOperationStatus( + queuedOperation.id, + 'running', + { now: afterLongBacklog, leaseDurationMs: 1_000 }, + database, + ); + assert.ok(claimed?.lease_token); + const expiredAt = new Date(afterLongBacklog.getTime() + 1_000); + assert.equal(await assertPrSplitOperationLease( + queuedOperation.id, + claimed.lease_token, + database, + expiredAt, + ), null); + assert.equal(await updatePrSplitOperationStatus( + queuedOperation.id, + 'completed', + { now: expiredAt, leaseToken: claimed.lease_token }, + database, + ), null); + assert.equal(await recoverStalePrSplitOperations( + BASE_SPLIT_INPUT.repositoryId, + BASE_SPLIT_INPUT.sourcePrNumber, + database, + expiredAt, + ), 1); + assert.equal((await getPrSplitOperation(queuedOperation.id, database))?.error_message, + STALE_SPLIT_OPERATION_ERROR); + }); + + test('administratively cancels abandoned queued work and releases its PR lock', async () => { + const queued = await createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database); + const operation = requiredOperation(queued); + const cancelledAt = new Date('2026-08-04T12:00:00.000Z'); + const cancelled = await cancelQueuedPrSplitOperation( + operation.id, + { now: cancelledAt }, + database, + ); + assert.equal(cancelled?.status, 'failed'); + assert.equal(cancelled?.error_message, CANCELLED_QUEUED_SPLIT_OPERATION_ERROR); + assert.equal(cancelled?.finished_at, cancelledAt.toISOString()); + assert.equal(await getActivePrSplitOperation( + BASE_SPLIT_INPUT.repositoryId, + BASE_SPLIT_INPUT.sourcePrNumber, + database, + ), null); + assert.equal(await cancelQueuedPrSplitOperation(operation.id, {}, database), null); + + const next = await createOrGetPrSplitOperation({ + ...BASE_SPLIT_INPUT, + originalCommentId: 9002, + instruction: 'extract API changes', + }, database); + assert.equal(next.receipt.outcome, 'queued'); + }); + + test('does not hide non-unique SQLite constraint failures', async () => { + await database.raw(` + CREATE TRIGGER reject_split_insert + BEFORE INSERT ON pr_split_operations + BEGIN + SELECT RAISE(ABORT, 'split trigger failure'); + END + `); + await assert.rejects( + createOrGetPrSplitOperation(BASE_SPLIT_INPUT, database), + /split trigger failure/, + ); + assert.equal( + (await getPrSplitCommandRecord(BASE_SPLIT_INPUT, database))?.receipt.outcome, + 'processing', + ); + }); + + test('uses separate processes to arbitrate a genuinely concurrent SQLite race', async () => { + const temporaryDirectory = await mkdtemp(path.join(tmpdir(), 'propr-split-race-')); + const filename = path.join(temporaryDirectory, 'operations.sqlite'); + const verificationDatabase = knex({ + client: 'better-sqlite3', + connection: { filename }, + useNullAsDefault: true, + pool: { min: 1, max: 1 }, + }); + + try { + await verificationDatabase.raw('PRAGMA journal_mode = WAL'); + await createPrSplitTables(verificationDatabase); + const childScript = path.resolve('test/prSplit/operationRaceChild.ts'); + const startAt = Date.now() + 500; + const runChild = async (input: typeof BASE_SPLIT_INPUT) => { + const { stdout } = await execFileAsync(process.execPath, [ + '--import', + 'tsx', + childScript, + filename, + JSON.stringify(input), + String(startAt), + ]); + return JSON.parse(stdout) as { outcome: string; processId: number }; + }; + const results = await Promise.all([ + runChild(BASE_SPLIT_INPUT), + runChild({ + ...BASE_SPLIT_INPUT, + originalCommentId: 9002, + instruction: 'extract API changes', + }), + ]); + assert.notEqual(results[0]?.processId, results[1]?.processId); + assert.deepEqual( + results.map(result => result.outcome).sort(), + ['active', 'queued'], + ); + assert.equal(await verificationDatabase('pr_split_operations').count('* as count').first() + .then(row => Number(row?.count)), 1); + assert.equal(await verificationDatabase('pr_split_command_receipts').count('* as count').first() + .then(row => Number(row?.count)), 2); + } finally { + await verificationDatabase.destroy(); + await rm(temporaryDirectory, { recursive: true, force: true }); + } + }); + + test('atomically reserves the per-user limit across separate processes', async () => { + const temporaryDirectory = await mkdtemp(path.join(tmpdir(), 'propr-split-limit-race-')); + const filename = path.join(temporaryDirectory, 'commands.sqlite'); + const verificationDatabase = knex({ + client: 'better-sqlite3', + connection: { filename }, + useNullAsDefault: true, + pool: { min: 1, max: 1 }, + }); + + try { + await verificationDatabase.raw('PRAGMA journal_mode = WAL'); + await createPrSplitTables(verificationDatabase); + const childScript = path.resolve('test/prSplit/operationRaceChild.ts'); + const startAt = Date.now() + 1_000; + const runChild = async (commentId: number) => { + const { stdout } = await execFileAsync(process.execPath, [ + '--import', + 'tsx', + childScript, + filename, + JSON.stringify({ ...BASE_SPLIT_INPUT, originalCommentId: commentId }), + String(startAt), + 'reserve', + ]); + return JSON.parse(stdout) as { outcome: string; processId: number }; + }; + const results = await Promise.all( + Array.from({ length: 8 }, (_, index) => runChild(9100 + index)), + ); + assert.equal(new Set(results.map(result => result.processId)).size, 8); + assert.deepEqual( + results.map(result => result.outcome).sort(), + ['processing', 'processing', 'processing', 'processing', 'processing', + 'rate_limited', 'rate_limited', 'rate_limited'], + ); + assert.equal(await verificationDatabase('pr_split_command_receipts') + .count('* as count').first().then(row => Number(row?.count)), 8); + assert.equal(await verificationDatabase('pr_split_operations') + .count('* as count').first().then(row => Number(row?.count)), 0); + } finally { + await verificationDatabase.destroy(); + await rm(temporaryDirectory, { recursive: true, force: true }); + } + }); +});