diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index dfb4fcd5285..0108e3626b7 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -16,6 +16,7 @@ import { issueComments, issueDocuments, issueRelations, + issueTreeHolds, issues, } from "@paperclipai/db"; import { @@ -119,6 +120,7 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = await db.delete(documentRevisions); await db.delete(documents); await db.delete(issueRelations); + await db.delete(issueTreeHolds); await db.delete(issues); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); @@ -343,4 +345,170 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = expect(promotedBlockedRun?.status).toBe("succeeded"); expect(blockedWakeRequestCount).toBeGreaterThanOrEqual(2); }); + + it("suppresses descendant wakeups while allowing root course-correction comment wakes under a pause hold", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const rootIssueId = randomUUID(); + const childIssueId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "SecurityEngineer", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + await db.insert(issues).values([ + { + id: rootIssueId, + companyId, + title: "Paused root", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + }, + { + id: childIssueId, + companyId, + parentId: rootIssueId, + title: "Paused child", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + }, + ]); + const [hold] = await db + .insert(issueTreeHolds) + .values({ + companyId, + rootIssueId, + mode: "pause", + status: "active", + reason: "security test hold", + releasePolicy: { strategy: "manual" }, + }) + .returning(); + + const blockedWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { issueId: childIssueId }, + contextSnapshot: { issueId: childIssueId, wakeReason: "issue_blockers_resolved" }, + }); + + expect(blockedWake).toBeNull(); + const skippedWake = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + }) + .from(agentWakeupRequests) + .where(sql`${agentWakeupRequests.payload} ->> 'issueId' = ${childIssueId}`) + .then((rows) => rows[0] ?? null); + expect(skippedWake).toMatchObject({ status: "skipped", reason: "issue_tree_hold_active" }); + + const rootCommentWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId: rootIssueId, commentId: randomUUID() }, + contextSnapshot: { issueId: rootIssueId, wakeReason: "issue_commented" }, + }); + + expect(rootCommentWake).not.toBeNull(); + const rootRun = await db + .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, rootCommentWake!.id)) + .then((rows) => rows[0] ?? null); + expect(rootRun?.contextSnapshot).toMatchObject({ + activeTreeHold: { + holdId: hold.id, + rootIssueId, + mode: "pause", + courseCorrectionOnly: true, + }, + }); + }); + + it("suppresses root comment wakes when the active hold is full_pause", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const rootIssueId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "SecurityEngineer", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + await db.insert(issues).values({ + id: rootIssueId, + companyId, + title: "Paused root", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + }); + await db.insert(issueTreeHolds).values({ + companyId, + rootIssueId, + mode: "pause", + status: "active", + reason: "full pause", + releasePolicy: { strategy: "manual", note: "full_pause" }, + }); + + const rootCommentWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId: rootIssueId, commentId: randomUUID() }, + contextSnapshot: { issueId: rootIssueId, wakeReason: "issue_commented" }, + }); + + expect(rootCommentWake).toBeNull(); + const skippedWake = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + }) + .from(agentWakeupRequests) + .where(sql`${agentWakeupRequests.payload} ->> 'issueId' = ${rootIssueId}`) + .then((rows) => rows[0] ?? null); + expect(skippedWake).toMatchObject({ status: "skipped", reason: "issue_tree_hold_active" }); + }); }); diff --git a/server/src/__tests__/issue-tree-control-routes.test.ts b/server/src/__tests__/issue-tree-control-routes.test.ts index bf5af98de3a..7bf87b849c3 100644 --- a/server/src/__tests__/issue-tree-control-routes.test.ts +++ b/server/src/__tests__/issue-tree-control-routes.test.ts @@ -11,11 +11,16 @@ const mockTreeControlService = vi.hoisted(() => ({ createHold: vi.fn(), getHold: vi.fn(), releaseHold: vi.fn(), + cancelUnclaimedWakeupsForTree: vi.fn(), })); const mockLogActivity = vi.hoisted(() => vi.fn()); +const mockHeartbeatService = vi.hoisted(() => ({ + cancelRun: vi.fn(), +})); vi.mock("../services/index.js", () => ({ + heartbeatService: () => mockHeartbeatService, issueService: () => mockIssueService, issueTreeControlService: () => mockTreeControlService, logActivity: mockLogActivity, @@ -44,6 +49,8 @@ describe("issue tree control routes", () => { id: "11111111-1111-4111-8111-111111111111", companyId: "company-2", }); + mockTreeControlService.cancelUnclaimedWakeupsForTree.mockResolvedValue([]); + mockHeartbeatService.cancelRun.mockResolvedValue(null); }); it("rejects cross-company preview requests before calling the preview service", async () => { @@ -81,4 +88,51 @@ describe("issue tree control routes", () => { expect(mockIssueService.getById).not.toHaveBeenCalled(); expect(mockTreeControlService.createHold).not.toHaveBeenCalled(); }); + + it("cancels active descendant runs when creating a pause hold", async () => { + const app = await createApp({ + type: "board", + userId: "user-1", + companyIds: ["company-2"], + source: "session", + isInstanceAdmin: false, + }); + mockTreeControlService.createHold.mockResolvedValue({ + hold: { + id: "33333333-3333-4333-8333-333333333333", + mode: "pause", + reason: "pause subtree", + }, + preview: { + mode: "pause", + totals: { affectedIssues: 1 }, + warnings: [], + activeRuns: [ + { + id: "44444444-4444-4444-8444-444444444444", + issueId: "11111111-1111-4111-8111-111111111111", + }, + ], + }, + }); + + const res = await request(app) + .post("/api/issues/11111111-1111-4111-8111-111111111111/tree-holds") + .send({ mode: "pause", reason: "pause subtree" }); + + expect(res.status).toBe(201); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("44444444-4444-4444-8444-444444444444"); + expect(mockTreeControlService.cancelUnclaimedWakeupsForTree).toHaveBeenCalledWith( + "company-2", + "11111111-1111-4111-8111-111111111111", + "Cancelled because an active subtree pause hold was created", + ); + expect(mockLogActivity).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + action: "issue.tree_hold_run_interrupted", + entityId: "44444444-4444-4444-8444-444444444444", + }), + ); + }); }); diff --git a/server/src/__tests__/issue-tree-control-service.test.ts b/server/src/__tests__/issue-tree-control-service.test.ts index 42b996e9922..81a1a3a0197 100644 --- a/server/src/__tests__/issue-tree-control-service.test.ts +++ b/server/src/__tests__/issue-tree-control-service.test.ts @@ -15,6 +15,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { issueTreeControlService } from "../services/issue-tree-control.js"; +import { issueService } from "../services/issues.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -209,4 +210,109 @@ describeEmbeddedPostgres("issueTreeControlService", () => { expect(released.releaseReason).toBe("operator resumed"); expect(released.members).toHaveLength(1); }); + + it("blocks held descendant checkout but allows root comment course-correction checkout", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const rootIssueId = randomUUID(); + const childIssueId = randomUUID(); + const rootRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "SecurityEngineer", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(issues).values([ + { + id: rootIssueId, + companyId, + title: "Paused root", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + }, + { + id: childIssueId, + companyId, + parentId: rootIssueId, + title: "Paused child", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + }, + ]); + await db.insert(heartbeatRuns).values({ + id: rootRunId, + companyId, + agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "queued", + contextSnapshot: { issueId: rootIssueId, wakeReason: "issue_commented" }, + }); + + const treeSvc = issueTreeControlService(db); + await treeSvc.createHold(companyId, rootIssueId, { + mode: "pause", + reason: "operator requested pause", + actor: { actorType: "user", actorId: "board-user", userId: "board-user" }, + }); + + const issueSvc = issueService(db); + await expect(issueSvc.checkout(childIssueId, agentId, ["todo"], randomUUID())).rejects.toMatchObject({ + status: 409, + details: expect.objectContaining({ + rootIssueId, + mode: "pause", + }), + }); + + const checkedOutRoot = await issueSvc.checkout(rootIssueId, agentId, ["todo"], rootRunId); + expect(checkedOutRoot.status).toBe("in_progress"); + expect(checkedOutRoot.checkoutRunId).toBe(rootRunId); + + await db.update(issues).set({ + status: "todo", + checkoutRunId: null, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }).where(eq(issues.id, rootIssueId)); + await db.update(issueTreeHolds).set({ + status: "released", + releasedAt: new Date(), + releasedByActorType: "user", + releasedByUserId: "board-user", + releaseReason: "switch to full pause", + updatedAt: new Date(), + }).where(eq(issueTreeHolds.rootIssueId, rootIssueId)); + await treeSvc.createHold(companyId, rootIssueId, { + mode: "pause", + reason: "full pause", + releasePolicy: { strategy: "manual", note: "full_pause" }, + actor: { actorType: "user", actorId: "board-user", userId: "board-user" }, + }); + + await expect(issueSvc.checkout(rootIssueId, agentId, ["todo"], rootRunId)).rejects.toMatchObject({ + status: 409, + details: expect.objectContaining({ + rootIssueId, + mode: "pause", + }), + }); + }); }); diff --git a/server/src/routes/issue-tree-control.ts b/server/src/routes/issue-tree-control.ts index e8aac4feb1b..53140348fd0 100644 --- a/server/src/routes/issue-tree-control.ts +++ b/server/src/routes/issue-tree-control.ts @@ -7,13 +7,14 @@ import { releaseIssueTreeHoldSchema, } from "@paperclipai/shared"; import { validate } from "../middleware/validate.js"; -import { issueService, issueTreeControlService, logActivity } from "../services/index.js"; +import { heartbeatService, issueService, issueTreeControlService, logActivity } from "../services/index.js"; import { assertBoard, assertCompanyAccess, getActorInfo } from "./authz.js"; export function issueTreeControlRoutes(db: Db) { const router = Router(); const issuesSvc = issueService(db); const treeControlSvc = issueTreeControlService(db); + const heartbeat = heartbeatService(db); async function resolveRootIssue(req: Request) { const rootIssueId = req.params.id as string; @@ -89,9 +90,90 @@ export function issueTreeControlRoutes(db: Db) { }, }); + if (result.hold.mode === "pause") { + const interruptedRunIds = [...new Set(result.preview.activeRuns.map((run) => run.id))]; + for (const runId of interruptedRunIds) { + await heartbeat.cancelRun(runId); + await logActivity(db, { + companyId: root.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + action: "issue.tree_hold_run_interrupted", + entityType: "heartbeat_run", + entityId: runId, + details: { + holdId: result.hold.id, + rootIssueId: root.id, + reason: "active_subtree_pause_hold", + }, + }); + } + + const cancelledWakeups = await treeControlSvc.cancelUnclaimedWakeupsForTree( + root.companyId, + root.id, + "Cancelled because an active subtree pause hold was created", + ); + for (const wakeup of cancelledWakeups) { + await logActivity(db, { + companyId: root.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + action: "issue.tree_hold_wakeup_deferred", + entityType: "agent_wakeup_request", + entityId: wakeup.id, + details: { + holdId: result.hold.id, + rootIssueId: root.id, + agentId: wakeup.agentId, + previousReason: wakeup.reason, + }, + }); + } + } + res.status(201).json(result); }); + router.get("/issues/:id/tree-control/state", async (req, res) => { + assertBoard(req); + const issueId = req.params.id as string; + const issue = await issuesSvc.getById(issueId); + if (!issue) { + res.status(404).json({ error: "Issue not found" }); + return; + } + assertCompanyAccess(req, issue.companyId); + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(issue.companyId, issue.id); + res.json({ activePauseHold }); + }); + + router.get("/issues/:id/tree-holds", async (req, res) => { + assertBoard(req); + const root = await resolveRootIssue(req); + if (!root) { + res.status(404).json({ error: "Root issue not found" }); + return; + } + assertCompanyAccess(req, root.companyId); + const statusParam = typeof req.query.status === "string" ? req.query.status : null; + const modeParam = typeof req.query.mode === "string" ? req.query.mode : null; + const includeMembers = req.query.includeMembers === "true"; + const holds = await treeControlSvc.listHolds(root.companyId, root.id, { + status: statusParam === "active" || statusParam === "released" ? statusParam : undefined, + mode: + modeParam === "pause" || modeParam === "resume" || modeParam === "cancel" || modeParam === "restore" + ? modeParam + : undefined, + includeMembers, + }); + res.json(holds); + }); + router.get("/issues/:id/tree-holds/:holdId", async (req, res) => { assertBoard(req); const root = await resolveRootIssue(req); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 06707b2119e..7fee1317dd1 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -80,6 +80,7 @@ import { sanitizeRuntimeServiceBaseEnv, } from "./workspace-runtime.js"; import { issueService } from "./issues.js"; +import { issueTreeControlService, isCourseCorrectionPauseReleasePolicy } from "./issue-tree-control.js"; import { getIssueContinuationSummaryDocument, refreshIssueContinuationSummary, @@ -167,6 +168,11 @@ function resolveCodexTransientFallbackMode(attempt: number): CodexTransientFallb return "fresh_session_safer_invocation"; } const RUNNING_ISSUE_WAKE_REASONS_REQUIRING_FOLLOWUP = new Set(["approval_approved"]); +const TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS = new Set([ + "issue_commented", + "issue_reopened_via_comment", + "issue_comment_mentioned", +]); const SESSIONED_LOCAL_ADAPTERS = new Set([ "claude_local", "codex_local", @@ -1843,6 +1849,7 @@ export function heartbeatService(db: Db) { const secretsSvc = secretService(db); const companySkills = companySkillService(db); const issuesSvc = issueService(db); + const treeControlSvc = issueTreeControlService(db); const executionWorkspacesSvc = executionWorkspaceService(db); const environmentsSvc = environmentService(db); const workspaceOperationsSvc = workspaceOperationService(db); @@ -3499,6 +3506,34 @@ export function heartbeatService(db: Db) { const issueId = readNonEmptyString(context.issueId); if (issueId) { + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(run.companyId, issueId); + const wakeReason = readNonEmptyString(context.wakeReason) ?? readNonEmptyString(context.reason); + const isRootCourseCorrectionWake = + activePauseHold?.isRoot === true && + isCourseCorrectionPauseReleasePolicy(activePauseHold.releasePolicy) && + Boolean(wakeReason && TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS.has(wakeReason)); + if (activePauseHold && !isRootCourseCorrectionWake) { + await cancelRunInternal(run.id, "Cancelled because issue is held by an active subtree pause hold"); + await logActivity(db, { + companyId: run.companyId, + actorType: "system", + actorId: "system", + agentId: run.agentId, + runId: run.id, + action: "issue.tree_hold_run_interrupted", + entityType: "heartbeat_run", + entityId: run.id, + details: { + issueId, + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + source: "heartbeat.claim_queued_run", + securityPrinciples: ["Complete Mediation", "Fail Securely", "Secure Defaults"], + }, + }); + return null; + } + const dependencyReadiness = await issuesSvc.listDependencyReadiness(run.companyId, [issueId]); const unresolvedBlockerCount = dependencyReadiness.get(issueId)?.unresolvedBlockerCount ?? 0; if (unresolvedBlockerCount > 0 && !allowsBlockedIssueInteractionWake(context)) { @@ -6083,7 +6118,37 @@ export function heartbeatService(db: Db) { const deferredPayload = parseObject(deferred.payload); const deferredContextSeed = parseObject(deferredPayload[DEFERRED_WAKE_CONTEXT_KEY]); + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(issue.companyId, issue.id); + const deferredWakeReason = readNonEmptyString(deferredContextSeed.wakeReason) ?? readNonEmptyString(deferred.reason); + const canPromoteRootCourseCorrection = + activePauseHold?.isRoot === true && + isCourseCorrectionPauseReleasePolicy(activePauseHold.releasePolicy) && + Boolean(deferredWakeReason && TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS.has(deferredWakeReason)); + if (activePauseHold && !canPromoteRootCourseCorrection) { + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: new Date(), + error: "Deferred wake suppressed by active subtree pause hold", + updatedAt: new Date(), + }) + .where(eq(agentWakeupRequests.id, deferred.id)); + continue; + } + const promotedContextSeed: Record = { ...deferredContextSeed }; + if (activePauseHold) { + const courseCorrectionOnly = isCourseCorrectionPauseReleasePolicy(activePauseHold.releasePolicy); + promotedContextSeed.activeTreeHold = { + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + mode: activePauseHold.mode, + reason: activePauseHold.reason, + releasePolicy: activePauseHold.releasePolicy, + courseCorrectionOnly, + }; + } const deferredCommentIds = extractWakeCommentIds(deferredContextSeed); const shouldReopenDeferredCommentWake = deferredCommentIds.length > 0 && (issue.status === "done" || issue.status === "cancelled"); @@ -6472,6 +6537,49 @@ export function heartbeatService(db: Db) { return null; } + if (issueId) { + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(agent.companyId, issueId); + if (activePauseHold) { + const wakeReason = readNonEmptyString(enrichedContextSnapshot.wakeReason) ?? reason; + const isRootCourseCorrectionWake = + activePauseHold.isRoot && + isCourseCorrectionPauseReleasePolicy(activePauseHold.releasePolicy) && + Boolean(wakeReason && TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS.has(wakeReason)); + + if (!isRootCourseCorrectionWake) { + await writeSkippedRequest("issue_tree_hold_active"); + await logActivity(db, { + companyId: agent.companyId, + actorType: "system", + actorId: "system", + agentId, + runId: null, + action: "issue.tree_hold_wakeup_deferred", + entityType: "issue", + entityId: issueId, + details: { + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + requestedReason: reason, + source, + triggerDetail, + securityPrinciples: ["Complete Mediation", "Fail Securely", "Secure Defaults"], + }, + }); + return null; + } + + enrichedContextSnapshot.activeTreeHold = { + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + mode: activePauseHold.mode, + reason: activePauseHold.reason, + releasePolicy: activePauseHold.releasePolicy, + courseCorrectionOnly: isCourseCorrectionPauseReleasePolicy(activePauseHold.releasePolicy), + }; + } + } + if (issueId) { // Mention-triggered wakes can request input from another agent, but they must // still respect the issue execution lock so a second agent cannot start on the diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 3226d33c927..15823013e55 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,6 +1,7 @@ -import { and, asc, eq, inArray } from "drizzle-orm"; +import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { + agentWakeupRequests, heartbeatRuns, issueTreeHoldMembers, issueTreeHolds, @@ -24,6 +25,15 @@ import { conflict, notFound, unprocessable } from "../errors.js"; type IssueRow = typeof issues.$inferSelect; type HoldRow = typeof issueTreeHolds.$inferSelect; type HoldMemberRow = typeof issueTreeHoldMembers.$inferSelect; +export type ActiveIssueTreePauseHoldGate = { + holdId: string; + rootIssueId: string; + issueId: string; + isRoot: boolean; + mode: "pause"; + reason: string | null; + releasePolicy: IssueTreeHoldReleasePolicy | null; +}; type ActorInput = { actorType: "user" | "agent" | "system"; actorId: string; @@ -51,6 +61,13 @@ function normalizeReleasePolicy( return releasePolicy ?? DEFAULT_RELEASE_POLICY; } +export function isCourseCorrectionPauseReleasePolicy( + releasePolicy: IssueTreeHoldReleasePolicy | null | undefined, +) { + const note = typeof releasePolicy?.note === "string" ? releasePolicy.note.trim().toLowerCase() : ""; + return note !== "full_pause"; +} + function coerceIssueStatus(status: string): IssueStatus { return ISSUE_STATUSES.includes(status as IssueStatus) ? (status as IssueStatus) : "backlog"; } @@ -242,18 +259,22 @@ export function issueTreeControlService(db: Db) { return result; } - async function activeRunsByIssueId(companyId: string, treeIssues: TreeIssue[]) { + async function activeRunsForTree(companyId: string, treeIssues: TreeIssue[]) { + const issueIds = treeIssues.map((issue) => issue.id); + if (issueIds.length === 0) return []; const runIds = treeIssues .map((issue) => issue.executionRunId) .filter((id): id is string => typeof id === "string" && id.length > 0); - const byIssueId = new Map(); - if (runIds.length === 0) return byIssueId; + const uniqueRunIds = [...new Set(runIds)]; + const issueIdFromContext = sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`; + const issueIdSet = new Set(issueIds); const rows = await db .select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, + issueIdFromContext, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, }) @@ -261,32 +282,36 @@ export function issueTreeControlService(db: Db) { .where( and( eq(heartbeatRuns.companyId, companyId), - inArray(heartbeatRuns.id, [...new Set(runIds)]), inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]), + uniqueRunIds.length > 0 + ? or(inArray(heartbeatRuns.id, uniqueRunIds), inArray(issueIdFromContext, issueIds)) + : inArray(issueIdFromContext, issueIds), ), ); - const runById = new Map(); - for (const row of rows) { - if (row.status === "queued" || row.status === "running") { - runById.set(row.id, row); - } - } - for (const issue of treeIssues) { - if (!issue.executionRunId) continue; - const run = runById.get(issue.executionRunId); - if (!run) continue; - const status = run.status === "queued" || run.status === "running" ? run.status : null; - if (!status) continue; - byIssueId.set(issue.id, { - id: run.id, - issueId: issue.id, - agentId: run.agentId, - status, - startedAt: run.startedAt, - createdAt: run.createdAt, - }); - } - return byIssueId; + + const issueIdByExecutionRunId = new Map( + treeIssues + .filter((issue) => issue.executionRunId) + .map((issue) => [issue.executionRunId as string, issue.id]), + ); + return rows + .map((run) => { + if (run.status !== "queued" && run.status !== "running") return null; + const issueId = run.issueIdFromContext && issueIdSet.has(run.issueIdFromContext) + ? run.issueIdFromContext + : issueIdByExecutionRunId.get(run.id) ?? null; + if (!issueId) return null; + return { + id: run.id, + issueId, + agentId: run.agentId, + status: run.status, + startedAt: run.startedAt, + createdAt: run.createdAt, + } satisfies ActiveRunRow; + }) + .filter((run): run is ActiveRunRow => run !== null) + .sort((a, b) => a.issueId.localeCompare(b.issueId) || a.createdAt.getTime() - b.createdAt.getTime()); } async function activeHoldsByIssueId(companyId: string, issueIds: string[]) { @@ -318,6 +343,58 @@ export function issueTreeControlService(db: Db) { return byIssueId; } + async function getActivePauseHoldGate( + companyId: string, + issueId: string, + ): Promise { + const activePauseHolds = await db + .select({ + id: issueTreeHolds.id, + rootIssueId: issueTreeHolds.rootIssueId, + reason: issueTreeHolds.reason, + releasePolicy: issueTreeHolds.releasePolicy, + }) + .from(issueTreeHolds) + .where( + and( + eq(issueTreeHolds.companyId, companyId), + eq(issueTreeHolds.status, "active"), + eq(issueTreeHolds.mode, "pause"), + ), + ) + .orderBy(asc(issueTreeHolds.createdAt), asc(issueTreeHolds.id)); + if (activePauseHolds.length === 0) return null; + + const holdByRootIssueId = new Map(activePauseHolds.map((hold) => [hold.rootIssueId, hold])); + let currentIssueId: string | null = issueId; + const visited = new Set(); + + while (currentIssueId && !visited.has(currentIssueId)) { + visited.add(currentIssueId); + const hold = holdByRootIssueId.get(currentIssueId); + if (hold) { + return { + holdId: hold.id, + rootIssueId: hold.rootIssueId, + issueId, + isRoot: hold.rootIssueId === issueId, + mode: "pause", + reason: hold.reason, + releasePolicy: (hold.releasePolicy as IssueTreeHoldReleasePolicy | null) ?? null, + }; + } + + const parent: { parentId: string | null } | null = await db + .select({ parentId: issues.parentId }) + .from(issues) + .where(and(eq(issues.id, currentIssueId), eq(issues.companyId, companyId))) + .then((rows) => rows[0] ?? null); + currentIssueId = parent?.parentId ?? null; + } + + return null; + } + async function preview( companyId: string, rootIssueId: string, @@ -328,10 +405,14 @@ export function issueTreeControlService(db: Db) { ): Promise { const treeIssues = await listTreeIssues(companyId, rootIssueId); const issueIds = treeIssues.map((issue) => issue.id); - const [runsByIssueId, holdsByIssueId] = await Promise.all([ - activeRunsByIssueId(companyId, treeIssues), + const [activeRunRows, holdsByIssueId] = await Promise.all([ + activeRunsForTree(companyId, treeIssues), activeHoldsByIssueId(companyId, issueIds), ]); + const runsByIssueId = new Map(); + for (const run of activeRunRows) { + if (!runsByIssueId.has(run.issueId)) runsByIssueId.set(run.issueId, run); + } const countsByStatus: Partial> = {}; const issuesToPreview = treeIssues.map((issue) => { @@ -361,7 +442,7 @@ export function issueTreeControlService(db: Db) { } satisfies IssueTreePreviewIssue; }); const skippedIssues = issuesToPreview.filter((issue) => issue.skipped); - const activeRuns = [...runsByIssueId.values()] + const activeRuns = activeRunRows .map(toPreviewRun) .sort((a, b) => a.issueId.localeCompare(b.issueId) || a.id.localeCompare(b.id)); const affectedAgents = buildAffectedAgents(issuesToPreview); @@ -467,6 +548,53 @@ export function issueTreeControlService(db: Db) { return toHold(hold, members); } + async function listHolds( + companyId: string, + rootIssueId: string, + input?: { + status?: IssueTreeHold["status"]; + mode?: IssueTreeControlMode; + includeMembers?: boolean; + }, + ) { + const whereClauses = [ + eq(issueTreeHolds.companyId, companyId), + eq(issueTreeHolds.rootIssueId, rootIssueId), + ]; + if (input?.status) whereClauses.push(eq(issueTreeHolds.status, input.status)); + if (input?.mode) whereClauses.push(eq(issueTreeHolds.mode, input.mode)); + + const holds = await db + .select() + .from(issueTreeHolds) + .where(and(...whereClauses)) + .orderBy(asc(issueTreeHolds.createdAt), asc(issueTreeHolds.id)); + if (!input?.includeMembers || holds.length === 0) { + return holds.map((hold) => toHold(hold)); + } + + const holdIds = holds.map((hold) => hold.id); + const members = await db + .select() + .from(issueTreeHoldMembers) + .where( + and( + eq(issueTreeHoldMembers.companyId, companyId), + inArray(issueTreeHoldMembers.holdId, holdIds), + ), + ) + .orderBy(asc(issueTreeHoldMembers.depth), asc(issueTreeHoldMembers.createdAt), asc(issueTreeHoldMembers.issueId)); + + const membersByHoldId = new Map(); + for (const member of members) { + const existing = membersByHoldId.get(member.holdId) ?? []; + existing.push(member); + membersByHoldId.set(member.holdId, existing); + } + + return holds.map((hold) => toHold(hold, membersByHoldId.get(hold.id) ?? [])); + } + async function releaseHold( companyId: string, rootIssueId: string, @@ -519,11 +647,43 @@ export function issueTreeControlService(db: Db) { return toHold(updated, members); } + async function cancelUnclaimedWakeupsForTree(companyId: string, rootIssueId: string, reason: string) { + const treeIssues = await listTreeIssues(companyId, rootIssueId); + const issueIds = treeIssues.map((issue) => issue.id); + if (issueIds.length === 0) return []; + const now = new Date(); + return db + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + inArray(sql`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds), + ), + ) + .returning({ + id: agentWakeupRequests.id, + agentId: agentWakeupRequests.agentId, + reason: agentWakeupRequests.reason, + payload: agentWakeupRequests.payload, + }); + } + return { listTreeIssues, preview, createHold, getHold, + listHolds, + getActivePauseHoldGate, releaseHold, + cancelUnclaimedWakeupsForTree, }; } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 751be6231f4..79dde01b71c 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -36,6 +36,11 @@ import { instanceSettingsService } from "./instance-settings.js"; import { redactCurrentUserText } from "../log-redaction.js"; import { resolveIssueGoalId, resolveNextIssueGoalId } from "./issue-goal-fallback.js"; import { getDefaultCompanyGoal } from "./goals.js"; +import { + issueTreeControlService, + isCourseCorrectionPauseReleasePolicy, + type ActiveIssueTreePauseHoldGate, +} from "./issue-tree-control.js"; const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]; const MAX_ISSUE_COMMENT_PAGE_LIMIT = 500; @@ -45,6 +50,11 @@ const ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE = 500; export const MAX_CHILD_ISSUES_CREATED_BY_HELPER = 25; const MAX_CHILD_COMPLETION_SUMMARIES = 20; const CHILD_COMPLETION_SUMMARY_BODY_MAX_CHARS = 500; +const TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS = new Set([ + "issue_commented", + "issue_reopened_via_comment", + "issue_comment_mentioned", +]); function assertTransition(from: string, to: string) { if (from === to) return; @@ -71,6 +81,12 @@ function applyStatusSideEffects( return patch; } +function readStringFromRecord(record: unknown, key: string) { + if (!record || typeof record !== "object") return null; + const value = (record as Record)[key]; + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + export interface IssueFilters { status?: string; assigneeAgentId?: string; @@ -871,6 +887,7 @@ async function lastActivityStatsForIssues( export function issueService(db: Db) { const instanceSettings = instanceSettingsService(db); + const treeControlSvc = issueTreeControlService(db); async function getIssueByUuid(id: string) { const row = await db @@ -924,6 +941,23 @@ export function issueService(db: Db) { } } + async function isRootCourseCorrectionCheckoutAllowed( + companyId: string, + checkoutRunId: string | null, + gate: ActiveIssueTreePauseHoldGate, + ) { + if (!gate.isRoot || !checkoutRunId || !isCourseCorrectionPauseReleasePolicy(gate.releasePolicy)) return false; + const run = await db + .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, checkoutRunId), eq(heartbeatRuns.companyId, companyId))) + .then((rows) => rows[0] ?? null); + const wakeReason = + readStringFromRecord(run?.contextSnapshot, "wakeReason") ?? + readStringFromRecord(run?.contextSnapshot, "reason"); + return Boolean(wakeReason && TREE_HOLD_COURSE_CORRECTION_WAKE_REASONS.has(wakeReason)); + } + async function assertAssignableUser(companyId: string, userId: string) { const membership = await db .select({ id: companyMemberships.id }) @@ -2191,6 +2225,19 @@ export function issueService(db: Db) { await assertAssignableAgent(issueCompany.companyId, agentId); const now = new Date(); + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(issueCompany.companyId, id); + if ( + activePauseHold && + !(await isRootCourseCorrectionCheckoutAllowed(issueCompany.companyId, checkoutRunId, activePauseHold)) + ) { + throw conflict("Issue checkout blocked by active subtree pause hold", { + issueId: id, + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + mode: activePauseHold.mode, + securityPrinciples: ["Complete Mediation", "Fail Securely", "Secure Defaults"], + }); + } await clearExecutionRunIfTerminal(id); diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index c2704128df1..b2856513747 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -1,6 +1,7 @@ import type { AskUserQuestionsAnswer, Approval, + CreateIssueTreeHold, DocumentRevision, FeedbackTargetType, FeedbackTrace, @@ -11,7 +12,11 @@ import type { IssueDocument, IssueLabel, IssueThreadInteraction, + IssueTreeControlPreview, + IssueTreeHold, IssueWorkProduct, + PreviewIssueTreeControl, + ReleaseIssueTreeHold, UpsertIssueDocument, } from "@paperclipai/shared"; import { api } from "./client"; @@ -79,6 +84,41 @@ export const issuesApi = { api.post(`/companies/${companyId}/issues`, data), update: (id: string, data: Record) => api.patch(`/issues/${id}`, data), + previewTreeControl: (id: string, data: PreviewIssueTreeControl) => + api.post(`/issues/${id}/tree-control/preview`, data), + createTreeHold: (id: string, data: CreateIssueTreeHold) => + api.post<{ hold: IssueTreeHold; preview: IssueTreeControlPreview }>(`/issues/${id}/tree-holds`, data), + getTreeHold: (id: string, holdId: string) => + api.get(`/issues/${id}/tree-holds/${holdId}`), + listTreeHolds: ( + id: string, + filters?: { + status?: "active" | "released"; + mode?: "pause" | "resume" | "cancel" | "restore"; + includeMembers?: boolean; + }, + ) => { + const params = new URLSearchParams(); + if (filters?.status) params.set("status", filters.status); + if (filters?.mode) params.set("mode", filters.mode); + if (filters?.includeMembers) params.set("includeMembers", "true"); + const qs = params.toString(); + return api.get(`/issues/${id}/tree-holds${qs ? `?${qs}` : ""}`); + }, + getTreeControlState: (id: string) => + api.get<{ + activePauseHold: { + holdId: string; + rootIssueId: string; + issueId: string; + isRoot: boolean; + mode: "pause"; + reason: string | null; + releasePolicy: { strategy: "manual" | "after_active_runs_finish"; note?: string | null } | null; + } | null; + }>(`/issues/${id}/tree-control/state`), + releaseTreeHold: (id: string, holdId: string, data: ReleaseIssueTreeHold) => + api.post(`/issues/${id}/tree-holds/${holdId}/release`, data), remove: (id: string) => api.delete(`/issues/${id}`), checkout: (id: string, agentId: string) => api.post(`/issues/${id}/checkout`, { diff --git a/ui/src/components/IssueChatThread.test.tsx b/ui/src/components/IssueChatThread.test.tsx index c6b163ac68e..9a85a862e86 100644 --- a/ui/src/components/IssueChatThread.test.tsx +++ b/ui/src/components/IssueChatThread.test.tsx @@ -571,6 +571,73 @@ describe("IssueChatThread", () => { }); }); + it("shows deferred wake badge only for hold-deferred queued comments", () => { + const root = createRoot(container); + + act(() => { + root.render( + + {}} + showComposer={false} + enableLiveTranscriptPolling={false} + /> + , + ); + }); + + expect(container.textContent).toContain("Deferred wake"); + + act(() => { + root.render( + + {}} + showComposer={false} + enableLiveTranscriptPolling={false} + /> + , + ); + }); + + expect(container.textContent).toContain("Queued"); + expect(container.textContent).not.toContain("Deferred wake"); + + act(() => { + root.unmount(); + }); + }); + it("stores and restores the composer draft per issue key", () => { vi.useFakeTimers(); const root = createRoot(container); diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 37b7bf5552d..6c5de990bc0 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -1155,6 +1155,8 @@ function IssueChatUserMessage({ message }: { message: ThreadMessage }) { const authorName = typeof custom.authorName === "string" ? custom.authorName : null; const authorUserId = typeof custom.authorUserId === "string" ? custom.authorUserId : null; const queued = custom.queueState === "queued" || custom.clientStatus === "queued"; + const queueReason = typeof custom.queueReason === "string" ? custom.queueReason : null; + const queueBadgeLabel = queueReason === "hold" ? "\u23f8 Deferred wake" : "Queued"; const pending = custom.clientStatus === "pending"; const queueTargetRunId = typeof custom.queueTargetRunId === "string" ? custom.queueTargetRunId : null; const [copied, setCopied] = useState(false); @@ -1191,7 +1193,7 @@ function IssueChatUserMessage({ message }: { message: ThreadMessage }) { {queued ? (
- ⏸ Deferred wake + {queueBadgeLabel} {queueTargetRunId && onInterruptQueued ? (