fix(server): suppress stale handoff alarms during live continuation (#9695)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The control plane records a successful-run handoff when productive work ends without a durable next-step disposition > - That handoff state was derived only from the latest activity event, without checking whether a corrective run or wake was currently alive > - As a result, actively progressing issues could still show a high-severity missing-disposition alarm and blocked-inbox row > - The same stale required event could also remain indefinitely when a later successful run correctly skipped recovery because another valid continuation path already existed > - This pull request makes the derived state liveness-aware, suppresses attention only while the live path exists, and resolves stale required events on valid-path skips > - The benefit is that productive work stays calm while genuine stalls still resurface automatically when liveness disappears ## Linked Issues or Issue Description - **Bug:** An issue whose latest successful-run handoff event is `required` continues to report a missing disposition even while a heartbeat run, scheduled retry, or queued/deferred/claimed wake is actively targeting that issue. - **Expected behavior:** The API should expose current continuation liveness, the blocked inbox should suppress the alarm only while that path remains live, and a later successful run that skips recovery because a valid path exists should durably resolve the stale event. - **Related but distinct:** #9370 changes disposition freshness at detection time; #8748 adds an explicit policy opt-out. This PR preserves detection/escalation policy and fixes read-time/current-liveness state. ## What Changed - Extended `SuccessfulRunHandoffState` with `hasLiveContinuation` and optional `liveRunId` evidence. - Added bounded liveness hydration for required handoff states using active heartbeat-run and wake-request signals. - Suppressed `missing_disposition` blocked-inbox rows only while a run, scheduled retry, or live wake targets the issue. - Added durable `issue.successful_run_handoff_resolved` logging when handoff detection skips because another valid continuation path owns the next action. - Added focused regressions for live/absent derived state, self-healing attention suppression, valid-path skip classification, and resolved-event logging. - Updated UI normalization and fixtures for the shared contract without changing rendering behavior. ## Verification - `pnpm --filter @paperclipai/shared typecheck` - `pnpm --filter @paperclipai/server typecheck` - `pnpm --filter @paperclipai/ui typecheck` - `pnpm vitest run server/src/services/recovery/successful-run-handoff.test.ts server/src/__tests__/issue-list-assignee-filter-routes.test.ts server/src/__tests__/issue-blocker-attention.test.ts` — 56 passed - `pnpm vitest run server/src/__tests__/heartbeat-process-recovery.test.ts -t "queues one finish-handoff wake when a successful run leaves in-progress work without a next action"` — 1 passed - `git diff --check` ## Risks - Low risk: no schema or migration changes, and detection, bounded correction attempts, and escalation behavior are unchanged. - Liveness lookups are limited to issues whose latest handoff state is `required`; blocked-inbox suppression reuses rows already loaded by that query path. - Suppression is read-time and self-healing: when the run or wake stops, the alarm returns on the next fetch. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex using `gpt-5.4`, tool-enabled software-engineering workflow with repository, shell, test, Git, GitHub, and Paperclip control-plane access. Context-window size is not exposed by this runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
a04a77c9d3
commit
6ec059ab4e
|
|
@ -522,6 +522,8 @@ export type SuccessfulRunHandoffStateKind = "required" | "resolved" | "escalated
|
|||
export interface SuccessfulRunHandoffState {
|
||||
state: SuccessfulRunHandoffStateKind;
|
||||
required: boolean;
|
||||
hasLiveContinuation: boolean;
|
||||
liveRunId?: string | null;
|
||||
sourceRunId: string | null;
|
||||
correctiveRunId: string | null;
|
||||
assigneeAgentId: string | null;
|
||||
|
|
|
|||
|
|
@ -803,6 +803,37 @@ describeEmbeddedPostgres("issue blocker attention", () => {
|
|||
owner: { type: "agent", agentId },
|
||||
action: { label: "Choose disposition" },
|
||||
});
|
||||
|
||||
const handoffRunId = await activeRun({ companyId, agentId, issueId: handoffId, current: false });
|
||||
const liveRows = await svc.list(companyId, { attention: "blocked" });
|
||||
expect(liveRows.some((row) => row.id === handoffId)).toBe(false);
|
||||
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, handoffRunId));
|
||||
const stoppedRows = await svc.list(companyId, { attention: "blocked" });
|
||||
expect(stoppedRows.find((row) => row.id === handoffId)?.blockedInboxAttention).toMatchObject({
|
||||
state: "missing_disposition",
|
||||
reason: "missing_successful_run_disposition",
|
||||
});
|
||||
|
||||
const scheduledRetryRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: scheduledRetryRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "scheduled_retry",
|
||||
contextSnapshot: { taskId: handoffId },
|
||||
scheduledRetryAt: new Date(Date.now() + 60_000),
|
||||
scheduledRetryAttempt: 1,
|
||||
});
|
||||
const scheduledRows = await svc.list(companyId, { attention: "blocked" });
|
||||
expect(scheduledRows.some((row) => row.id === handoffId)).toBe(false);
|
||||
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, scheduledRetryRunId));
|
||||
const exhaustedRows = await svc.list(companyId, { attention: "blocked" });
|
||||
expect(exhaustedRows.find((row) => row.id === handoffId)?.blockedInboxAttention).toMatchObject({
|
||||
state: "missing_disposition",
|
||||
reason: "missing_successful_run_disposition",
|
||||
});
|
||||
});
|
||||
|
||||
it("applies assigneeAgentId='null' as an IS NULL filter on the blocked-inbox path", async () => {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { activityLog, agents, companies, companyMemberships, createDb, heartbeatRuns, issues, principalPermissionGrants } from "@paperclipai/db";
|
||||
import {
|
||||
|
|
@ -16,6 +17,7 @@ import {
|
|||
} from "../routes/issues.js";
|
||||
import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
|
||||
import { ensureHumanRoleDefaultGrants } from "../services/principal-access-compatibility.js";
|
||||
import { resolveRequiredSuccessfulRunHandoffOnValidPath } from "../services/successful-run-handoff-state.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
|
@ -38,8 +40,8 @@ describeEmbeddedPostgres("issue list routes assigneeAgentId filter", () => {
|
|||
afterEach(async () => {
|
||||
__clearIssueListResponseCacheForTests();
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
await db.delete(principalPermissionGrants);
|
||||
await db.delete(companyMemberships);
|
||||
|
|
@ -233,6 +235,7 @@ describeEmbeddedPostgres("issue list routes assigneeAgentId filter", () => {
|
|||
successfulRunHandoff: {
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId,
|
||||
assigneeAgentId: ownerAgentId,
|
||||
},
|
||||
|
|
@ -242,6 +245,146 @@ describeEmbeddedPostgres("issue list routes assigneeAgentId filter", () => {
|
|||
expect(res.body[0]).not.toHaveProperty("goal");
|
||||
});
|
||||
|
||||
it("marks a required successful-run handoff live while a run targets the issue", async () => {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: uniqueIssuePrefix(),
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await seedCloudTenantMember(companyId);
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Assignee",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Live handoff issue",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "system",
|
||||
actorId: "system",
|
||||
action: "issue.successful_run_handoff_required",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
agentId,
|
||||
details: { sourceRunId: randomUUID() },
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
contextSnapshot: { taskId: issueId },
|
||||
});
|
||||
|
||||
const app = createApp(companyId);
|
||||
const res = await request(app)
|
||||
.get(`/api/companies/${companyId}/issues`)
|
||||
.query({ view: "compact", limit: "20" });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(200);
|
||||
expect(res.body[0]?.successfulRunHandoff).toMatchObject({
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: true,
|
||||
liveRunId: runId,
|
||||
});
|
||||
});
|
||||
|
||||
it("logs resolved when a valid-path skip closes a stale required handoff", async () => {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const resolverRunId = randomUUID();
|
||||
const sourceRunId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: uniqueIssuePrefix(),
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Assignee",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: resolverRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "succeeded",
|
||||
contextSnapshot: { issueId },
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
identifier: `${uniqueIssuePrefix()}-1`,
|
||||
title: "Stale handoff",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "system",
|
||||
actorId: "heartbeat",
|
||||
action: "issue.successful_run_handoff_required",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
agentId,
|
||||
details: { sourceRunId },
|
||||
createdAt: new Date("2026-07-01T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
await expect(resolveRequiredSuccessfulRunHandoffOnValidPath(db, {
|
||||
companyId,
|
||||
issueId,
|
||||
issueIdentifier: "PAP-1",
|
||||
agentId,
|
||||
runId: resolverRunId,
|
||||
skipReason: "persisted issue monitor owns the next action",
|
||||
})).resolves.toBe(true);
|
||||
|
||||
const resolved = await db
|
||||
.select()
|
||||
.from(activityLog)
|
||||
.where(eq(activityLog.entityId, issueId))
|
||||
.then((rows) => rows.find((row) => row.action === "issue.successful_run_handoff_resolved"));
|
||||
expect(resolved).toMatchObject({
|
||||
runId: resolverRunId,
|
||||
details: {
|
||||
sourceRunId,
|
||||
resolvedByRunId: resolverRunId,
|
||||
resolvedBySkipReason: "persisted issue monitor owns the next action",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("returns 304 for unchanged compact issue list ETags", async () => {
|
||||
const companyId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
|
|
|
|||
|
|
@ -120,6 +120,7 @@ import {
|
|||
workProductService,
|
||||
} from "../services/index.js";
|
||||
import { buildPlanReviewContext } from "../services/plan-review-context.js";
|
||||
import { hydrateSuccessfulRunHandoffLiveness } from "../services/successful-run-handoff-state.js";
|
||||
import {
|
||||
TASK_WATCHDOG_ORIGIN_KIND,
|
||||
resolveTaskWatchdogMutationScope,
|
||||
|
|
@ -690,6 +691,7 @@ function successfulRunHandoffStateFromActivity(row: {
|
|||
return {
|
||||
state,
|
||||
required: state === "required",
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId:
|
||||
readNonEmptyString(details.sourceRunId)
|
||||
?? readNonEmptyString(details.source_run_id)
|
||||
|
|
@ -716,6 +718,7 @@ async function listSuccessfulRunHandoffStates(
|
|||
db: Db,
|
||||
companyId: string,
|
||||
issueIds: string[],
|
||||
options?: { hydrateLiveness?: boolean },
|
||||
): Promise<Map<string, SuccessfulRunHandoffState>> {
|
||||
if (issueIds.length === 0) return new Map();
|
||||
const rows = await db
|
||||
|
|
@ -742,7 +745,9 @@ async function listSuccessfulRunHandoffStates(
|
|||
const state = successfulRunHandoffStateFromActivity(row);
|
||||
if (state) states.set(row.entityId, state);
|
||||
}
|
||||
return states;
|
||||
return options?.hydrateLiveness === false
|
||||
? states
|
||||
: hydrateSuccessfulRunHandoffLiveness(db, companyId, states);
|
||||
}
|
||||
|
||||
type RecoveryActionsLister = {
|
||||
|
|
@ -8058,7 +8063,7 @@ export function issueRoutes(
|
|||
});
|
||||
|
||||
if (existing.status === "in_progress" && issue.status !== existing.status && issue.status !== "in_progress") {
|
||||
await listSuccessfulRunHandoffStates(db, issue.companyId, [issue.id])
|
||||
await listSuccessfulRunHandoffStates(db, issue.companyId, [issue.id], { hydrateLiveness: false })
|
||||
.then(async (handoffStates) => {
|
||||
const handoff = handoffStates.get(issue.id);
|
||||
if (handoff?.state !== "required") return;
|
||||
|
|
|
|||
|
|
@ -198,6 +198,7 @@ import {
|
|||
decideSuccessfulRunHandoff,
|
||||
findExistingFinishSuccessfulRunHandoffWake,
|
||||
findExistingRunLivenessContinuationWake,
|
||||
isSuccessfulRunHandoffValidPathSkip,
|
||||
SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY,
|
||||
readContinuationAttempt,
|
||||
} from "./recovery/index.js";
|
||||
|
|
@ -208,6 +209,7 @@ import {
|
|||
} from "./recovery/model-profile-hint.js";
|
||||
import { recoveryService } from "./recovery/service.js";
|
||||
import { productivityReviewService } from "./productivity-review.js";
|
||||
import { resolveRequiredSuccessfulRunHandoffOnValidPath } from "./successful-run-handoff-state.js";
|
||||
import { taskWatchdogService } from "./task-watchdogs.js";
|
||||
import { withAgentStartLock } from "./agent-start-lock.js";
|
||||
import {
|
||||
|
|
@ -8089,6 +8091,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
idempotentWakeExists: Boolean(existingWake),
|
||||
});
|
||||
|
||||
if (isSuccessfulRunHandoffValidPathSkip(decision) && issue) {
|
||||
await resolveRequiredSuccessfulRunHandoffOnValidPath(db, {
|
||||
companyId: issue.companyId,
|
||||
issueId: issue.id,
|
||||
issueIdentifier: issue.identifier,
|
||||
agentId: run.agentId,
|
||||
runId: run.id,
|
||||
skipReason: decision.reason,
|
||||
});
|
||||
}
|
||||
|
||||
if (decision.kind !== "enqueue" || !issue) return;
|
||||
|
||||
if (hasUnmanagedBackgroundTaskEvidence(parseObject(run.resultJson))) {
|
||||
|
|
|
|||
|
|
@ -66,6 +66,10 @@ import {
|
|||
import { conflict, HttpError, notFound, unprocessable } from "../errors.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { parseObject } from "../adapters/utils.js";
|
||||
import {
|
||||
hydrateSuccessfulRunHandoffLiveness,
|
||||
SUCCESSFUL_RUN_HANDOFF_LIVE_WAKE_STATUSES,
|
||||
} from "./successful-run-handoff-state.js";
|
||||
import {
|
||||
defaultIssueExecutionWorkspaceSettingsForProject,
|
||||
gateProjectExecutionWorkspacePolicy,
|
||||
|
|
@ -2672,7 +2676,7 @@ async function blockedByMapForIssues(
|
|||
|
||||
const BLOCKED_INBOX_TERMINAL_STATUSES = ["done", "cancelled"] as const;
|
||||
const BLOCKED_INBOX_ACTIVE_RUN_STATUSES = ["queued", "running"] as const;
|
||||
const BLOCKED_INBOX_ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const;
|
||||
const BLOCKED_INBOX_ACTIVE_WAKE_STATUSES = SUCCESSFUL_RUN_HANDOFF_LIVE_WAKE_STATUSES;
|
||||
const BLOCKED_INBOX_PENDING_INTERACTION_STATUSES = ["pending"] as const;
|
||||
const BLOCKED_INBOX_PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"] as const;
|
||||
const BLOCKED_INBOX_RECOVERY_ORIGIN_KINDS = ["harness_liveness_escalation", "stranded_issue_recovery"] as const;
|
||||
|
|
@ -2794,6 +2798,7 @@ function readSuccessfulRunHandoffFromActivity(row: {
|
|||
return {
|
||||
state,
|
||||
required: state === "required",
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId:
|
||||
readStringFromRecord(details, "sourceRunId")
|
||||
?? readStringFromRecord(details, "source_run_id")
|
||||
|
|
@ -2818,6 +2823,7 @@ async function listSuccessfulRunHandoffMapForIssues(
|
|||
dbOrTx: any,
|
||||
companyId: string,
|
||||
issueIds: string[],
|
||||
options?: { hydrateLiveness?: boolean },
|
||||
): Promise<Map<string, SuccessfulRunHandoffState>> {
|
||||
const uniqueIssueIds = [...new Set(issueIds)];
|
||||
const states = new Map<string, SuccessfulRunHandoffState>();
|
||||
|
|
@ -2856,7 +2862,9 @@ async function listSuccessfulRunHandoffMapForIssues(
|
|||
}
|
||||
}
|
||||
|
||||
return states;
|
||||
return options?.hydrateLiveness === false
|
||||
? states
|
||||
: hydrateSuccessfulRunHandoffLiveness(dbOrTx, companyId, states);
|
||||
}
|
||||
|
||||
function externalWaitFromDescription(description: string | null): { owner: string; action: string } | null {
|
||||
|
|
@ -3033,7 +3041,10 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
: dbOrTx
|
||||
.select({
|
||||
companyId: heartbeatRuns.companyId,
|
||||
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
|
||||
issueId: sql<string | null>`coalesce(
|
||||
${heartbeatRuns.contextSnapshot} ->> 'issueId',
|
||||
${heartbeatRuns.contextSnapshot} ->> 'taskId'
|
||||
)`,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
})
|
||||
|
|
@ -3041,14 +3052,22 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.status, [...BLOCKED_INBOX_ACTIVE_RUN_STATUSES]),
|
||||
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, graphIssueIds),
|
||||
inArray(sql<string>`coalesce(
|
||||
${heartbeatRuns.contextSnapshot} ->> 'issueId',
|
||||
${heartbeatRuns.contextSnapshot} ->> 'taskId'
|
||||
)`, graphIssueIds),
|
||||
)),
|
||||
graphIssueIds.length === 0
|
||||
? Promise.resolve([])
|
||||
: dbOrTx
|
||||
.select({
|
||||
companyId: agentWakeupRequests.companyId,
|
||||
issueId: sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`,
|
||||
issueId: sql<string | null>`coalesce(
|
||||
${agentWakeupRequests.payload} ->> 'issueId',
|
||||
${agentWakeupRequests.payload} ->> 'taskId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'issueId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'taskId'
|
||||
)`,
|
||||
agentId: agentWakeupRequests.agentId,
|
||||
status: agentWakeupRequests.status,
|
||||
})
|
||||
|
|
@ -3056,15 +3075,22 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
inArray(agentWakeupRequests.status, [...BLOCKED_INBOX_ACTIVE_WAKE_STATUSES]),
|
||||
sql`${agentWakeupRequests.runId} is null`,
|
||||
inArray(sql<string>`${agentWakeupRequests.payload} ->> 'issueId'`, graphIssueIds),
|
||||
inArray(sql<string>`coalesce(
|
||||
${agentWakeupRequests.payload} ->> 'issueId',
|
||||
${agentWakeupRequests.payload} ->> 'taskId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'issueId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'taskId'
|
||||
)`, graphIssueIds),
|
||||
)),
|
||||
graphIssueIds.length === 0
|
||||
? Promise.resolve([])
|
||||
: dbOrTx
|
||||
.select({
|
||||
companyId: heartbeatRuns.companyId,
|
||||
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
|
||||
issueId: sql<string | null>`coalesce(
|
||||
${heartbeatRuns.contextSnapshot} ->> 'issueId',
|
||||
${heartbeatRuns.contextSnapshot} ->> 'taskId'
|
||||
)`,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
})
|
||||
|
|
@ -3072,7 +3098,10 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.status, "scheduled_retry"),
|
||||
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, graphIssueIds),
|
||||
inArray(sql<string>`coalesce(
|
||||
${heartbeatRuns.contextSnapshot} ->> 'issueId',
|
||||
${heartbeatRuns.contextSnapshot} ->> 'taskId'
|
||||
)`, graphIssueIds),
|
||||
)),
|
||||
graphIssueIds.length === 0
|
||||
? Promise.resolve([])
|
||||
|
|
@ -3105,7 +3134,7 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
inArray(approvals.status, [...BLOCKED_INBOX_PENDING_APPROVAL_STATUSES]),
|
||||
inArray(issueApprovals.issueId, graphIssueIds),
|
||||
)),
|
||||
listSuccessfulRunHandoffMapForIssues(dbOrTx, companyId, rowIssueIds),
|
||||
listSuccessfulRunHandoffMapForIssues(dbOrTx, companyId, rowIssueIds, { hydrateLiveness: false }),
|
||||
]);
|
||||
|
||||
const pendingInteractions = (interactionRows as BlockedInboxInteractionRow[]).map((row) => ({
|
||||
|
|
@ -3186,6 +3215,13 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
for (const row of approvalRows as BlockedInboxApprovalRow[]) {
|
||||
if (!approvalByIssueId.has(row.issueId)) approvalByIssueId.set(row.issueId, row);
|
||||
}
|
||||
const liveHandoffRunIssueIds = new Set([
|
||||
...(activeRunRows as Array<{ issueId: string | null }>),
|
||||
...(scheduledRetryRows as Array<{ issueId: string | null }>),
|
||||
].flatMap((row) => row.issueId ? [row.issueId] : []));
|
||||
const liveHandoffWakeIssueIds = new Set(
|
||||
(wakeRows as Array<{ issueId: string | null }>).flatMap((row) => row.issueId ? [row.issueId] : []),
|
||||
);
|
||||
|
||||
for (const row of issueRows) {
|
||||
if (row.companyId !== companyId || BLOCKED_INBOX_TERMINAL_STATUSES.includes(row.status as typeof BLOCKED_INBOX_TERMINAL_STATUSES[number]) || row.hiddenAt) {
|
||||
|
|
@ -3193,7 +3229,11 @@ async function listIssueBlockedInboxAttentionMap(
|
|||
}
|
||||
const source = issueRef(row);
|
||||
const handoff = handoffMap.get(row.id);
|
||||
if (handoff && (handoff.required || handoff.state === "escalated")) {
|
||||
const hasLiveHandoffContinuation = Boolean(
|
||||
handoff?.state === "required"
|
||||
&& (liveHandoffRunIssueIds.has(row.id) || liveHandoffWakeIssueIds.has(row.id))
|
||||
);
|
||||
if (handoff && !hasLiveHandoffContinuation && (handoff.required || handoff.state === "escalated")) {
|
||||
result.set(row.id, attentionBase({
|
||||
state: "missing_disposition",
|
||||
reason: "missing_successful_run_disposition",
|
||||
|
|
|
|||
|
|
@ -56,6 +56,7 @@ export {
|
|||
buildSuccessfulRunHandoffRequiredNotice,
|
||||
decideSuccessfulRunHandoff,
|
||||
findExistingFinishSuccessfulRunHandoffWake,
|
||||
isSuccessfulRunHandoffValidPathSkip,
|
||||
isSuccessfulRunHandoffRequiredNoticeBody,
|
||||
} from "./successful-run-handoff.js";
|
||||
export type {
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import {
|
|||
buildSuccessfulRunHandoffRequiredNotice,
|
||||
decideSuccessfulRunHandoff,
|
||||
isIdempotentFinishSuccessfulRunHandoffWakeStatus,
|
||||
isSuccessfulRunHandoffValidPathSkip,
|
||||
isSuccessfulRunHandoffRequiredNoticeBody,
|
||||
noticeMetadataReferencesRecoveryAction,
|
||||
} from "./successful-run-handoff.js";
|
||||
|
|
@ -130,6 +131,12 @@ describe("successful run handoff decision", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("identifies valid-path skips that can durably resolve a stale required event", () => {
|
||||
expect(isSuccessfulRunHandoffValidPathSkip(decide({ hasActiveExecutionPath: true }))).toBe(true);
|
||||
expect(isSuccessfulRunHandoffValidPathSkip(decide({ hasQueuedWake: true }))).toBe(true);
|
||||
expect(isSuccessfulRunHandoffValidPathSkip(decide({ budgetBlocked: true }))).toBe(false);
|
||||
});
|
||||
|
||||
it("does not treat killed background-task evidence as a missing live path when a durable monitor owns the wait", () => {
|
||||
expect(decide({
|
||||
detectedProgressSummary: UNMANAGED_BACKGROUND_TASK_LIVENESS_REASON,
|
||||
|
|
|
|||
|
|
@ -88,6 +88,25 @@ export type SuccessfulRunHandoffDecision =
|
|||
reason: string;
|
||||
};
|
||||
|
||||
const SUCCESSFUL_RUN_HANDOFF_VALID_PATH_SKIP_REASONS = new Set([
|
||||
"issue has execution policy state",
|
||||
"active routine continuation owns the next action",
|
||||
"issue already has an active execution path",
|
||||
"issue already has a queued or deferred wake",
|
||||
"pending interaction or approval owns the next action",
|
||||
"persisted issue monitor owns the next action",
|
||||
"explicit blocker path owns the next action",
|
||||
"open recovery issue owns the ambiguity",
|
||||
"issue is under an active pause hold",
|
||||
"corrective handoff wake already exists for this source run",
|
||||
]);
|
||||
|
||||
export function isSuccessfulRunHandoffValidPathSkip(
|
||||
decision: SuccessfulRunHandoffDecision,
|
||||
): decision is Extract<SuccessfulRunHandoffDecision, { kind: "skip" }> {
|
||||
return decision.kind === "skip" && SUCCESSFUL_RUN_HANDOFF_VALID_PATH_SKIP_REASONS.has(decision.reason);
|
||||
}
|
||||
|
||||
function metadataText(value: unknown, fallback = "unknown") {
|
||||
const text = typeof value === "string" ? value.trim() : value == null ? "" : String(value).trim();
|
||||
const resolved = text.length > 0 ? text : fallback;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,128 @@
|
|||
import { and, desc, eq, inArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { activityLog, agentWakeupRequests, heartbeatRuns } from "@paperclipai/db";
|
||||
import type { SuccessfulRunHandoffState } from "@paperclipai/shared";
|
||||
import { logActivity } from "./activity-log.js";
|
||||
|
||||
export const SUCCESSFUL_RUN_HANDOFF_LIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
|
||||
export const SUCCESSFUL_RUN_HANDOFF_LIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution", "claimed"] as const;
|
||||
|
||||
const heartbeatRunIssueId = sql<string>`coalesce(
|
||||
${heartbeatRuns.contextSnapshot} ->> 'issueId',
|
||||
${heartbeatRuns.contextSnapshot} ->> 'taskId'
|
||||
)`;
|
||||
|
||||
const wakeRequestIssueId = sql<string>`coalesce(
|
||||
${agentWakeupRequests.payload} ->> 'issueId',
|
||||
${agentWakeupRequests.payload} ->> 'taskId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'issueId',
|
||||
${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'taskId'
|
||||
)`;
|
||||
|
||||
export async function hydrateSuccessfulRunHandoffLiveness(
|
||||
dbOrTx: any,
|
||||
companyId: string,
|
||||
states: Map<string, SuccessfulRunHandoffState>,
|
||||
) {
|
||||
const requiredIssueIds = [...states.entries()]
|
||||
.filter(([, state]) => state.state === "required")
|
||||
.map(([issueId]) => issueId);
|
||||
if (requiredIssueIds.length === 0) return states;
|
||||
|
||||
const [activeRuns, activeWakes] = await Promise.all([
|
||||
dbOrTx
|
||||
.select({ id: heartbeatRuns.id, issueId: heartbeatRunIssueId })
|
||||
.from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
inArray(heartbeatRuns.status, [...SUCCESSFUL_RUN_HANDOFF_LIVE_RUN_STATUSES]),
|
||||
inArray(heartbeatRunIssueId, requiredIssueIds),
|
||||
)),
|
||||
dbOrTx
|
||||
.select({ issueId: wakeRequestIssueId })
|
||||
.from(agentWakeupRequests)
|
||||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
inArray(agentWakeupRequests.status, [...SUCCESSFUL_RUN_HANDOFF_LIVE_WAKE_STATUSES]),
|
||||
inArray(wakeRequestIssueId, requiredIssueIds),
|
||||
)),
|
||||
]);
|
||||
|
||||
const liveRunByIssueId = new Map<string, string>();
|
||||
for (const row of activeRuns as Array<{ id: string; issueId: string | null }>) {
|
||||
if (row.issueId && !liveRunByIssueId.has(row.issueId)) liveRunByIssueId.set(row.issueId, row.id);
|
||||
}
|
||||
const liveWakeIssueIds = new Set(
|
||||
(activeWakes as Array<{ issueId: string | null }>)
|
||||
.map((row) => row.issueId)
|
||||
.filter((issueId): issueId is string => Boolean(issueId)),
|
||||
);
|
||||
|
||||
for (const issueId of requiredIssueIds) {
|
||||
const state = states.get(issueId);
|
||||
if (!state) continue;
|
||||
const liveRunId = liveRunByIssueId.get(issueId);
|
||||
states.set(issueId, {
|
||||
...state,
|
||||
hasLiveContinuation: Boolean(liveRunId || liveWakeIssueIds.has(issueId)),
|
||||
...(liveRunId ? { liveRunId } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
return states;
|
||||
}
|
||||
|
||||
export async function resolveRequiredSuccessfulRunHandoffOnValidPath(
|
||||
db: Db,
|
||||
input: {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
issueIdentifier: string | null;
|
||||
agentId: string;
|
||||
runId: string;
|
||||
skipReason: string;
|
||||
},
|
||||
) {
|
||||
const latestHandoff = await db
|
||||
.select({ action: activityLog.action, runId: activityLog.runId, details: activityLog.details })
|
||||
.from(activityLog)
|
||||
.where(and(
|
||||
eq(activityLog.companyId, input.companyId),
|
||||
eq(activityLog.entityType, "issue"),
|
||||
eq(activityLog.entityId, input.issueId),
|
||||
inArray(activityLog.action, [
|
||||
"issue.successful_run_handoff_required",
|
||||
"issue.successful_run_handoff_resolved",
|
||||
"issue.successful_run_handoff_escalated",
|
||||
]),
|
||||
))
|
||||
.orderBy(desc(activityLog.createdAt), desc(activityLog.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (latestHandoff?.action !== "issue.successful_run_handoff_required") return false;
|
||||
|
||||
const details = latestHandoff.details && typeof latestHandoff.details === "object"
|
||||
? latestHandoff.details as Record<string, unknown>
|
||||
: {};
|
||||
const sourceRunId = [details.sourceRunId, details.source_run_id, details.resumeFromRunId]
|
||||
.find((value): value is string => typeof value === "string" && value.trim().length > 0)
|
||||
?.trim() ?? latestHandoff.runId;
|
||||
await logActivity(db, {
|
||||
companyId: input.companyId,
|
||||
actorType: "system",
|
||||
actorId: "heartbeat",
|
||||
agentId: input.agentId,
|
||||
runId: input.runId,
|
||||
action: "issue.successful_run_handoff_resolved",
|
||||
entityType: "issue",
|
||||
entityId: input.issueId,
|
||||
details: {
|
||||
label: "Successful run handoff continuation confirmed",
|
||||
sourceRunId,
|
||||
resolvedByRunId: input.runId,
|
||||
resolvedBySkipReason: input.skipReason,
|
||||
issue: { id: input.issueId, identifier: input.issueIdentifier },
|
||||
},
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
|
@ -127,6 +127,7 @@ describe("IssueBlockedNotice", () => {
|
|||
successfulRunHandoff={{
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "12345678-aaaa-bbbb-cccc-123456789abc",
|
||||
correctiveRunId: null,
|
||||
assigneeAgentId: "agent-1",
|
||||
|
|
@ -156,6 +157,7 @@ describe("IssueBlockedNotice", () => {
|
|||
successfulRunHandoff={{
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "12345678-aaaa-bbbb-cccc-123456789abc",
|
||||
correctiveRunId: null,
|
||||
assigneeAgentId: "agent-1",
|
||||
|
|
@ -190,6 +192,7 @@ describe("IssueBlockedNotice", () => {
|
|||
successfulRunHandoff={{
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "12345678-aaaa-bbbb-cccc-123456789abc",
|
||||
correctiveRunId: null,
|
||||
assigneeAgentId: "agent-1",
|
||||
|
|
|
|||
|
|
@ -454,6 +454,7 @@ describe("IssueChatThread system notice routing", () => {
|
|||
successfulRunHandoff: {
|
||||
state: "resolved",
|
||||
required: false,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "run-stale",
|
||||
correctiveRunId: "run-corrective",
|
||||
assigneeAgentId: "agent-codex",
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ export function successfulRunHandoffFromActivity(event: ActivityEvent): Successf
|
|||
return {
|
||||
state,
|
||||
required: state === "required",
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId:
|
||||
readString(details.sourceRunId)
|
||||
?? readString(details.source_run_id)
|
||||
|
|
|
|||
|
|
@ -866,6 +866,7 @@ function IssueThreadNoticeReview() {
|
|||
successfulRunHandoff={{
|
||||
state: "resolved",
|
||||
required: false,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "run-notice-source",
|
||||
correctiveRunId: "run-notice-corrective",
|
||||
assigneeAgentId: codexAgent.id,
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ function handoffIssue() {
|
|||
successfulRunHandoff: {
|
||||
state: "required",
|
||||
required: true,
|
||||
hasLiveContinuation: false,
|
||||
sourceRunId: "9cdba892-c7ca-4d93-8604-4843873b127c",
|
||||
correctiveRunId: "61fdb79b-8012-4676-ac71-2971830e126a",
|
||||
assigneeAgentId: "agent-codex",
|
||||
|
|
|
|||
Loading…
Reference in New Issue