RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: recover legacy Daytona startup failures from task and inbox (#13272) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Legacy conversation adapters can run in Daytona sandboxes.
> - A server restart during provisioning can occur before the invocation
event exists.
> - Recovery then lacks the old adapter identity and leaves a hold that
ordinary user retries cannot clear.
> - A remote launch can also fail when its host relay looks for Node in
the sandbox PATH.
> - This pull request records the adapter at claim time and restores
explicit user continuation after verified cleanup.
> - Users can recover from the task or inbox while the failed run and
uncertain action history remain intact.

## Linked Issues or Issue Description

Refs #13237, #13239, #13254. Those changes cover recorded conversation
runs, native user continuation, and explicit remote Stop. This change
covers legacy failure before `adapter.invoke` and exact task/inbox
Retry.

Refs #9771 for overlapping generated-command quoting. This change also
supplies the absolute host Node executable. Refs #13163 and #13264 for
the separate native restart and retained-workspace work.

**What happened?**
A legacy Daytona run interrupted during provisioning became
`process_lost` without an invocation event. Recovery preserved an
execution hold, and Retry or a new task reply could not resume it.
Cleanup could also run before the Daytona plugin was ready. On a macOS
host, a subsequent ACP relay launch failed with `env: node: No such file
or directory` because the remote launch environment did not contain the
host Node path.

**Expected behavior**
An interrupted conversation can continue after its previous execution
stops. Explicit Retry and new user replies should start a fresh turn
with the task history. Cleanup failures must remain visible and
recoverable. The host relay must use the host Node executable.

**Steps to reproduce**
1. Use a legacy Claude adapter with a Daytona environment.
2. Interrupt the server after it acquires the sandbox lease and before
it records `adapter.invoke`.
3. Restart and inspect the task hold.
4. Retry from the task or inbox, or send a new task reply.
5. Confirm the old sandbox has stopped and one new response arrives.

**Paperclip version or commit**
Reproduced from master at `3bafac12f796fbea02e609e1074a9639f872e9c4`.
The branch is rebased on `51b0e01ea`, including #13261 and #13270.

**Deployment mode**
Built from source on macOS with a real Daytona sandbox and the legacy
Claude ACP adapter.

## What Changed

- Count new browser specs with the scheduler's median duration in the
shard-balance check. This fixes a false policy failure after new specs
arrive from both branches. The balance threshold is unchanged.

- Persist server-owned adapter identity in the queued-to-running claim
before provisioning starts.
- Wait for provider plugin startup before restart cleanup. Keep failed
cleanup leases as active ownership blockers.
- Admit exact board retries and new user comments after verified
termination. Retain the old run, task history, approvals, and unknown
action outcomes.
- Adopt repeated Retry requests. Permit one scoped cleanup attempt per
explicit user Retry after the automatic limit, with an activity record.
A later user Retry can recover after a transient provider failure;
automatic attempts remain capped.
- Resume replies deferred during cleanup, including historical legacy
startup failures.
- Launch the host ACP relay through the absolute host Node executable.
- Add a task-level Retry button and return actionable blockers when
retry admission is refused.
- Add database regressions and three browser recovery journeys. Exclude
installed third-party dependency skills from the shipped-skill audit.

## Verification

- Current head: `d23c84181`, rebased on `51b0e01ea`. Conflict resolution
retains the saved-message recovery, local stop receipts, and wait
reasons from #13270 alongside exact legacy Retry support.
- Real Daytona: interrupted the server after lease acquisition and
before adapter invocation. Restart cleanup confirmed provider
termination. Task Retry cleared a seeded historical hold and a real
Claude agent returned `Recovery verified.` in the task. Removed the
disposable sandbox and environment after testing.
- All three browser recovery journeys passed again after the final
rebase. Task Retry, Inbox Retry, and a new reply each produced one fresh
successor, completed the task, preserved the failed run, and retained
the answer after reload.
- All 29 e2e/server shard-partition tests passed. The balance check now
uses the scheduler's median fallback for unmeasured specs, with the same
balance threshold.
- Server typecheck passed after rebuilding the generated runner
dependencies. The combined recovery/route run passed 136 of 137 tests.
Its remaining route test timed out during the first cold module import
at its explicit 10-second limit; an isolated rerun reproduced that
timeout and passed the other 51 route cases. The complete CI suite
passed on this head. The same route file passed all 52 cases in CI,
including the first cold import in 7.5 seconds.
- Before the final rebase, recursive typecheck, full build, UI token
gates, 132 targeted server tests, and the complete [CI
workflow](https://github.com/paperclipai/paperclip/actions/runs/34650004085)
passed. The subsequent CI failure was the shard-balance accounting
mismatch fixed here.
- Greptile reviewed `d23c84181` at 5/5 with no outstanding actionable
findings. The complete [current CI
workflow](https://github.com/paperclipai/paperclip/actions/runs/34653327949)
passed on attempt 2. All test, typecheck, build, and canary jobs passed
on the first attempt. Docker setup timed out fetching BuildKit from
Docker Hub; retrying that job and its dependent aggregate succeeded.

## Risks

- Recovery admission changes executable authority. Company, task, agent,
user, approvals, process ownership, and provider termination checks
remain required.
- Explicit continuation starts a fresh conversation with history. It
does not certify unknown external action outcomes or rerun
non-conversation adapters automatically.
- Changing task status alone does not clear an execution hold. The task
now offers an explicit Retry action.
- Historical adapter claims and invocation events take precedence over
current agent settings. Known process or webhook runs retain their hold.
Pre-upgrade rows with no adapter evidence may receive only a new
explicit user turn after termination proof; they do not become eligible
for automatic replay.
- No schema migration or sandbox-image change is required. This branch
has not been deployed to production.

## Model Used

OpenAI GPT-6 through Codex, with repository inspection, code execution,
browser automation, and test execution. The exact deployment model ID
and context window are not exposed in this session.

## 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
- [ ] 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 2026-09-11 18:08:14 -05:00 committed by GitHub
parent 37d7dfb0e3
commit 9031516a7e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 517 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
doc/SPEC-implementation.md
View File

@ -1568,10 +1568,10 @@ Export/import behavior in V1:
- import preview reports skill-policy and legacy-grant mappings before apply and rejects unknown policy schema versions
- GitHub imports warn on unpinned refs instead of blocking
### User messages after native execution recovery stops
### User continuation after execution recovery stops
An authenticated user message can start a fresh native conversation turn once
the prior execution is confirmed stopped. Retain the source history and uncertain
An authenticated user message or an exact failed-run Retry can start a fresh
native or legacy conversation turn once the prior execution is confirmed stopped. Retain the source history and uncertain
action outcomes; do not replay tool calls or reset the failed incident's automatic
retry budget. Existing pause, approval, budget, ownership, and dependency gates
remain in effect. See `doc/execution-semantics.md` for admission and stop-proof

33
doc/execution-semantics.md
View File

@ -154,6 +154,11 @@ New comments received during an execution hold retain their individual deferred
The conversation groups repeated empty pre-start reconciliation cancellations into a neutral waiting notice. Started runs, actual startup failures, and run history remain inspectable. No historical run records are deleted.
The legacy remote ACP process-session relay runs on the control-plane host. Its
launch command uses the host's absolute Node executable even when the adapter's
launch environment is sanitized for a remote sandbox; the sandbox PATH remains
owned by the sandbox image.
### Pre-dispatch configuration validation
Pre-dispatch configuration validation is a distinct gate that runs after ownership and checkout are resolved but before the control plane actually dispatches a run.
@ -846,7 +851,7 @@ Local recovery records a server-authored stop receipt before it clears a verifie
If cleanup or another execution gate is still pending, the message stays in its existing queue receipt. Startup and periodic scheduling reconsider up to 50 due receipts per pass, at most once per 30 seconds per receipt, without calling a model or resetting recovery attempts. Cleanup callbacks use the same admission path. The issue lock prevents concurrent workers from delivering an adopted or discarded receipt again. The queued-message area shows the current wait reason. Pauses, approvals, budgets, ownership, and external chat authorization remain enforced. A message sent before the run finished does not grant new post-stop authority.
Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Classification uses the run’s saved adapter invocation or continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the hold. A terminal row with a live predecessor process or unreleased environment lease still blocks actual admission and Resume. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced.
Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Automatic classification uses the server-owned adapter identity saved atomically at run claim, the saved adapter invocation, or the continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the automatic hold; an explicit user continuation can retire it after proving the predecessor stopped. A terminal row with a live predecessor process, an unreleased environment lease, or failed/pending cleanup still blocks actual admission and Resume; a release timestamp alone does not prove cleanup succeeded. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced.
The server projection remains available for diagnostics. Normal working, finishing, and interaction waits add no badges or cards to task lists or feeds. Active transcript headers keep saying Working during automatic retry and execution confirmation; attempts, causes, and recovery decisions belong in the run log. Recovery uses the existing transcript and run log rather than adding a reconciliation form. A cancelled run that never started says “Couldn't start” instead of implying that the agent answered.
@ -870,16 +875,23 @@ new run. Preserve the baseline across recovery of the same run and start a new
delta when attaching a new run. Other stale-event and authority checks remain.
### Explicit user continuation after a native failure
### Explicit user continuation after execution failure
An execution recovery hold blocks automatic replay. A new authenticated user
comment can authorize a fresh native conversation turn after the predecessor's
comment or exact failed-run Retry can authorize a fresh native or legacy
conversation turn after the predecessor's
execution is confirmed stopped. This is a new request, not another automatic
attempt in the failed incident. The old attempt count and unknown action outcomes
remain unchanged.
remain unchanged. Known non-conversation adapter evidence still requires its
original reconciliation flow even if the agent's current settings change.
Pre-upgrade runs with no adapter evidence may receive a new explicit user turn
only after termination is proven; their old adapter and action outcomes remain
unknown, and they do not gain automatic replay eligibility.
Admission validates the persisted comment's author, task, and time against every
held predecessor. An agent-authored comment, an old queued request, or a generic
held predecessor. Retry validates the selected failed run's company, task, and
agent and preserves that run's identity through admission and history loading.
Duplicate Retry requests adopt the same successor. An agent-authored comment, an old queued request, or a generic
system wake cannot release a hold. The source task keeps its assignee. Process
ownership, active controllers, cleanup leases, pause, approval, budget, and normal
execution gates still apply. Dependency-blocked interaction mode remains limited
@ -891,7 +903,7 @@ request, task history, completed work, and the interruption notice. It receives
no instruction to repeat old tool calls. Later messages cannot reset the old
incident's retry budget or create another automatic replacement for it.
Native admission verifies local process identities for local runs. Remote runs
Explicit continuation verifies local process identities for local runs. Remote runs
instead require a provider termination receipt for every lease, with successful
cleanup and no active ownership. This applies to both per-turn and warm native
runners. A stop receipt retires only the settled cleanup owner for that exact company, run, provider, and sandbox resource, without changing its checkpoint or recorded action outcomes. Independent remote sandboxes have separate cleanup gates, including when one run owns multiple sandboxes. Successful pending-cleanup retries persist the same receipt and reconsider deferred user messages; a delivery failure never reverts successful provider cleanup. A failed checkpoint does not prevent destruction of a terminal run's isolated sandbox; busy ownership still prevents it.
@ -900,6 +912,15 @@ no receipt remain supported but cannot authorize remote continuation. A terminal
database status or a PID check on the wrong host is insufficient.
No historical task is automatically awakened by this change.
Startup waits for provider plugin initialization before remote recovery and
lease cleanup. The task's blocked notice offers Retry, and a refused retry
shows the actual recovery hold. Each explicit user Retry can make one scoped
cleanup attempt for its failed run even after automatic cleanup is exhausted.
If that attempt fails, a later user Retry may try again after the provider
recovers. The failed cleanup keeps the execution hold in place. Retry does not reset
the automatic limit or clean up another task's leases. Provider shutdown must
still be confirmed before a new conversation is admitted.
### Explicit Recovery Action
Paperclip opens an explicit recovery action when the system can identify a problem but cannot safely complete the work itself.

3
packages/adapter-utils/src/acpx-engine/execute.test.ts
View File

@ -2092,6 +2092,9 @@ describe("shared ACPX engine runtime behavior", () => {
expect(runtimeOptions[0]!.cwd).toBe(remoteCwd);
expect(sessionInputs[0]!.cwd).toBe(remoteCwd);
expect(runtimeOptions[0]!.spawnCwd).toBe(localCwd);
const proxyCommand = (runtimeOptions[0]!.agentRegistry as { resolve(name: string): string }).resolve("custom");
expect(proxyCommand.startsWith(`${JSON.stringify(process.execPath.replaceAll("\\", "/"))} `)).toBe(true);
expect(proxyCommand).toContain("paperclip-process-session-proxy.mjs");
expect(runtimeOptions[0]!.spawnCwd).not.toBe(sessionInputs[0]!.cwd);
const payloadEnv = ((sessionPayload as Record<string, unknown> | null)?.env ?? {}) as Record<string, unknown>;
expect(payloadEnv).toMatchObject({

7
packages/adapter-utils/src/acpx-engine/execute.ts
View File

@ -2484,7 +2484,12 @@ async function buildRuntime(input: {
await emitRunPhaseTiming(input.ctx, "start_transport", nowMs() - startTransportStart, "failed");
throw err;
}
const overrideCommand = processSessionBridge?.agentCommand ?? agentCommand;
// The relay runs on the host with the sanitized remote launch environment.
// Its /usr/bin/env node shebang cannot rely on that environment's PATH.
const overrideCommand = processSessionBridge?.agentCommand
? [process.execPath, processSessionBridge.agentCommand]
.map((part) => JSON.stringify(part.replaceAll("\\", "/"))).join(" ")
: agentCommand;
const overrides = overrideCommand ? { [acpxAgent]: overrideCommand } : undefined;
const agentRegistry = createAgentRegistry({ overrides });
const loggedEnv = buildInvocationEnvForLogs(env, {

3
packages/skills-catalog/src/shipped-catalog.test.ts
View File

@ -40,6 +40,9 @@ const SKILL_FRONTMATTER_ROOTS = [
function listSkillFiles(dir: string): string[] {
return readdirSync(dir, { withFileTypes: true }).flatMap((entry) => {
// Standalone provider installs can contain third-party skills. They are not
// shipped Paperclip skills and must not participate in this repo audit.
if (entry.name === "node_modules") return [];
const entryPath = path.join(dir, entry.name);
if (entry.isDirectory()) return listSkillFiles(entryPath);
if (entry.isFile() && entry.name === "SKILL.md") return [entryPath];

8
scripts/__tests__/e2e-shard.test.mjs
View File

@ -6,7 +6,7 @@ import path from "node:path";
import { fileURLToPath } from "node:url";
import test from "node:test";
import { loadShardDurations } from "../general-server-shard.mjs";
import { defaultSuiteWeight, loadShardDurations } from "../general-server-shard.mjs";
import { IGNORED_SPECS, listE2eSpecs, selectE2eShard } from "../e2e-shard.mjs";
const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..", "..");
@ -128,8 +128,10 @@ test("the duration manifest only names specs that still exist", () => {
test("the weighted partition keeps the shards close to balanced", () => {
const durations = loadShardDurations(durationsManifest);
const specs = listE2eSpecs();
// New specs use the scheduler's median estimate until measured durations exist.
const fallbackWeight = defaultSuiteWeight(durations);
const weights = Array.from({ length: SHARD_COUNT }, (_, index) =>
selectE2eShard(specs, index, SHARD_COUNT, durations).reduce((sum, file) => sum + (durations[file] ?? 0), 0),
selectE2eShard(specs, index, SHARD_COUNT, durations).reduce((sum, file) => sum + (durations[file] ?? fallbackWeight), 0),
);
const heaviest = Math.max(...weights);
@ -140,7 +142,7 @@ test("the weighted partition keeps the shards close to balanced", () => {
// of on the PR critical path. A single indivisible spec (smoke-lab) can
// legitimately exceed the even cut on its own, so the bound is floored at
// the largest per-spec weight — the best any file-level partition can do.
const largestSpec = Math.max(...specs.map((file) => durations[file] ?? 0));
const largestSpec = Math.max(...specs.map((file) => durations[file] ?? fallbackWeight));
const bound = Math.max((total / SHARD_COUNT) * 1.15, largestSpec);
assert.ok(
heaviest <= bound,

14
server/src/__tests__/heartbeat-process-recovery.test.ts
View File

@ -2346,6 +2346,20 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
expect(runs).toHaveLength(0);
});
it("recovers legacy startup before adapter.invoke using the claimed adapter identity", async () => {
const f = await seedRunFixture({ agentStatus: "idle", adapterType: "claude_local" });
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId));
await db.update(heartbeatRuns).set({ runnerProfileJson: {
adapterDispatch: { adapterType: "claude_local" },
} }).where(eq(heartbeatRuns.id, f.runId));
await heartbeatService(db).reapOrphanedRuns();
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
expect(source.resultJson).toMatchObject({ conversationContinuation: "continue_conversation_v1" });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agentId));
expect(runs.filter(run => run.retryOfRunId === f.runId)).toHaveLength(1);
});
it("schedules one conversation continuation after losing the provider", async () => {
const { agentId, runId, issueId } = await seedRunFixture({
agentStatus: "idle",

3
server/src/index.ts
View File

@ -1428,6 +1428,9 @@ async function startServerWithDatabaseTeardown(
);
} else {
const startupHeartbeatRecovery = (async () => {
// Legacy remote recovery releases sandbox leases. Wait for provider
// workers before cleanup or retry admission, including unmanaged installs.
await app.locals.bundledPluginsStartup;
try {
const nativeRecovery =
await heartbeat.recoverNativeRunsAfterRestart();

15
server/src/routes/agents.ts
View File

@ -1,4 +1,5 @@
import { applyConnectorSkills, resolveConnectorAssignments, annotateConnectorSkills, isConnectorSkill } from "../services/connector-runtime.js";
import { getExecutionBlocker } from "../services/execution-blocker.js";
import { paperclipRunnerTransitionConfig, normalizeLegacyRunnerProvider, isPaperclipRunnerProvider } from "@paperclipai/adapter-utils";
import { executionProjectionForRun, executionProjectionsForRuns } from "../services/execution-projection.js";
import { Router, type NextFunction, type Request, type Response } from "express";
@ -1993,6 +1994,13 @@ export function agentRoutes(
.where(and(eq(issuesTable.id, issueId), eq(issuesTable.companyId, agent.companyId)))
.then((rows) => rows[0] ?? null);
const blocker = issue ? await getExecutionBlocker(db, agent.companyId, issueId) : null;
if (blocker) return {
status: "skipped" as const, reason: "execution_reconciliation_required",
message: blocker.nextAction, issueId,
executionRunId: blocker.runId, executionAgentId: blocker.agentId, executionAgentName: null,
};
if (!issue?.executionRunId) {
return {
status: "skipped" as const,
@ -5409,7 +5417,7 @@ export function agentRoutes(
type HeartbeatSource = "timer" | "assignment" | "on_demand" | "automation";
type WakeupRouteOpts = {
source: HeartbeatSource | undefined;
skippedResponse: (agent: NonNullable<Awaited<ReturnType<typeof svc.getById>>>) => unknown | Promise<unknown>;
skippedResponse: (agent: NonNullable<Awaited<ReturnType<typeof svc.getById>>>, payload: Record<string, unknown> | null) => unknown | Promise<unknown>;
};
const handleWakeupRoute = async (
req: Request,
@ -5533,6 +5541,7 @@ export function agentRoutes(
);
}
const run = await heartbeat.wakeup(id, {
failedRunId: req.body.failedRunId ?? null,
source: opts.source,
triggerDetail: req.body.triggerDetail ?? "manual",
reason: req.body.reason ?? null,
@ -5562,7 +5571,7 @@ export function agentRoutes(
});
if (!run) {
res.status(202).json(await opts.skippedResponse(agent));
res.status(202).json(await opts.skippedResponse(agent, wakePayload));
return;
}
@ -5602,7 +5611,7 @@ export function agentRoutes(
router.post("/agents/:id/wakeup", validate(wakeAgentSchema), async (req, res) => {
await handleWakeupRoute(req, res, {
source: req.body.source,
skippedResponse: (agent) => buildSkippedWakeupResponse(agent, req.body.payload ?? null),
skippedResponse: (agent, payload) => buildSkippedWakeupResponse(agent, payload),
});
});

24
server/src/services/conversation-continuation.ts
View File

@ -19,8 +19,15 @@ export function hasConversationContinuationPolicy(result: Record<string, unknown
return result?.conversationContinuation === CONVERSATION_CONTINUATION_POLICY;
}
/** Persisted by the server when it claims the run, before remote provisioning. */
export function claimedAdapterType(run: Pick<typeof heartbeatRuns.$inferSelect, "runnerProfileJson">): string | null {
const dispatch = run.runnerProfileJson?.adapterDispatch as Record<string, unknown> | undefined;
return typeof dispatch?.adapterType === "string" ? dispatch.adapterType : null;
}
function conversationRunPredicate() {
return or(
inArray(sql`${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType'`, [...CONVERSATION_ADAPTER_TYPES]),
sql`${heartbeatRuns.resultJson}->>'conversationContinuation' = ${CONVERSATION_CONTINUATION_POLICY}`,
sql`exists (
select 1 from ${heartbeatRunEvents}
@ -33,14 +40,21 @@ function conversationRunPredicate() {
}
/** Recovery must not infer the old adapter from the agent's mutable settings. */
export async function runUsedConversationAdapter(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<boolean> {
if (hasConversationContinuationPolicy(run.resultJson)) return true;
export async function historicalAdapterType(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<string | null> {
const selected = claimedAdapterType(run);
if (selected) return selected;
const [invocation] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents)
.where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
eq(heartbeatRunEvents.eventType, "adapter.invoke")))
.orderBy(desc(heartbeatRunEvents.seq)).limit(1);
const adapterType = invocation?.payload?.adapterType;
return typeof adapterType === "string" && isConversationAdapter(adapterType);
return typeof adapterType === "string" ? adapterType : null;
}
export async function runUsedConversationAdapter(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<boolean> {
if (hasConversationContinuationPolicy(run.resultJson)) return true;
const adapterType = await historicalAdapterType(db, run);
return adapterType !== null && isConversationAdapter(adapterType);
}
/** Only immutable run evidence can retire a historical conversation hold.
@ -85,7 +99,9 @@ export async function getConversationOwnershipBlocker(db: Db, companyId: string,
const activeLease = sql`exists (select 1 from ${environmentLeases}
where ${environmentLeases.companyId} = "heartbeat_runs"."company_id"
and ${environmentLeases.heartbeatRunId} = "heartbeat_runs"."id"
and ${environmentLeases.releasedAt} is null)`;
and (${environmentLeases.releasedAt} is null
or ${environmentLeases.status} = 'pending_cleanup'
or ${environmentLeases.cleanupStatus} = 'failed'))`;
const candidates = await db.select({ run: heartbeatRuns, activeLease }).from(heartbeatRuns)
.where(and(
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.runtimeMode, "legacy"),

24
server/src/services/execution-continuation.ts
View File

@ -1,5 +1,6 @@
import { and, asc, desc, eq, isNotNull, isNull, sql } from "drizzle-orm";
import {
agentWakeupRequests,
heartbeatRuns,
issueComments,
issueRecoveryActions,
@ -73,6 +74,8 @@ export async function buildExecutionContinuation(input: {
agentId: string;
context: Record<string, unknown>;
previousContextRunId?: string | null;
/** Server-owned current run identity when validating dispatch authority. */
runId?: string;
summary: string | null;
exposeLowTrustRaw: boolean;
}): Promise<ExecutionContinuationEnvelope> {
@ -209,7 +212,7 @@ export async function buildExecutionContinuation(input: {
row.authorType === "user" && !row.createdByRunId && !row.deleted && row.body.trim().length > 0,
);
const priorRuns = await db
.select({ id: heartbeatRuns.id, result: heartbeatRuns.resultJson, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode })
.select({ id: heartbeatRuns.id, result: heartbeatRuns.resultJson, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode, retryOfRunId: heartbeatRuns.retryOfRunId })
.from(heartbeatRuns)
.where(
and(
@ -258,14 +261,25 @@ export async function buildExecutionContinuation(input: {
const explicitUserSource = string(explicitContinuation.previousRunId);
if (explicitUserSource) {
const predecessor = priorRuns.find(run => run.id === explicitUserSource &&
run.runtimeMode === "native" && ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
const failedRunId = string(explicitContinuation.failedRunId);
const retryWakes = failedRunId ? await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, input.agentId),
eq(agentWakeupRequests.reason, "retry_failed_run"), eq(agentWakeupRequests.requestedByActorType, "user"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)) : [];
const authorization = reconciliations.map(row => object(row.evidence.explicitUserContinuation))
.find(value => value.previousRunId === explicitUserSource &&
(!input.runId || value.runId === input.runId) &&
value.commentId === explicitContinuation.commentId &&
priorRuns.some(run => run.id === value.runId) &&
rows.some(comment => comment.id === value.commentId &&
comment.authorType === "user" && comment.authorUserId === value.actorId &&
!comment.createdByRunId && !comment.deletedAt));
(failedRunId
? value.failedRunId === failedRunId && retryWakes.some(wake =>
wake.runId === value.runId && wake.requestedByActorId === value.actorId &&
priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId))
: rows.some(comment => comment.id === value.commentId &&
comment.authorType === "user" && comment.authorUserId === value.actorId &&
!comment.createdByRunId && !comment.deletedAt)));
if (!predecessor || !authorization || explicitUserSource !== sourceRunId)
throw new Error("continuation_user_authorization_missing");
}

158
server/src/services/explicit-native-continuation.test.ts
View File

@ -147,6 +147,154 @@ const support = await getEmbeddedPostgresTestSupport();
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
});
it.each(["issue_commented", "retry_failed_run"])("continues a legacy Daytona run lost before adapter.invoke: %s", async reason => {
const f = await seed();
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", processPid: null,
errorCode: "process_lost" }).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
const [environment] = await db.insert(environments).values({ name: `Daytona startup ${f.sourceRunId}`, driver: "sandbox" }).returning();
const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
provider: "daytona", providerLeaseId: "startup-sandbox" };
await db.insert(environmentLeases).values({ ...identity, environmentId: environment.id,
status: "released", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "success",
metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity,
{ providerLeaseId: identity.providerLeaseId, state: "destroyed" }) } });
const result = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, reason,
commentId: reason === "issue_commented" ? f.commentId : null,
failedRunId: reason === "retry_failed_run" ? f.sourceRunId : null,
db: tx as unknown as typeof db }));
expect(result).toMatchObject({ previousRunId: f.sourceRunId });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
expect(source.resultJson).toBeNull();
});
it.each(["claim", "invocation"])("does not convert a known process run after switching the agent to Claude: %s", async evidence => {
const f = await seed();
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", errorCode: "process_lost",
runnerProfileJson: evidence === "claim" ? { adapterDispatch: { adapterType: "process" } } : null,
}).where(eq(heartbeatRuns.id, f.sourceRunId));
if (evidence === "invocation") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId,
runId: f.sourceRunId, agentId: f.agentId, seq: 1, eventType: "adapter.invoke", payload: { adapterType: "process" } });
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
expect(await admit(f)).toBeNull();
expect(await admitExplicitNativeContinuation({ ...f, db, reason: "retry_failed_run",
commentId: null, failedRunId: f.sourceRunId })).toBeNull();
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
});
it("keeps failed remote cleanup blocked even after the lease release timestamp is recorded", async () => {
const f = await seed();
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", processPid: null,
resultJson: { conversationContinuation: "continue_conversation_v1" } }).where(eq(heartbeatRuns.id, f.sourceRunId));
const [environment] = await db.insert(environments).values({ name: `Cleanup ${f.sourceRunId}`, driver: "sandbox" }).returning();
await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId,
environmentId: environment.id, provider: "daytona", providerLeaseId: "still-running",
status: "pending_cleanup", releasedAt: new Date(), cleanupStatus: "failed", leasePolicy: "ephemeral" });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ cause: "execution_owner_active" });
await db.delete(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.sourceRunId));
});
it("retries exhausted cleanup only for the selected failed run and adopts concurrent Retry clicks", async () => {
const f = await seed(), other = await seed();
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
const identities = [f, other].map(fixture => ({ id: randomUUID(), companyId: fixture.companyId,
heartbeatRunId: fixture.sourceRunId, provider: "daytona", providerLeaseId: fixture.sourceRunId }));
for (const identity of identities) await db.insert(environmentLeases).values({ ...identity,
status: "pending_cleanup", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "failed",
metadata: { pendingCleanupRetryAttempts: 5, pendingCleanupRetryCapWarned: true } });
const destroyed: string[] = [];
let readyCount = 0;
let bothReady!: () => void;
const ready = new Promise<void>(resolve => { bothReady = resolve; });
const heartbeat = heartbeatService(db, { environmentRuntime: {
isPendingCleanupWorkerReady: async () => { if (++readyCount === 2) bothReady(); await ready; return true; },
retryPendingSandboxTeardown: async ({ lease }: { lease: { id: string; providerLeaseId: string } }) => {
destroyed.push(lease.id);
return { providerLeaseId: lease.providerLeaseId, state: "destroyed" };
},
} as unknown as HeartbeatEnvironmentRuntime });
const request = { source: "on_demand" as const, triggerDetail: "manual" as const,
reason: "retry_failed_run", failedRunId: f.sourceRunId,
requestedByActorType: "user" as const, requestedByActorId: "board", payload: { issueId: f.issueId } };
try {
const [first, second] = await Promise.all([heartbeat.wakeup(f.agentId, request), heartbeat.wakeup(f.agentId, request)]);
// A losing cleanup claim can still see the hold until the winner finishes;
// a subsequent click adopts the already admitted successor.
const successor = first ?? second;
expect(successor?.id).toBeTruthy();
expect((await heartbeat.wakeup(f.agentId, request))?.id).toBe(successor?.id);
expect(destroyed).toEqual([identities[0].id]);
const [untouched] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, identities[1].id));
expect(untouched).toMatchObject({ status: "pending_cleanup", metadata: { pendingCleanupRetryAttempts: 5 } });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
} finally {
for (const identity of identities) await db.delete(environmentLeases).where(eq(environmentLeases.id, identity.id));
}
});
it("allows a later user cleanup attempt after transient failure without resetting automatic retries", async () => {
const f = await seed();
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
provider: "daytona", providerLeaseId: f.sourceRunId };
await db.insert(environmentLeases).values({ ...identity, status: "pending_cleanup", leasePolicy: "ephemeral",
releasedAt: new Date(), cleanupStatus: "failed", metadata: { pendingCleanupRetryAttempts: 5 } });
let attempts = 0;
const heartbeat = heartbeatService(db, { environmentRuntime: {
retryPendingSandboxTeardown: async () => {
if (++attempts < 3) throw new Error("provider temporarily unavailable");
return { providerLeaseId: identity.providerLeaseId, state: "destroyed" };
},
} as unknown as HeartbeatEnvironmentRuntime });
const request = { source: "on_demand" as const, triggerDetail: "manual" as const,
reason: "retry_failed_run", failedRunId: f.sourceRunId, requestedByActorType: "user" as const,
requestedByActorId: "board", payload: { issueId: f.issueId } };
try {
expect(await heartbeat.wakeup(f.agentId, request)).toBeNull();
expect(attempts).toBe(1);
expect(await heartbeat.wakeup(f.agentId, request)).toBeNull();
expect(attempts).toBe(2);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
await heartbeat.sweepPendingCleanupLeases();
expect(attempts).toBe(2);
const successor = await heartbeat.wakeup(f.agentId, request);
expect(attempts).toBe(3);
expect(successor?.retryOfRunId).toBe(f.sourceRunId);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
expect((await heartbeat.wakeup(f.agentId, request))?.id).toBe(successor?.id);
expect(attempts).toBe(3);
} finally {
await db.delete(environmentLeases).where(eq(environmentLeases.id, identity.id));
}
});
it("queues one exact Retry with fresh history and adopts repeated clicks", async () => {
const f = await seed();
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
const service = heartbeatService(db);
const request = { source: "on_demand" as const, triggerDetail: "manual" as const,
reason: "retry_failed_run", failedRunId: f.sourceRunId,
requestedByActorType: "user" as const, requestedByActorId: "board", payload: { issueId: f.issueId } };
const [first, second] = await Promise.all([service.wakeup(f.agentId, request), service.wakeup(f.agentId, request)]);
expect(first?.id).toBeTruthy();
expect(second?.id).toBe(first?.id);
expect(first).toMatchObject({ retryOfRunId: f.sourceRunId,
contextSnapshot: { previousRunId: f.sourceRunId, forceFreshSession: true } });
const envelope = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
agentId: f.agentId, runId: first!.id, context: first!.contextSnapshot!, summary: null, exposeLowTrustRaw: false });
expect(envelope.interruptedRunId).toBe(f.sourceRunId);
await expect(buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
agentId: f.agentId, runId: randomUUID(), context: first!.contextSnapshot!, summary: null, exposeLowTrustRaw: false }))
.rejects.toThrow("continuation_user_authorization_missing");
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
});
it.each([true, false])("acknowledges a legacy remote Stop only after confirmed lease cleanup: %s", async confirmed => {
const f = await seed();
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
@ -199,13 +347,17 @@ const support = await getEmbeddedPostgresTestSupport();
it.each([
{ runtime: "native", retry: false }, { runtime: "native", retry: true },
{ runtime: "legacy", retry: false }, { runtime: "legacy", retry: true },
{ runtime: "legacy_startup", retry: false }, { runtime: "legacy_startup", retry: true },
])("resumes a user message after confirmed cleanup: %j", async ({ runtime, retry }) => {
const f = await seed();
if (runtime === "legacy") {
if (runtime.startsWith("legacy")) {
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", processPid: null,
resultJson: { executionCancellation: { state: "requested" } } }).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId,
if (runtime === "legacy_startup") {
await db.update(heartbeatRuns).set({ status: "failed", resultJson: null, errorCode: "process_lost" })
.where(eq(heartbeatRuns.id, f.sourceRunId));
} else await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId,
agentId: f.agentId, seq: 1, eventType: "adapter.invoke", payload: { adapterType: "claude_local" } });
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
@ -242,7 +394,7 @@ const support = await getEmbeddedPostgresTestSupport();
await heartbeat.resumeRemoteStopComments(source);
const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
expect(runs).toHaveLength(1);
if (runtime === "native") expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId,
if (runtime !== "legacy") expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId,
explicitUserContinuation: { commentId: f.commentId } });
else expect(runs[0].contextSnapshot).toMatchObject({ wakeCommentId: f.commentId });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();

59
server/src/services/explicit-native-continuation.ts
View File

@ -4,7 +4,7 @@ import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-e
import { z } from "zod";
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
import {
approvals, issueApprovals, issueThreadInteractions,
agents, approvals, issueApprovals, issueThreadInteractions,
environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions,
issues, nativeRunFinalizations, type Db,
} from "@paperclipai/db";
@ -13,6 +13,8 @@ import { buildExecutionContinuation } from "./execution-continuation.js";
import { adapterExecutionControls } from "./adapter-execution-control.js";
import { persistActivity } from "./activity-log.js";
import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js";
type Run = typeof heartbeatRuns.$inferSelect;
const terminal = ["failed", "interrupted", "timed_out", "cancelled"];
@ -29,25 +31,31 @@ export async function admitExplicitNativeContinuation(input: {
db: Db; companyId: string; issueId: string; agentId: string;
actorType: string | null | undefined; actorId: string | null | undefined;
reason: string | null; commentId: string | null; successorRunId: string;
failedRunId?: string | null;
dryRun?: boolean;
onBlocked?: (reason: string, message: string) => void;
}): Promise<{ previousRunId: string; commentId: string } | null> {
}): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string } | null> {
const { db, companyId, issueId, agentId, actorId, commentId } = input;
const blocked = (reason: string, message: string) => { input.onBlocked?.(reason, message); return null; };
if (input.actorType !== "user" || !actorId || !commentId ||
!["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? "")) return null;
if (!z.string().guid().safeParse(commentId).success) return null;
if (input.actorType !== "user" || !actorId) return null;
const retry = input.reason === "retry_failed_run" &&
z.string().guid().safeParse(input.failedRunId).success;
if (!retry && (!commentId || !z.string().guid().safeParse(commentId).success ||
!["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? ""))) return null;
const [task] = await db.select().from(issues).where(and(
eq(issues.companyId, companyId), eq(issues.id, issueId),
));
if (!task || task.assigneeAgentId !== agentId || ["done", "cancelled"].includes(task.status)) return null;
const [comment] = await db.select().from(issueComments).where(and(
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"),
eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"),
eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId),
isNull(issueComments.deletedAt),
));
if (!comment?.body.trim()) return null;
if (!retry && !comment?.body.trim()) return null;
const authorizedAt = comment?.createdAt ?? new Date();
const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId)));
if (!agent || (!isConversationAdapter(agent.adapterType) && agent.adapterType !== "paperclip_runner")) return null;
const actions = await db.select().from(issueRecoveryActions).where(and(
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
executionBlockerPredicate(),
@ -76,12 +84,24 @@ 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 (comment.createdAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
if (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" &&
!run.processPid && !run.processGroupId && !run.nativeSessionId;
if (run.runtimeMode !== "native" && !unusedAdmission) return null;
const legacyUserTurn = run.runtimeMode === "legacy" &&
action.cause === "legacy_execution_requires_reconciliation" &&
isConversationAdapter(agent.adapterType);
if (legacyUserTurn) {
const historicalAdapter = await historicalAdapterType(db, run);
// A settings change never converts a known process/webhook execution into
// a conversation. Those adapters retain their reconciliation contract.
if (historicalAdapter && !isConversationAdapter(historicalAdapter)) return null;
}
// For pre-upgrade rows without adapter evidence, only a new explicit user
// turn is allowed, after the termination proofs below. This does not infer
// an old adapter type, certify old outcomes, or authorize automatic replay.
if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn) return null;
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
)).for("update");
@ -95,7 +115,7 @@ export async function admitExplicitNativeContinuation(input: {
if (remote) {
// Never interpret remote PIDs using the control-plane host's process table.
if (!leases.every(hasRemoteTerminationReceipt)) return blocked("remote_cleanup", "Waiting for the previous environment to stop. Your message will start automatically.");
if (!input.dryRun && !leases.every(lease => completeTerminatedRemoteNativeSessionCleanup({
if (run.runtimeMode === "native" && !input.dryRun && !leases.every(lease => completeTerminatedRemoteNativeSessionCleanup({
companyId, runId: run.id, remoteCleanupScope: remoteLeaseCleanupScope(lease)!,
}))) return null;
} else {
@ -111,7 +131,8 @@ export async function admitExplicitNativeContinuation(input: {
sources.push(run);
}
const nativeSources = sources.filter(run => run.runtimeMode === "native");
if (!nativeSources.length) return null;
const executedSources = sources.filter(run => run.runtimeMode === "native" || run.errorCode !== "execution_reconciliation_required" || run.startedAt);
if (!executedSources.length || (retry && !sources.some(run => run.id === input.failedRunId))) return null;
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, companyId),
or(eq(heartbeatRuns.nativeIssueId, issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`),
@ -119,22 +140,22 @@ export async function admitExplicitNativeContinuation(input: {
ne(heartbeatRuns.id, input.successorRunId),
)).limit(1);
if (active) return blocked("execution_active", "Waiting for the current run. Your message is saved.");
const previous = nativeSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!;
const previous = executedSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!;
// Prove required task history is available before retiring any hold.
await buildExecutionContinuation({ db, companyId, issueId, agentId,
context: { previousRunId: previous.id, wakeCommentId: commentId },
summary: null, exposeLowTrustRaw: false });
if (input.dryRun) return { previousRunId: previous.id, commentId };
const authorization = { actorId, commentId, runId: input.successorRunId,
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), runId: input.successorRunId,
previousRunId: previous.id, recordedAt: new Date().toISOString() };
await db.update(nativeRunFinalizations).set({
if (nativeSources.length) await db.update(nativeRunFinalizations).set({
failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`,
updatedAt: new Date(),
}).where(and(eq(nativeRunFinalizations.companyId, companyId), inArray(nativeRunFinalizations.runId, nativeSources.map(run => run.id))));
for (const action of actions) {
await db.update(issueRecoveryActions).set({
status: "resolved", outcome: "cancelled", resolvedAt: new Date(), updatedAt: new Date(),
nextAction: "A new user message starts a fresh conversation turn. Prior action outcomes remain recorded.",
nextAction: "The user started a fresh conversation turn. Prior action outcomes remain recorded.",
resolutionNote: "The user continued after the prior execution stopped. No action outcomes were inferred.",
wakePolicy: null, monitorPolicy: null,
evidence: { ...action.evidence, explicitUserContinuation: authorization,
@ -146,8 +167,8 @@ export async function admitExplicitNativeContinuation(input: {
}
await persistActivity(db, { companyId, actorType: "user", actorId,
action: "issue.execution_recovery_settled", entityType: "issue", entityId: issueId,
details: { continuation: "explicit_user_message", ...authorization,
details: { continuation: retry ? "explicit_user_retry" : "explicit_user_message", ...authorization,
recoveryActionIds: actions.map(action => action.id), previousRunIds: sources.map(run => run.id) },
});
return { previousRunId: previous.id, commentId };
return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
}

94
server/src/services/heartbeat.ts
View File

@ -4,7 +4,7 @@ import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteClean
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js";
import { CONVERSATION_CONTINUATION_POLICY, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js";
import { CONVERSATION_CONTINUATION_POLICY, claimedAdapterType, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js";
import { recordExecutionWait } from "./execution-wait.js";
import {
legacyExecutionNeedsReconciliation,
@ -3493,6 +3493,8 @@ function normalizeMaxConcurrentRuns(value: unknown) {
}
interface WakeupOptions {
/** Exact failed run selected by an authenticated board Retry request. */
failedRunId?: string | null;
durableChatRequest?: DurableChatWakeupRequest;
source?: "timer" | "assignment" | "on_demand" | "automation";
triggerDetail?: "manual" | "ping" | "callback" | "system";
@ -9964,10 +9966,10 @@ export function heartbeatService(
if (run.runtimeMode !== "native" && !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
if (!issueId) return;
const legacyContinuation = run.runtimeMode === "legacy" && run.status === "cancelled" &&
const legacyContinuation = run.runtimeMode === "legacy" &&
hasConversationContinuationPolicy((await getRun(run.id))?.resultJson) &&
!(await getExecutionBlocker(db, run.companyId, issueId));
if (run.runtimeMode !== "native" && !legacyContinuation) return;
if (run.runtimeMode !== "native" && run.runtimeMode !== "legacy") return;
const pending = await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
@ -16911,6 +16913,7 @@ export function heartbeatService(
.update(heartbeatRuns)
.set({
status: "running",
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
responsibleUserId,
startedAt: lockedRun.startedAt ?? claimedAt,
updatedAt: claimedAt,
@ -17007,6 +17010,7 @@ export function heartbeatService(
.update(heartbeatRuns)
.set({
status: "running",
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
responsibleUserId,
startedAt: lockedRun.startedAt ?? claimedAt,
contextSnapshot: withQueuedCommentIdsInRunContext(
@ -17073,6 +17077,7 @@ export function heartbeatService(
.update(heartbeatRuns)
.set({
status: "running",
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
responsibleUserId,
startedAt: run.startedAt ?? claimedAt,
updatedAt: claimedAt,
@ -17635,12 +17640,14 @@ export function heartbeatService(
async function claimPendingCleanupRetryAttempt(
leaseId: string,
expectedAttempts: number,
manualAttempt?: { previousId: unknown },
): Promise<boolean> {
const now = new Date();
const claimed = await db
.update(environmentLeases)
.set({
metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], to_jsonb(${expectedAttempts + 1}::int), true)`,
metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], to_jsonb(${expectedAttempts + 1}::int), true)
|| ${JSON.stringify(manualAttempt ? { pendingCleanupManualAttemptId: randomUUID() } : {})}::jsonb`,
lastUsedAt: now,
updatedAt: now,
})
@ -17649,6 +17656,7 @@ export function heartbeatService(
eq(environmentLeases.id, leaseId),
eq(environmentLeases.status, "pending_cleanup"),
sql`${pendingCleanupAttemptsSql()} = ${expectedAttempts}`,
manualAttempt ? sql`coalesce(${environmentLeases.metadata}->'pendingCleanupManualAttemptId', 'null'::jsonb) is not distinct from ${JSON.stringify(manualAttempt.previousId ?? null)}::jsonb` : undefined,
),
)
.returning({ id: environmentLeases.id });
@ -17718,6 +17726,11 @@ export function heartbeatService(
// cap and then stops the retries for that lease.
async function sweepPendingCleanupLeases(opts?: {
backoffMs?: number;
/** One cleanup attempt per explicit user Retry, for this failed run only.
* A later user Retry may try again after a provider failure; automatic
* sweeps retain their exhausted budget and never gain extra attempts.
*/
explicitRetry?: { companyId: string; runId: string; actorId: string };
}): Promise<{
swept: number;
destroyed: number;
@ -17733,7 +17746,7 @@ export function heartbeatService(
// `pending_cleanup` row lands once the database recovers. The flush runs
// before the read below, so this same tick tears down a freshly-landed row.
try {
const flushed = await environmentRuntime.flushDeferredOrphanCleanups?.();
const flushed = opts?.explicitRetry ? null : await environmentRuntime.flushDeferredOrphanCleanups?.();
if (flushed && (flushed.recovered > 0 || flushed.pending > 0)) {
logger.info(
{ recovered: flushed.recovered, pending: flushed.pending },
@ -17756,6 +17769,8 @@ export function heartbeatService(
.where(
and(
eq(environmentLeases.status, "pending_cleanup"),
opts?.explicitRetry ? eq(environmentLeases.companyId, opts.explicitRetry.companyId) : undefined,
opts?.explicitRetry ? eq(environmentLeases.heartbeatRunId, opts.explicitRetry.runId) : undefined,
backoffMs > 0 ? lte(environmentLeases.updatedAt, cutoff) : undefined,
),
)
@ -17768,7 +17783,7 @@ export function heartbeatService(
const metadata = { ...(row.metadata ?? {}) } as Record<string, unknown>;
const attempts = readPendingCleanupRetryAttempts(metadata);
if (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) {
if (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP && !opts?.explicitRetry) {
capped += 1;
// Warn once, then leave the lease for manual cleanup. The atomic claim
// keeps the warning to one log line even when two sweeps overlap.
@ -17839,8 +17854,14 @@ export function heartbeatService(
// never tears the same sandbox down twice or exceeds the attempt cap. The
// claim records the attempt before the retry, so a thrown driver error
// still counts against the cap.
const claimed = await claimPendingCleanupRetryAttempt(row.id, attempts);
const claimed = await claimPendingCleanupRetryAttempt(row.id, attempts,
opts?.explicitRetry ? { previousId: metadata.pendingCleanupManualAttemptId } : undefined);
if (!claimed) continue;
if (opts?.explicitRetry) await logActivity(db, {
companyId: row.companyId, actorType: "user", actorId: opts.explicitRetry.actorId,
action: "environment_lease.cleanup_retried", entityType: "environment_lease", entityId: row.id,
runId: opts.explicitRetry.runId, details: { attempt: attempts + 1, reason: "retry_failed_run" },
});
try {
if (useRecordedTeardown) {
@ -19266,6 +19287,13 @@ export function heartbeatService(
return;
}
// The claimed adapter identity is immutable recovery evidence. Do not
// execute a newly selected adapter under a previous adapter's claim.
const selectedAdapter = claimedAdapterType(run);
if (selectedAdapter && selectedAdapter !== agent.adapterType) {
throw new Error("Agent adapter changed during startup; start a new turn with the updated agent.");
}
const runtime = await ensureRuntimeState(agent);
const context = parseObject(run.contextSnapshot);
const authorizeFailedChatRetryExecution = () =>
@ -19754,6 +19782,7 @@ export function heartbeatService(
companyId: agent.companyId,
issueId: issueRef.id,
agentId: agent.id,
runId: run.id,
context,
previousContextRunId: taskSession?.lastRunId,
summary: safeContinuationSummary?.body ?? null,
@ -22603,12 +22632,14 @@ export function heartbeatService(
nativeRuntimeResolution.resolverVersion,
runtimeModeReason: nativeRuntimeResolution.reason,
runtimeModeResolvedAt: run.runtimeModeResolvedAt ?? new Date(),
// Preserve only this row's server-owned admission field at the
// atomic write, never an input or previous runner's profile.
runnerProfileJson: sql`case when ${heartbeatRuns.runnerProfileJson} ? ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}
then ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : {})}::jsonb
|| jsonb_build_object(${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}::text, ${heartbeatRuns.runnerProfileJson} -> ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY})
else ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : null)}::jsonb end`,
// Preserve server-owned admission and dispatch evidence on this
// row; never copy another run's execution profile.
runnerProfileJson: sql`(case when ${heartbeatRuns.runnerProfileJson} ? ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}
then jsonb_build_object(${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}::text, ${heartbeatRuns.runnerProfileJson}->${CHAT_CONTROL_RECOVERY_ADMISSION_KEY})
else '{}'::jsonb end)
|| (case when ${heartbeatRuns.runnerProfileJson} ? 'adapterDispatch'
then jsonb_build_object('adapterDispatch', ${heartbeatRuns.runnerProfileJson}->'adapterDispatch')
else '{}'::jsonb end) || ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : {})}::jsonb`,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, run.id));
@ -24953,6 +24984,26 @@ export function heartbeatService(
}
}
if (opts.failedRunId) {
const failed = await getRun(opts.failedRunId);
if (opts.requestedByActorType !== "user" || !opts.requestedByActorId ||
reason !== "retry_failed_run" || source !== "on_demand" || triggerDetail !== "manual" ||
!failed || failed.companyId !== agent.companyId || failed.agentId !== agentId ||
!["failed", "timed_out"].includes(failed.status) ||
(failed.nativeIssueId ?? readNonEmptyString(failed.contextSnapshot?.issueId)) !== issueId) {
throw conflict("The selected failed run cannot be retried for this task.");
}
if (!activeRunExecutions.has(failed.id) && !adapterExecutionControls.has(failed.id)) {
await sweepPendingCleanupLeases({ explicitRetry: {
companyId: failed.companyId, runId: failed.id, actorId: opts.requestedByActorId,
} });
}
if (isConversationAdapter(agent.adapterType) || agent.adapterType === "paperclip_runner") {
enrichedContextSnapshot.previousRunId = failed.id;
enrichedContextSnapshot.forceFreshSession = true;
}
}
const durableRequest = opts.durableChatRequest;
if (durableRequest) {
assertDurableChatWakeupRequest(durableRequest, {
@ -25608,6 +25659,17 @@ export function heartbeatService(
return { kind: "skipped" as const };
}
if (opts.failedRunId) {
// The issue lock makes double-clicks and network retries adopt the
// same successor, including after it has already finished.
const [previousRetry] = await tx.select().from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, issue.companyId), eq(heartbeatRuns.agentId, agentId),
eq(heartbeatRuns.retryOfRunId, opts.failedRunId),
sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'retry_failed_run'`,
)).orderBy(desc(heartbeatRuns.createdAt)).limit(1);
if (previousRetry) return { kind: "replayed" as const, run: previousRetry };
}
let reconciledSourceRunId: string | null = null;
if (executionReconciliationWake) {
const actionId = readNonEmptyString(
@ -25771,7 +25833,7 @@ export function heartbeatService(
if (executionBlocker && !(await admitExplicitNativeContinuation({
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId,
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
dryRun: true,
onBlocked: (reason, message) => { continuationWait = { reason, message }; },
}))) return deferBlockedExecution(executionBlocker);
@ -26533,7 +26595,7 @@ export function heartbeatService(
const explicitContinuation = await admitExplicitNativeContinuation({
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId,
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
});
if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker);
if (explicitContinuation) {
@ -26614,7 +26676,7 @@ export function heartbeatService(
wakeupRequestId: wakeupRequest.id,
retryOfRunId: failedChatRetry
? durableRequest!.failedRunRetry!.failedRunId
: automaticParentRunId,
: opts.failedRunId ?? automaticParentRunId,
contextSnapshot: adoptedComments.length
? withQueuedCommentIdsInRunContext(
enrichedContextSnapshot,

76
tests/e2e/legacy-failure-continuation.spec.ts Normal file
View File

@ -0,0 +1,76 @@
import { randomUUID } from "node:crypto";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
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";
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) {
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-"));
const config = JSON.parse(await readFile(process.env.PAPERCLIP_E2E_SERVER_CONFIG!, "utf8"));
// Use the running test server's actual port, including fallback allocation.
const pid = await readFile(path.join(config.database.embeddedPostgresDataDir, "postmaster.pid"), "utf8");
const url = `postgres://paperclip:paperclip@127.0.0.1:${pid.split("\n")[3]}/paperclip`;
const db = createDb(url);
const company = await json(await request.post("/api/companies", { data: { name: `Legacy recovery ${action} ${Date.now()}` } }));
try {
await writeFile(path.join(root, "continued"), "ready");
const agent = await json(await request.post(`/api/companies/${company.id}/agents`, { data: {
name: "Recovery fixture", role: "engineer", adapterType: "claude_local",
adapterConfig: { engine: "acp", cwd: root, stateDir: path.join(root, "state"),
agentCommand: `${JSON.stringify(process.execPath)} ${JSON.stringify(path.resolve("scripts/mcp-fixtures/servers/acp-stop-agent.mjs"))}`,
env: { PAPERCLIP_STOP_FIXTURE_ROOT: root, PAPERCLIP_STOP_FIXTURE_FINISH_TASK: "1" } },
runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true } },
} }));
const issue = await json(await request.post(`/api/companies/${company.id}/issues`, { data: {
title: "Continue after startup failure", description: "Answer the pending follow-up once.",
status: "backlog", assigneeAgentId: agent.id,
} }));
const sourceRunId = randomUUID();
// 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,
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 },
});
await db.insert(issueRecoveryActions).values({ companyId: company.id, sourceIssueId: issue.id,
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: sourceRunId,
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
evidence: { runId: sourceRunId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
});
await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issue.id));
const taskUrl = `/${company.issuePrefix}/issues/${issue.identifier}`;
await page.goto(action === "inbox_retry" ? `/${company.issuePrefix}/inbox/all` : taskUrl);
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 {
await page.getByRole("button", { name: "Retry", exact: true }).click();
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);
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 });
await page.reload();
await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible();
} finally {
await request.patch(`/api/companies/${company.id}`, { data: { status: "archived" } });
await closeRegisteredClients(url);
await rm(root, { recursive: true, force: true });
}
});
}

3
tests/e2e/playwright.config.ts
View File

@ -19,6 +19,9 @@ const PLAYWRIGHT_CHANNEL = process.env.PAPERCLIP_PLAYWRIGHT_CHANNEL;
process.env.PAPERCLIP_HOME = PAPERCLIP_HOME;
process.env.PAPERCLIP_CONFIG = PAPERCLIP_CONFIG;
// Worker processes reload this config; retain the main process's server path
// for specs that seed historical database state in the throwaway instance.
process.env.PAPERCLIP_E2E_SERVER_CONFIG ??= PAPERCLIP_CONFIG;
// Specs that mint agent JWTs in-process (via createLocalAgentJwt) must derive
// the same per-instance signing key as the webServer, or verification fails
// with a 401 instead of authenticating as the agent.

48
ui/src/components/ExecutionBlockerNotice.tsx Normal file
View File

@ -0,0 +1,48 @@
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";
import { queryKeys } from "../lib/queryKeys";
import { Button } from "./ui/button";
export function ExecutionBlockerNotice({ companyId, issueId, blocker, onRetried }: {
companyId: string;
issueId: string;
blocker: ExecutionBlocker;
onRetried: () => void;
}) {
const queryClient = useQueryClient();
const { data: runs } = useQuery({
queryKey: queryKeys.issues.runs(issueId),
queryFn: () => activityApi.runsForIssue(issueId),
});
const failedRun = runs?.find(run => run.runId === blocker.runId &&
["failed", "timed_out"].includes(run.status));
const retry = useMutation({
mutationFn: () => agentsApi.retryFailedRun(failedRun!.agentId, failedRun!.runId, companyId),
onSuccess: () => {
onRetried();
for (const queryKey of [queryKeys.issues.detail(issueId), queryKeys.issues.runs(issueId),
queryKeys.issues.liveRuns(issueId), queryKeys.issues.activeRun(issueId)]) {
void queryClient.invalidateQueries({ queryKey });
}
},
});
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>{" "}
{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>
)}
</div>
);
}

19
ui/src/pages/IssueDetail.tsx
View File

@ -1,3 +1,4 @@
import { ExecutionBlockerNotice } from "../components/ExecutionBlockerNotice";
import type { TaskComposerPause } from "../components/task-chat/TaskChatPausedTakeover";
import { TaskDetailTasksPanel } from "@/components/task-detail/TaskDetailTasksPanel";
import { EmailThreadProvider } from "../components/EmailMessageCard";
@ -7659,23 +7660,7 @@ export function IssueDetail({ tasksTab }: { tasksTab?: TaskSidePanelProps["tasks
}
>
{issue.executionBlocker && (
<div
role="status"
className="px-(--sz-execution-blocker-inline) py-(--sz-execution-blocker-block) text-sm text-muted-foreground"
>
<span>
Work cannot start. {issue.executionBlocker.nextAction}
</span>{" "}
{issue.executionBlocker.runId &&
issue.executionBlocker.agentId && (
<Link
className="underline"
to={`/agents/${issue.executionBlocker.agentId}/runs/${issue.executionBlocker.runId}`}
>
View stopped run
</Link>
)}
</div>
<ExecutionBlockerNotice companyId={issue.companyId} issueId={issue.id} blocker={issue.executionBlocker} onRetried={invalidateIssueDetail} />
)}
{resolvedDetailTab === "chat" ? (
<IssueDetailChatTab