Integrate current master through work-folder PR 13264
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
commit
0f6aa4555a
|
|
@ -1033,9 +1033,13 @@ Legacy finalization retries deferred input after adapter and lease cleanup.
|
|||
The scheduler also revisits bounded batches of stranded queues after restart
|
||||
or a late enqueue. Both use normal admission; an existing queued successor
|
||||
owns the next turn even before it acquires the task execution lock. A recovery
|
||||
hold or a plain operator Stop does not by itself authorize old input. The
|
||||
successor guard is scoped to the same agent so another agent's review
|
||||
participation keeps its independent recovery path.
|
||||
hold does not block an undelivered user message in a durable queue. The server
|
||||
validates the saved comment and its author, even if the queue began as a system
|
||||
wake. It can then start a fresh legacy conversation after proving the old
|
||||
process stopped. It preserves unknown action outcomes and does not replay
|
||||
comments already delivered to the failed run. A plain operator Stop still
|
||||
requires a new user action. The successor guard is scoped to the same agent so
|
||||
another agent's review participation keeps its independent recovery path.
|
||||
|
||||
An explicit queued-message Interrupt also grants one scoped cleanup retry for
|
||||
the stopped run. Old ephemeral leases whose cleanup predates provider stop
|
||||
|
|
@ -1045,6 +1049,11 @@ Delivery still requires the provider's verified stop receipt. Periodic queue
|
|||
retries do not gain extra cleanup attempts, and the queue displays the server's
|
||||
waiting reason while cleanup remains unresolved.
|
||||
|
||||
The legacy task recovery notice shows “Automatic recovery of this task stopped.” in
|
||||
a bordered container with Retry for a failed or timed-out run. A failed Retry
|
||||
shows its error in the same container. New user messages and saved undelivered
|
||||
messages pass normal admission independently of automatic recovery exhaustion.
|
||||
|
||||
### Operator identity and permission for manual dispatch
|
||||
|
||||
A legacy queued-message Interrupt is a new instruction from the user who clicks
|
||||
|
|
|
|||
|
|
@ -314,6 +314,75 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
|||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
|
||||
});
|
||||
|
||||
it.each(["user", "system", "manual_receipt_other_actor", "rejected_admission", "consumed_first", "consumed_last", "deleted_first", "agent_first", "cancelled_queued"])("delivers saved user messages on a %s queue after automatic recovery stopped, without another click", async (actorType) => {
|
||||
const seeded = await seedQueue();
|
||||
await db.update(agentWakeupRequests).set({ requestedByActorType: actorType === "user" ? "user" : "system" })
|
||||
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
if (actorType === "manual_receipt_other_actor") {
|
||||
const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
await db.update(agentWakeupRequests).set({ requestedByActorType: "user", requestedByActorId: "other-operator",
|
||||
payload: { ...saved.payload, manualUserWake: true },
|
||||
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
}
|
||||
await db.update(agents).set({ adapterType: "claude_local",
|
||||
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
||||
}).where(eq(agents.id, seeded.agentId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "failed",
|
||||
processPid: process.pid, errorCode: "process_lost", finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
||||
}).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
||||
await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId,
|
||||
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
|
||||
evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
||||
});
|
||||
// Leave the agent's only slot occupied on another task so dispatch stays queued.
|
||||
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
||||
status: "running", contextSnapshot: { issueId: randomUUID() },
|
||||
});
|
||||
await heartbeatService(db).resumeQueuedRuns();
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status)
|
||||
.toBe("deferred_issue_execution");
|
||||
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
const excludedIndex = actorType === "consumed_last" ? 1 : 0;
|
||||
const filtersInput = ["consumed_first", "consumed_last", "deleted_first", "agent_first"].includes(actorType);
|
||||
if (actorType.startsWith("consumed_")) await db.update(heartbeatRuns).set({
|
||||
contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[excludedIndex]] },
|
||||
}).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
if (actorType === "deleted_first") await db.update(issueComments).set({ deletedAt: new Date() })
|
||||
.where(eq(issueComments.id, seeded.commentIds[0]));
|
||||
if (actorType === "agent_first") await db.update(issueComments).set({
|
||||
authorType: "agent", authorUserId: null, authorAgentId: seeded.agentId,
|
||||
}).where(eq(issueComments.id, seeded.commentIds[0]));
|
||||
const expectedIds = seeded.commentIds.filter((_, index) => !filtersInput || index !== excludedIndex);
|
||||
if (actorType === "cancelled_queued") await db.insert(heartbeatRuns).values({
|
||||
companyId: seeded.companyId, agentId: seeded.agentId, status: "cancelled", runtimeMode: "legacy",
|
||||
errorCode: "agent_paused", createdAt: new Date(0), finishedAt: new Date(1),
|
||||
contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[0]] },
|
||||
});
|
||||
if (actorType === "rejected_admission") {
|
||||
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
||||
status: "cancelled", runtimeMode: "legacy", errorCode: "execution_reconciliation_required",
|
||||
contextSnapshot: { issueId: seeded.issueId }, finishedAt: new Date(),
|
||||
});
|
||||
}
|
||||
await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]);
|
||||
const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
expect(delivered.status).toBe("coalesced");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
|
||||
expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds,
|
||||
previousRunId: seeded.runId, forceFreshSession: true });
|
||||
if (actorType === "manual_receipt_other_actor") {
|
||||
const [dispatch] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, successor.wakeupRequestId!));
|
||||
expect(dispatch.requestedByActorId).toBe("queue-owner");
|
||||
expect(dispatch.payload?.manualUserWake).toBeUndefined();
|
||||
expect(successor.responsibleUserId).toBe("queue-owner");
|
||||
}
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(["rejected_admission", "cancelled_queued"].includes(actorType) ? 4 : 3);
|
||||
const [recovery] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
|
||||
expect(recovery.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown" });
|
||||
});
|
||||
|
||||
it("recovers a message deferred after legacy finalization released the task lock", async () => {
|
||||
const seeded = await seedQueue();
|
||||
await db.update(agents).set({ adapterType: "claude_local",
|
||||
|
|
|
|||
|
|
@ -118,7 +118,10 @@ export async function buildExecutionContinuation(input: {
|
|||
const triggerInteraction = interactions.find(
|
||||
(row) => row.id === input.context.interactionId,
|
||||
);
|
||||
const explicitContinuation = object(input.context.explicitUserContinuation);
|
||||
const explicitUserSource = string(explicitContinuation.previousRunId);
|
||||
const sourceRunId =
|
||||
explicitUserSource ??
|
||||
triggerInteraction?.sourceRunId ??
|
||||
string(input.context.retryOfRunId) ??
|
||||
string(input.context.previousRunId);
|
||||
|
|
@ -137,7 +140,7 @@ export async function buildExecutionContinuation(input: {
|
|||
)[0]
|
||||
: null;
|
||||
if (sourceRunId && !sourceRun)
|
||||
throw new Error("continuation_source_context_missing");
|
||||
throw new Error(explicitUserSource ? "continuation_user_authorization_missing" : "continuation_source_context_missing");
|
||||
const originCommentIds = [
|
||||
...new Set([
|
||||
...continuationOriginCommentIds(input.context),
|
||||
|
|
@ -259,8 +262,6 @@ export async function buildExecutionContinuation(input: {
|
|||
["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
|
||||
!(run.status === "cancelled" && run.errorCode === "execution_reconciliation_required"),
|
||||
);
|
||||
const explicitContinuation = object(input.context.explicitUserContinuation);
|
||||
const explicitUserSource = string(explicitContinuation.previousRunId);
|
||||
if (explicitUserSource) {
|
||||
const predecessor = priorRuns.find(run => run.id === explicitUserSource &&
|
||||
["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
|
||||
|
|
@ -289,6 +290,7 @@ export async function buildExecutionContinuation(input: {
|
|||
eq(agentWakeupRequests.status, "coalesced"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
|
||||
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
|
||||
)) : [];
|
||||
const authorization = continuationAuthorizations.find(value => failedRunId
|
||||
? value.failedRunId === failedRunId && retryWakes.some(wake =>
|
||||
|
|
|
|||
|
|
@ -180,6 +180,90 @@ const support = await getEmbeddedPostgresTestSupport();
|
|||
} }).where(eq(issueRecoveryActions.id, action.id));
|
||||
await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing");
|
||||
});
|
||||
it.each(["valid", "stale_retry_context", "wrong_actor", "wrong_run", "wrong_company", "discarded"])("verifies another author's queued Interrupt at execution setup: %s", async kind => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null })
|
||||
.where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") })
|
||||
.where(eq(issueComments.id, f.commentId));
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
||||
const queueId = randomUUID();
|
||||
const payload = { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] },
|
||||
queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } };
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "system", payload });
|
||||
await heartbeatService(db).resumeQueuedCommentInterrupt(f.companyId, queueId);
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
|
||||
expect(wake.status).toBe("coalesced");
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
||||
if (kind === "wrong_actor") await db.update(agentWakeupRequests).set({
|
||||
payload: { ...payload, queuedCommentInterrupt: { ...payload.queuedCommentInterrupt, actorId: "forged" } },
|
||||
}).where(eq(agentWakeupRequests.id, queueId));
|
||||
if (kind === "wrong_run") await db.update(agentWakeupRequests).set({ runId: f.sourceRunId })
|
||||
.where(eq(agentWakeupRequests.id, queueId));
|
||||
if (kind === "wrong_company") await db.update(agentWakeupRequests).set({ companyId: (await seed()).companyId })
|
||||
.where(eq(agentWakeupRequests.id, queueId));
|
||||
if (kind === "discarded") await db.update(issueComments).set({ deletedAt: new Date() })
|
||||
.where(eq(issueComments.id, f.commentId));
|
||||
const result = buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
agentId: f.agentId, runId: run.id, context: {
|
||||
...run.contextSnapshot, ...(kind === "stale_retry_context" ? { retryOfRunId: randomUUID() } : {}),
|
||||
}, summary: null, exposeLowTrustRaw: false });
|
||||
if (kind === "valid" || kind === "stale_retry_context") await expect(result).resolves.toMatchObject({ interruptedRunId: f.sourceRunId });
|
||||
else await expect(result).rejects.toThrow("continuation_user_authorization_missing");
|
||||
});
|
||||
|
||||
it.each(["valid", "wrong_actor", "consumed", "discarded", "operator_stop", "already_delivered", "earlier_delivered", "unstarted_cancelled", "unstarted_cancelled_metadata", "foreign_queue"])("validates automatic saved-message delivery: %s", async kind => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null,
|
||||
status: kind === "operator_stop" ? "cancelled" : "failed",
|
||||
contextSnapshot: { issueId: f.issueId, ...(kind === "already_delivered" ? { wakeCommentIds: [f.commentId] } : {}) },
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:00:00Z"),
|
||||
...(kind === "discarded" ? { deletedAt: new Date() } : {}),
|
||||
}).where(eq(issueComments.id, f.commentId));
|
||||
const queueId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_commented", status: kind === "consumed" ? "coalesced" : "deferred_issue_execution",
|
||||
requestedByActorType: "system", payload: { issueId: kind === "foreign_queue" ? randomUUID() : f.issueId,
|
||||
_paperclipWakeContext: { wakeCommentIds: [f.commentId] } },
|
||||
});
|
||||
if (kind.startsWith("unstarted_cancelled")) {
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "cancelled", runtimeMode: "legacy", errorCode: "agent_paused", finishedAt: new Date(),
|
||||
...(kind === "unstarted_cancelled_metadata" ? { nativeIssueId: f.issueId, processPid: 999999999, processGroupId: 999999999 } : {}),
|
||||
contextSnapshot: { issueId: f.issueId, wakeCommentIds: [f.commentId] },
|
||||
});
|
||||
}
|
||||
if (kind === "earlier_delivered") {
|
||||
const earlierId = randomUUID();
|
||||
await db.insert(issueComments).values({ id: earlierId, companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: f.actorId, body: "Already handled" });
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeCommentIds: [earlierId] } })
|
||||
.where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(agentWakeupRequests).set({ payload: { issueId: f.issueId,
|
||||
_paperclipWakeContext: { wakeCommentIds: [earlierId, f.commentId] } } })
|
||||
.where(eq(agentWakeupRequests.id, queueId));
|
||||
}
|
||||
const result = await db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
return admitExplicitNativeContinuation({ ...f, actorId: kind === "wrong_actor" ? "someone-else" : f.actorId,
|
||||
db: tx as unknown as typeof db, queuedCommentRequestId: queueId, dryRun: true });
|
||||
});
|
||||
if (kind === "valid" || kind.startsWith("unstarted_cancelled")) expect(result).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId });
|
||||
else expect(result).toBeNull();
|
||||
});
|
||||
|
||||
const admit = (f: Fixture, dryRun = false) => db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
const result = await admitExplicitNativeContinuation({ ...f, dryRun, db: tx as unknown as typeof db });
|
||||
|
|
|
|||
|
|
@ -25,6 +25,43 @@ function processStopped(pid: number): boolean {
|
|||
catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; }
|
||||
}
|
||||
|
||||
/** Validate the whole saved queue, preserving order and original authors.
|
||||
* Call again under the task lock before adopting IDs into a new run.
|
||||
*/
|
||||
export async function undeliveredLegacyUserCommentIds(
|
||||
db: Db, companyId: string, issueId: string, agentId: string, commentIds: string[],
|
||||
): Promise<string[]> {
|
||||
if (!commentIds.length) return [];
|
||||
const comments = await db.select({ id: issueComments.id }).from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
inArray(issueComments.id, commentIds), eq(issueComments.authorType, "user"),
|
||||
isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt),
|
||||
sql`nullif(trim(${issueComments.body}), '') is not null`,
|
||||
sql`nullif(trim(${issueComments.authorUserId}), '') is not null`,
|
||||
));
|
||||
const valid = new Set(comments.map(comment => comment.id));
|
||||
const previous = await db.select({ context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
||||
// A queued turn cancelled before dispatch has not consumed input, whatever
|
||||
// cancelled it (pause, stale assignment, or rejected admission). Reserved
|
||||
// native identity/process metadata is not proof that its prompt was sent.
|
||||
// Admission separately verifies process termination before a new turn.
|
||||
sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`,
|
||||
or(...commentIds.map(id => or(
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`,
|
||||
sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`,
|
||||
))),
|
||||
));
|
||||
for (const { context } of previous) {
|
||||
if (typeof context?.wakeCommentId === "string") valid.delete(context.wakeCommentId);
|
||||
if (Array.isArray(context?.wakeCommentIds)) {
|
||||
for (const id of context.wakeCommentIds) if (typeof id === "string") valid.delete(id);
|
||||
}
|
||||
}
|
||||
return commentIds.filter(id => valid.has(id));
|
||||
}
|
||||
|
||||
/** Called under the issue lock, in the transaction that creates the new turn.
|
||||
* A user request authorizes a new conversation, not replay of the failed run.
|
||||
* Unknown action outcomes and all prior records stay intact.
|
||||
|
|
@ -36,6 +73,8 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
failedRunId?: string | null;
|
||||
/** Server-recorded board intent to send an existing legacy message queue. */
|
||||
queuedCommentInterruptId?: string;
|
||||
/** Internal delivery of an unconsumed, user-authored legacy queue entry. */
|
||||
queuedCommentRequestId?: string;
|
||||
dryRun?: boolean;
|
||||
resumingSavedMessage?: boolean;
|
||||
onBlocked?: (reason: string, message: string) => void;
|
||||
|
|
@ -61,6 +100,21 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
const queuedInterrupt = Boolean(interruptQueue && commentId &&
|
||||
queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId));
|
||||
if (input.queuedCommentInterruptId && !queuedInterrupt) return null;
|
||||
const [savedQueue] = input.queuedCommentRequestId ? await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, input.queuedCommentRequestId),
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
)) : [];
|
||||
const queuedRequest = Boolean(savedQueue && commentId &&
|
||||
!savedQueue.idempotencyKey?.startsWith("chat-inbound:") &&
|
||||
queuedCommentIdsFromWakePayload(savedQueue.payload).includes(commentId));
|
||||
if (input.queuedCommentRequestId && !queuedRequest) return null;
|
||||
if (queuedRequest) {
|
||||
const ids = queuedCommentIdsFromWakePayload(savedQueue!.payload);
|
||||
const undelivered = await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, ids);
|
||||
if (undelivered.length !== ids.length) return null;
|
||||
}
|
||||
const [comment] = retry ? [] : await db.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"),
|
||||
|
|
@ -104,7 +158,7 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
|
||||
(run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId ||
|
||||
!run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
|
||||
if (!queuedInterrupt && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
|
||||
if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
|
||||
if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
|
||||
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
|
||||
run.errorCode === "execution_reconciliation_required" &&
|
||||
|
|
@ -112,7 +166,12 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
const legacyUserTurn = run.runtimeMode === "legacy" &&
|
||||
action.cause === "legacy_execution_requires_reconciliation" &&
|
||||
isConversationAdapter(agent.adapterType);
|
||||
if (queuedInterrupt && !legacyUserTurn) return null;
|
||||
if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission) return null;
|
||||
// Saved input is a request for a new turn, never permission to undo an
|
||||
// operator Stop or redeliver a message already consumed by this run.
|
||||
if (queuedRequest && !queuedInterrupt && ((run.status === "cancelled" && !unusedAdmission) ||
|
||||
run.contextSnapshot?.wakeCommentId === commentId ||
|
||||
(Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null;
|
||||
if (legacyUserTurn) {
|
||||
const historicalAdapter = await historicalAdapterType(db, run);
|
||||
// A settings change never converts a known process/webhook execution into
|
||||
|
|
@ -181,7 +240,8 @@ export async function admitExplicitNativeContinuation(input: {
|
|||
summary: null, exposeLowTrustRaw: false });
|
||||
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
|
||||
const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}),
|
||||
...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), runId: input.successorRunId,
|
||||
...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}),
|
||||
...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId,
|
||||
previousRunId: previous.id, recordedAt: new Date().toISOString() };
|
||||
for (const runId of cancelledStartupIds) {
|
||||
await db.update(nativeRunFinalizations).set({
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLea
|
|||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { admitExplicitNativeContinuation, undeliveredLegacyUserCommentIds } from "./explicit-native-continuation.js";
|
||||
import { connectionIntentService } from "./connection-intents.js";
|
||||
import { prepareManagedAiRuntime, assertManagedAiProjectAuth, stripAiAuthBindings, AI_AUTH_ENV_KEYS } from "./ai-connection-runtime.js";
|
||||
import { aiConnectionBindingSchema } from "@paperclipai/shared";
|
||||
|
|
@ -3538,6 +3538,8 @@ interface WakeupOptions {
|
|||
manualUserWake?: boolean;
|
||||
/** Internal resume of a queue with persisted board interruption intent. */
|
||||
queuedCommentInterruptId?: string;
|
||||
/** Internal delivery of an existing undelivered user comment. */
|
||||
queuedCommentRequestId?: string;
|
||||
/** Exact failed run selected by an authenticated board Retry request. */
|
||||
failedRunId?: string | null;
|
||||
durableChatRequest?: DurableChatWakeupRequest;
|
||||
|
|
@ -10202,16 +10204,34 @@ export function heartbeatService(
|
|||
}
|
||||
|
||||
async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) {
|
||||
return resumeSavedLegacyComments(companyId, queueId, true, opts);
|
||||
}
|
||||
|
||||
async function resumeSavedLegacyComments(companyId: string, queueId: string, interrupted = false, opts?: { retryCleanup?: boolean }) {
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
));
|
||||
if (!wake) return;
|
||||
const payload = parseObject(wake.payload);
|
||||
const actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
|
||||
const commentIds = queuedCommentIdsFromWakePayload(payload);
|
||||
let actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
|
||||
let commentIds = queuedCommentIdsFromWakePayload(payload);
|
||||
const issueId = readNonEmptyString(payload.issueId);
|
||||
if (!actorId || !issueId || !commentIds.length) return;
|
||||
if (!issueId || !commentIds.length || wake.idempotencyKey?.startsWith("chat-inbound:")) return;
|
||||
if (!interrupted) {
|
||||
commentIds = await undeliveredLegacyUserCommentIds(db, companyId, issueId, wake.agentId, commentIds);
|
||||
if (!commentIds.length) return;
|
||||
// The queue itself may have begun as a system wake. The saved human
|
||||
// comment, not that wake's origin or mutable caller payload, is authority.
|
||||
const [comment] = await db.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
eq(issueComments.id, commentIds[commentIds.length - 1]!), eq(issueComments.authorType, "user"),
|
||||
isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt),
|
||||
)).orderBy(desc(issueComments.createdAt)).limit(1);
|
||||
if (!comment?.body.trim() || !comment.authorUserId) return;
|
||||
actorId = comment.authorUserId;
|
||||
}
|
||||
if (!actorId) return;
|
||||
const agent = await getAgent(wake.agentId);
|
||||
if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return;
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
|
|
@ -10220,7 +10240,7 @@ export function heartbeatService(
|
|||
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
||||
)).limit(1);
|
||||
if (active) return;
|
||||
if (opts?.retryCleanup) {
|
||||
if (interrupted && opts?.retryCleanup) {
|
||||
// Only the HTTP click grants an extra cleanup attempt. Periodic retries
|
||||
// reuse the intent to deliver, never a fresh provider teardown budget.
|
||||
const sourceRun = await db.transaction(async tx => {
|
||||
|
|
@ -10259,6 +10279,9 @@ export function heartbeatService(
|
|||
)).for("update");
|
||||
for (const lease of historical) {
|
||||
if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId || hasRemoteTerminationReceipt(lease)) continue;
|
||||
// Provider resource IDs identify physical sandboxes. A lease in any
|
||||
// company can still own this resource; never destroy it on behalf of
|
||||
// this company. This existence-only guard exposes no foreign data.
|
||||
const [otherOwner] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider),
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
|
||||
|
|
@ -10274,7 +10297,7 @@ export function heartbeatService(
|
|||
companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt",
|
||||
} });
|
||||
}
|
||||
const deliveryPayload = { ...payload };
|
||||
const deliveryPayload = withQueuedCommentIdsInWakePayload(payload, commentIds);
|
||||
delete deliveryPayload.queuedCommentInterrupt;
|
||||
await enqueueWakeup(wake.agentId, {
|
||||
source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
||||
|
|
@ -10282,9 +10305,9 @@ export function heartbeatService(
|
|||
issueId, triggeredBy: "board", actorId, responsibleUserId: actorId,
|
||||
}, commentIds),
|
||||
requestedByActorType: "user", requestedByActorId: actorId,
|
||||
queuedCommentInterruptId: queueId,
|
||||
...(interrupted ? { queuedCommentInterruptId: queueId } : { queuedCommentRequestId: queueId }),
|
||||
issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] },
|
||||
idempotencyKey: `queued-comment-interrupt:${queueId}`,
|
||||
idempotencyKey: `queued-comment-${interrupted ? "interrupt" : "delivery"}:${queueId}`,
|
||||
}, queueId);
|
||||
}
|
||||
|
||||
|
|
@ -18941,6 +18964,13 @@ export function heartbeatService(
|
|||
eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
));
|
||||
if (!latest || latest.runtimeMode !== "legacy" || !isHeartbeatRunTerminalStatus(latest.status)) continue;
|
||||
const cancelledAdmission = latest.status === "cancelled" && !latest.startedAt &&
|
||||
latest.errorCode === "execution_reconciliation_required";
|
||||
if ((latest.status !== "cancelled" || cancelledAdmission) && await getExecutionBlocker(db, wake.companyId, String(wake.payload?.issueId))) {
|
||||
await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => {
|
||||
logger.warn({ err, queueId: wake.id }, "failed to deliver saved legacy comment after recovery stopped");
|
||||
});
|
||||
}
|
||||
await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => {
|
||||
logger.warn({ err, queueId: wake.id }, "failed to promote stranded legacy comments");
|
||||
});
|
||||
|
|
@ -26252,12 +26282,13 @@ export function heartbeatService(
|
|||
eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId),
|
||||
eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
// A user message can join a queue originally created by a
|
||||
// system wake. The recorded board click supplies fresh authority.
|
||||
opts.queuedCommentInterruptId === executionWaitRequestId
|
||||
// system wake. Admission validates the saved user comment or board click.
|
||||
(opts.queuedCommentInterruptId ?? opts.queuedCommentRequestId) === executionWaitRequestId
|
||||
? undefined : eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
opts.queuedCommentInterruptId === executionWaitRequestId
|
||||
? sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}`
|
||||
: eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
|
||||
: opts.queuedCommentRequestId === executionWaitRequestId ? undefined
|
||||
: eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
));
|
||||
// The issue lock serializes cleanup callbacks and periodic workers.
|
||||
|
|
@ -26274,7 +26305,16 @@ export function heartbeatService(
|
|||
sql`length(trim(${issueComments.body})) > 0`,
|
||||
));
|
||||
if (!comment) return { kind: "deferred" as const };
|
||||
if (!opts.queuedCommentInterruptId && pending.payload?.manualUserWake === true) {
|
||||
if (opts.queuedCommentRequestId) {
|
||||
const ids = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
|
||||
agent.companyId, issueId, agentId, queuedCommentIdsFromWakePayload(pending.payload));
|
||||
if (!ids.includes(wakeCommentId)) return { kind: "deferred" as const };
|
||||
pending.payload = withQueuedCommentIdsInWakePayload(parseObject(pending.payload), ids);
|
||||
await tx.update(agentWakeupRequests).set({ payload: pending.payload }).where(and(
|
||||
eq(agentWakeupRequests.id, pending.id), eq(agentWakeupRequests.companyId, agent.companyId),
|
||||
));
|
||||
}
|
||||
if (!opts.queuedCommentInterruptId && !opts.queuedCommentRequestId && pending.payload?.manualUserWake === true) {
|
||||
// A persisted manual wake keeps its actor when an execution wait
|
||||
// resumes. The locked receipt above has revalidated that actor.
|
||||
payload = { ...payload, manualUserWake: true };
|
||||
|
|
@ -26284,6 +26324,8 @@ export function heartbeatService(
|
|||
// The locked board receipt supplies execution authority even when
|
||||
// another user authored the messages. Dispatch revalidates the receipt.
|
||||
operatorResponsibleUserId = opts.requestedByActorId!;
|
||||
}
|
||||
if (opts.queuedCommentInterruptId || opts.queuedCommentRequestId) {
|
||||
// Edits/discards between the click and dispatch remain authoritative.
|
||||
Object.assign(enrichedContextSnapshot, withQueuedCommentIdsInRunContext(
|
||||
enrichedContextSnapshot, queuedCommentIdsFromWakePayload(pending.payload),
|
||||
|
|
@ -26669,6 +26711,7 @@ export function heartbeatService(
|
|||
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
|
||||
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
|
||||
queuedCommentInterruptId: opts.queuedCommentInterruptId,
|
||||
queuedCommentRequestId: opts.queuedCommentRequestId,
|
||||
dryRun: true,
|
||||
onBlocked: (reason, message) => { continuationWait = { reason, message }; },
|
||||
}))) return deferBlockedExecution(executionBlocker);
|
||||
|
|
@ -27438,6 +27481,7 @@ export function heartbeatService(
|
|||
continuationWait = { reason, message };
|
||||
},
|
||||
queuedCommentInterruptId: opts.queuedCommentInterruptId,
|
||||
queuedCommentRequestId: opts.queuedCommentRequestId,
|
||||
});
|
||||
// Recovery can change while earlier admission gates await I/O. Use
|
||||
// the current blocker, not the snapshot from the start of admission.
|
||||
|
|
@ -27488,7 +27532,7 @@ export function heartbeatService(
|
|||
.orderBy(asc(agentWakeupRequests.requestedAt))
|
||||
: [];
|
||||
const adoptedComments = pendingComments.filter((wake) => {
|
||||
if (wake.id === opts.queuedCommentInterruptId) return true;
|
||||
if (wake.id === opts.queuedCommentInterruptId || wake.id === opts.queuedCommentRequestId) return true;
|
||||
const deferredPayload = parseObject(wake.payload);
|
||||
const deferredContext = parseObject(
|
||||
deferredPayload[DEFERRED_WAKE_CONTEXT_KEY],
|
||||
|
|
@ -27505,7 +27549,7 @@ export function heartbeatService(
|
|||
queuedCommentIdsFromWakePayload(wake.payload).length > 0
|
||||
);
|
||||
});
|
||||
const adoptedCommentIds = [
|
||||
let adoptedCommentIds = [
|
||||
...new Set([
|
||||
...adoptedComments.flatMap((wake) =>
|
||||
queuedCommentIdsFromWakePayload(wake.payload),
|
||||
|
|
@ -27513,6 +27557,10 @@ export function heartbeatService(
|
|||
...queuedCommentIdsFromRunContext(enrichedContextSnapshot),
|
||||
]),
|
||||
];
|
||||
if (opts.queuedCommentRequestId) {
|
||||
adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
|
||||
agent.companyId, issueId, agentId, adoptedCommentIds);
|
||||
}
|
||||
const newRun = await tx
|
||||
.insert(heartbeatRuns)
|
||||
.values({
|
||||
|
|
|
|||
|
|
@ -3435,10 +3435,10 @@ test.describe("iMessage Photon setup and management", () => {
|
|||
await page
|
||||
.getByRole("button", { name: "Connect selected number" })
|
||||
.click();
|
||||
expect(mock.configuredCredentialKeys).toEqual(["projectSecret"]);
|
||||
await expect(
|
||||
page.getByRole("heading", { name: "Try Maya in iMessage Photon" }),
|
||||
).toBeVisible();
|
||||
expect(mock.configuredCredentialKeys).toEqual(["projectSecret"]);
|
||||
await expect(
|
||||
page.getByRole("button", { name: "Copy +15555550100" }),
|
||||
).toBeVisible();
|
||||
|
|
|
|||
|
|
@ -4,14 +4,14 @@ import os from "node:os";
|
|||
import path from "node:path";
|
||||
import { test, expect, type APIResponse } from "@playwright/test";
|
||||
import { and, eq } from "../../server/node_modules/drizzle-orm/index.js";
|
||||
import { createDb, closeRegisteredClients, heartbeatRuns, issueRecoveryActions, issues } from "../../packages/db/src/index.ts";
|
||||
import { createDb, closeRegisteredClients, heartbeatRuns, issueRecoveryActions, issues, issueComments, agentWakeupRequests, authUsers, companyMemberships } from "../../packages/db/src/index.ts";
|
||||
|
||||
async function json(response: APIResponse) {
|
||||
expect(response.ok(), `${response.status()} ${await response.text()}`).toBe(true);
|
||||
return response.json();
|
||||
}
|
||||
|
||||
for (const action of ["task_retry", "inbox_retry", "message"] as const) {
|
||||
for (const action of ["task_retry", "inbox_retry", "message", "queued_interrupt", "automatic_message"] as const) {
|
||||
test(`legacy startup hold: ${action} reaches a new agent response`, async ({ page, request }) => {
|
||||
test.setTimeout(120_000);
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "legacy-recovery-browser-"));
|
||||
|
|
@ -38,7 +38,7 @@ for (const action of ["task_retry", "inbox_retry", "message"] as const) {
|
|||
// Seed the historical incident, then exercise all recovery through the UI.
|
||||
// No adapter.invoke or new dispatch identity exists on this pre-upgrade run.
|
||||
await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId: company.id, agentId: agent.id,
|
||||
status: "failed", runtimeMode: "legacy", processPid: 999999999,
|
||||
status: "failed", runtimeMode: "legacy", processPid: action === "queued_interrupt" ? process.pid : 999999999,
|
||||
responsibleUserId: issue.responsibleUserId, errorCode: "process_lost", error: "Server restarted during startup",
|
||||
startedAt: new Date(Date.now() - 10_000), finishedAt: new Date(Date.now() - 5_000),
|
||||
contextSnapshot: { issueId: issue.id },
|
||||
|
|
@ -49,9 +49,48 @@ for (const action of ["task_retry", "inbox_retry", "message"] as const) {
|
|||
evidence: { runId: sourceRunId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
||||
});
|
||||
await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issue.id));
|
||||
if (action === "queued_interrupt") {
|
||||
await db.insert(authUsers).values({ id: "original-board", name: "Original author", email: "original-author@example.test",
|
||||
createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
|
||||
await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user",
|
||||
principalId: "original-board", membershipRole: "operator", status: "active" });
|
||||
}
|
||||
if (action === "queued_interrupt" || action === "automatic_message") {
|
||||
const commentId = randomUUID();
|
||||
// Reproduce a real user comment saved while the failed run was active,
|
||||
// including queues originally created by a system wake.
|
||||
await db.insert(issueComments).values({ id: commentId, companyId: company.id, issueId: issue.id,
|
||||
authorType: "user", authorUserId: action === "queued_interrupt" ? "original-board" : "local-board", body: "Approved",
|
||||
createdAt: new Date(Date.now() - 8_000),
|
||||
});
|
||||
await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id,
|
||||
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "system", payload: { issueId: issue.id, commentId,
|
||||
_paperclipWakeContext: { issueId: issue.id, wakeCommentIds: [commentId], wakeCommentId: commentId } },
|
||||
});
|
||||
}
|
||||
const taskUrl = `/${company.issuePrefix}/issues/${issue.identifier}`;
|
||||
await page.goto(action === "inbox_retry" ? `/${company.issuePrefix}/inbox/all` : taskUrl);
|
||||
if (action === "message") {
|
||||
if (action === "task_retry") {
|
||||
const notice = page.getByRole("status", { name: "Task recovery" });
|
||||
await expect(notice).toHaveText("Automatic recovery of this task stopped.Retry");
|
||||
await expect(notice.getByRole("link")).toHaveCount(0);
|
||||
const presentation = await notice.evaluate(element => {
|
||||
const style = getComputedStyle(element);
|
||||
return { border: style.borderTopWidth, background: style.backgroundColor };
|
||||
});
|
||||
expect(parseFloat(presentation.border)).toBeGreaterThan(0);
|
||||
expect(presentation.background).not.toBe("rgba(0, 0, 0, 0)");
|
||||
await test.info().attach("recovery-notice", { body: await notice.screenshot(), contentType: "image/png" });
|
||||
}
|
||||
if (action === "queued_interrupt") {
|
||||
const interrupt = page.getByRole("button", { name: "Interrupt", exact: true });
|
||||
await expect(interrupt).toBeEnabled();
|
||||
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, sourceRunId));
|
||||
await interrupt.click();
|
||||
} else if (action === "automatic_message") {
|
||||
// No Retry, duplicate message, or status change: the saved input runs.
|
||||
} else if (action === "message") {
|
||||
await page.getByRole("textbox", { name: "editable markdown" }).fill("Please continue the pending follow-up.");
|
||||
await page.getByRole("button", { name: "Send", exact: true }).click();
|
||||
} else {
|
||||
|
|
@ -59,12 +98,18 @@ for (const action of ["task_retry", "inbox_retry", "message"] as const) {
|
|||
if (action === "inbox_retry") await page.goto(taskUrl);
|
||||
}
|
||||
await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible({ timeout: 45_000 });
|
||||
await expect(page.getByText("Work cannot start.", { exact: false })).toHaveCount(0);
|
||||
await expect(page.getByRole("status", { name: "Task recovery" })).toHaveCount(0);
|
||||
const completed = await json(await request.get(`/api/issues/${issue.id}`));
|
||||
expect(completed).toMatchObject({ status: "done", executionBlocker: null });
|
||||
const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, company.id), eq(heartbeatRuns.agentId, agent.id)));
|
||||
expect(runs.filter(run => run.id !== sourceRunId)).toHaveLength(1);
|
||||
expect(runs.find(run => run.id === sourceRunId)).toMatchObject({ status: "failed", resultJson: null });
|
||||
const prompts = await readFile(path.join(root, "prompts"), "utf8");
|
||||
if (action === "queued_interrupt" || action === "automatic_message") {
|
||||
expect(prompts).toContain("Approved");
|
||||
await expect(page.getByRole("button", { name: "Interrupt", exact: true })).toHaveCount(0);
|
||||
}
|
||||
if (action === "message") expect(prompts).toContain("Please continue the pending follow-up.");
|
||||
await page.reload();
|
||||
await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible();
|
||||
} finally {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
// @vitest-environment jsdom
|
||||
import { act } from "react";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { ExecutionBlockerNotice } from "./ExecutionBlockerNotice";
|
||||
import { agentsApi } from "../api/agents";
|
||||
import { activityApi } from "../api/activity";
|
||||
vi.mock("../api/agents", () => ({ agentsApi: { retryFailedRun: vi.fn() } }));
|
||||
vi.mock("../api/activity", () => ({ activityApi: { runsForIssue: vi.fn() } }));
|
||||
|
||||
describe("stopped task recovery notice", () => {
|
||||
let root: Root;
|
||||
let container: HTMLDivElement;
|
||||
let client: QueryClient;
|
||||
const onRetried = vi.fn();
|
||||
beforeEach(async () => {
|
||||
vi.clearAllMocks();
|
||||
container = document.createElement("div"); document.body.append(container);
|
||||
root = createRoot(container);
|
||||
client = new QueryClient({ defaultOptions: { queries: { retry: false }, mutations: { retry: false } } });
|
||||
vi.mocked(activityApi.runsForIssue).mockResolvedValue([{ runId: "failed-run", agentId: "agent", status: "failed" }] as never);
|
||||
await act(async () => root.render(<QueryClientProvider client={client}>
|
||||
<ExecutionBlockerNotice companyId="company" issueId="task" onRetried={onRetried} blocker={{
|
||||
recoveryActionId: "recovery", runId: "failed-run", agentId: "agent", cause: "legacy_execution_requires_reconciliation",
|
||||
nextAction: "Automatic recovery stopped. Recorded work is preserved; actions with unverified outcomes will not be repeated.",
|
||||
}} />
|
||||
</QueryClientProvider>));
|
||||
await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); });
|
||||
});
|
||||
afterEach(async () => { await act(async () => root.unmount()); client.clear(); container.remove(); });
|
||||
it("shows only the requested sentence and Retry, inside a distinct recovery container", () => {
|
||||
const notice = container.querySelector('[role="status"][aria-label="Task recovery"]')!;
|
||||
expect(notice.textContent).toBe("Automatic recovery of this task stopped.Retry");
|
||||
expect(notice.classList.contains("border")).toBe(true);
|
||||
expect(notice.classList.contains("bg-muted")).toBe(true);
|
||||
expect(notice.querySelector("a")).toBeNull();
|
||||
});
|
||||
it("keeps the required next action for other reconciliation causes", async () => {
|
||||
await act(async () => root.render(<QueryClientProvider client={client}>
|
||||
<ExecutionBlockerNotice companyId="company" issueId="task" onRetried={onRetried} blocker={{
|
||||
recoveryActionId: "recovery", runId: "failed-run", agentId: "agent", cause: "action_outcome_unknown",
|
||||
nextAction: "Verify the external action outcome before continuing.",
|
||||
}} />
|
||||
</QueryClientProvider>));
|
||||
expect(container.textContent).toContain("Verify the external action outcome before continuing.");
|
||||
expect(container.textContent).not.toContain("Automatic recovery of this task stopped.");
|
||||
});
|
||||
it("retries the exact failed run and refreshes the task", async () => {
|
||||
vi.mocked(agentsApi.retryFailedRun).mockResolvedValue({} as never);
|
||||
await act(async () => container.querySelector<HTMLButtonElement>("button")!.click());
|
||||
await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); });
|
||||
expect(agentsApi.retryFailedRun).toHaveBeenCalledWith("agent", "failed-run", "company");
|
||||
expect(onRetried).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("shows a failed Retry in the same container and allows another attempt", async () => {
|
||||
vi.mocked(agentsApi.retryFailedRun).mockRejectedValue(new Error("Environment cleanup is still running."));
|
||||
await act(async () => container.querySelector<HTMLButtonElement>("button")!.click());
|
||||
await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); });
|
||||
expect(container.querySelector('[role="alert"]')?.textContent).toBe("Environment cleanup is still running.");
|
||||
expect(container.querySelector<HTMLButtonElement>("button")!.disabled).toBe(false);
|
||||
expect(onRetried).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
|
@ -1,5 +1,4 @@
|
|||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import { Link } from "react-router-dom";
|
||||
import type { ExecutionBlocker } from "@paperclipai/shared";
|
||||
import { agentsApi } from "../api/agents";
|
||||
import { activityApi } from "../api/activity";
|
||||
|
|
@ -30,18 +29,15 @@ export function ExecutionBlockerNotice({ companyId, issueId, blocker, onRetried
|
|||
},
|
||||
});
|
||||
return (
|
||||
<div role="status" className="px-(--sz-execution-blocker-inline) py-(--sz-execution-blocker-block) text-sm text-muted-foreground">
|
||||
<span>Work cannot start. {blocker.nextAction}</span>{" "}
|
||||
<div role="status" aria-label="Task recovery" className="mx-(--sz-execution-blocker-inline) my-(--sz-execution-blocker-block) flex flex-wrap items-center justify-between execution-blocker-notice border border-border bg-muted text-foreground">
|
||||
<span>{blocker.cause === "legacy_execution_requires_reconciliation" ? "Automatic recovery of this task stopped." : blocker.nextAction}</span>
|
||||
{failedRun && (
|
||||
<Button variant="outline" size="sm" disabled={retry.isPending} onClick={() => retry.mutate()}>
|
||||
{retry.isPending ? "Retrying…" : "Retry"}
|
||||
</Button>
|
||||
)}{" "}
|
||||
{blocker.runId && blocker.agentId && (
|
||||
<Link className="underline" to={`/agents/${blocker.agentId}/runs/${blocker.runId}`}>View stopped run</Link>
|
||||
)}
|
||||
{retry.isError && (
|
||||
<p role="alert" className="text-destructive">{retry.error.message}</p>
|
||||
<p role="alert" className="w-full text-destructive">{retry.error.message}</p>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
|
|
|
|||
|
|
@ -2984,6 +2984,15 @@ span.paperclip-mention-chip[data-mention-kind="external-object"] {
|
|||
animation: none;
|
||||
}
|
||||
|
||||
/* Task recovery panel uses the shared spacing, radius, and type scales. */
|
||||
.execution-blocker-notice {
|
||||
gap: calc(var(--spacing) * 2);
|
||||
padding: calc(var(--spacing) * 2) calc(var(--spacing) * 3);
|
||||
border-radius: var(--radius-md);
|
||||
font-size: var(--text-sm);
|
||||
line-height: var(--text-sm--line-height);
|
||||
}
|
||||
|
||||
/* The runner keeps one fixed-height activity row between commentary updates. */
|
||||
@keyframes runner-activity-roll-in {
|
||||
from { transform: translateY(100%); opacity: 0; }
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
import type { Meta, StoryObj } from "@storybook/react-vite";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { ExecutionBlockerNotice } from "@/components/ExecutionBlockerNotice";
|
||||
import { queryKeys } from "@/lib/queryKeys";
|
||||
|
||||
const client = new QueryClient({ defaultOptions: { queries: { staleTime: Infinity } } });
|
||||
client.setQueryData(queryKeys.issues.runs("storybook-recovery"), [
|
||||
{ runId: "stopped-run", agentId: "agent", status: "failed" },
|
||||
]);
|
||||
const meta = {
|
||||
title: "Task chat/Recovery notice",
|
||||
component: ExecutionBlockerNotice,
|
||||
decorators: [Story => <QueryClientProvider client={client}><div className="max-w-2xl"><Story /></div></QueryClientProvider>],
|
||||
args: {
|
||||
companyId: "storybook-company", issueId: "storybook-recovery", onRetried: () => {},
|
||||
blocker: {
|
||||
recoveryActionId: "recovery", runId: "stopped-run", agentId: "agent",
|
||||
cause: "legacy_execution_requires_reconciliation",
|
||||
nextAction: "Automatic recovery stopped. Recorded work is preserved; actions with unverified outcomes will not be repeated.",
|
||||
},
|
||||
},
|
||||
} satisfies Meta<typeof ExecutionBlockerNotice>;
|
||||
export default meta;
|
||||
export const Stopped: StoryObj<typeof meta> = {};
|
||||
Loading…
Reference in New Issue