fix: verify ACP Stop and preserve safe continuation (#13119)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Task controls coordinate provider execution and queued user messages. > - Stop could finish before an embedded ACP provider stopped its tools. > - A later request could be held for reconciliation without a clear task response. > - A restored provider could also retain the stopped run's API credential. > - This pull request verifies provider termination and preserves safe session continuation. > - Operators can continue known-safe work and see why uncertain work cannot start. ## Linked Issues or Issue Description **What happened?** Stop could leave an embedded ACP provider running. A queued follow-up followed by “go” could fail before it reached the provider. Task chat could show a generic missing-response message. Even a restored session could use the previous run's credential and fail its task update. **Expected behavior** Stop waits for confirmed provider termination. A later explicit wake continues the same compatible session only when recorded actions have known outcomes. It carries pending comments and the current run's environment. Uncertain actions retain a visible reconciliation hold. Composer Stop preserves the existing pause rule: conversation can continue while paused, but task work requires Resume. **Steps to reproduce** 1. Start an embedded ACP task. 2. Send a second request while the provider is running. 3. Interrupt the run, then send “go”. Also test composer Stop followed by Resume work. 4. Check that the request is delivered once and that the provider can complete the task through the current run's API credential. 5. Repeat with an unfinished write. Confirm that the write stops and that further execution stays blocked with a visible reason. **Paperclip version or commit** Built from source on master at `3bc60dd8b` plus this branch. **Deployment mode** Local source build with an isolated embedded PostgreSQL instance. Refs #11183. Refs #12552. Those changes address recovery after operator cancellation. This change also covers embedded ACP termination, session proof, pending-comment delivery, and task feedback. ## What Changed - Propagate Stop into embedded ACP and wait for bounded adapter cleanup and provider exit. Retain the actual ChildProcess object for forced termination on all platforms; never signal a recycled numeric PID. - Preserve interrupted checkpoints only for acknowledged, local, persistent sessions with settled reads or no tools. Keep writes, incomplete actions, and forced termination blocked. - Restore the same compatible provider session with the current run's environment. Reject fresh-session fallback for an interrupted checkpoint. - Adopt pending comments on the next explicit wake. Stop alone does not dispatch them. - Share the execution-blocker rule across dispatch, Resume, and task detail. Show Stopped or Couldn't start with the recorded reason. Resolve the stopped agent for the run link, including reviewer runs. - Keep execution reconciliation holds intact when generic recovery sees queued comments or healthy child tasks. - Add process, service, component, and browser regression coverage. Fix disposable database cleanup and React test settling exposed by the full suite. ## Verification - Passed `pnpm -r typecheck`, `pnpm build`, and `pnpm check:token-gates`. - Passed all three `acp-stop-continuation.spec.ts` browser journeys. They use an actual ACP child process and require task completion through the agent API. - Passed 165 adapter execution, operator-stop, and child-process control tests, 17 queued-comment route tests, and 65 tests in the two adjusted UI suites. Earlier focused recovery, heartbeat, and task-control tests also passed. - Manually used the browser to queue a request, Stop, send “go” while paused, and Resume. The same session answered once and moved the task to Done with the current run's credential. - Manually interrupted an unfinished write. Its file size stayed fixed for five seconds. “Go” showed the reconciliation reason and did not start another provider prompt. - Separate live Claude ACP smoke checks confirmed that Stop ended a disposable local write and that a no-tool interruption could resume the exact provider session. The browser fixture does not call Drive or another external app. - Passed all 5,615 UI tests and 3,090 other workspace tests. The CLI and general server groups pass with targeted retries: two transient server failures passed together on retry, and two embedded-database startup failures passed after removing abandoned shared-memory segments from this task's completed browser fixtures. All 144 serialized server suites completed, with 2,189 tests passing after two transient HTTP socket failures passed on retry. - Passed all 135 heartbeat process/recovery tests, including a deterministic regression that failed before the recovery-sweep fix. - Passed 18 dispatch integration tests, including stopped-reviewer links, company boundaries, and malformed run IDs. - Greptile is 5/5 on `7dd170d83`, with zero unresolved review threads. The security scan and all required CI gates pass for the same commit. ## Risks - Safe continuation depends on complete tool reporting and a restorable local provider session. Unknown outcomes remain blocked and require reconciliation. - Provider cleanup can take time. A timeout does not grant replay permission. - The change adds optional adapter context fields and an optional issue projection. It does not change the database schema or require a migration. - Test cleanup truncates company data only in a disposable test database. ## Model Used OpenAI GPT-6, running as Codex with repository tools, code execution, and browser interaction. The runtime does not expose a more specific model deployment ID or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
bd1fdc2887
commit
018ca5daaf
|
|
@ -33,6 +33,56 @@ eligible tasks still receive their wake requests. No new endpoint is introduced.
|
|||
The deterministic E2E fixtures prove interruption, then record their known lack
|
||||
of external effects through the existing reconciliation API before continuing.
|
||||
|
||||
Embedded ACP also supports verified continuation of an interrupted local session.
|
||||
The adapter must acknowledge cancellation and prove a preserved session with
|
||||
settled read-only work. Stop waits for provider cleanup. Forced local termination uses the actual
|
||||
child-process handle captured at spawn, including on Windows. An unavailable
|
||||
handle does not authorize a signal or replay. Unknown actions remain
|
||||
blocked, and task detail shows the reason even after recovery bookkeeping resolves.
|
||||
A run-level Stop leaves the task unpaused; a subsequent comment can continue the
|
||||
same session with the earlier queued messages. Composer Stop still creates a
|
||||
pause hold. Human comments can receive a response within the existing paused
|
||||
conversation scope; task execution requires Resume. Neither path permits a fresh-session fallback
|
||||
when the interrupted checkpoint cannot be restored.
|
||||
|
||||
The credential-free ACP regression journey uses an actual ACP child process:
|
||||
|
||||
```sh
|
||||
pnpm exec playwright test --config tests/e2e/playwright.config.ts tests/e2e/acp-stop-continuation.spec.ts
|
||||
```
|
||||
|
||||
It covers the queued-follow-up sequence (queue a second request, stop, then send “go”),
|
||||
same-session delivery of both messages, and an unfinished write that stops
|
||||
mutating its file but retains a visible execution blocker. Unit and integration
|
||||
tests additionally cover pre-start Stop, unavailable/changed sessions, rotating
|
||||
scratch directories, cancellation acknowledgment, a provider that hangs during
|
||||
cleanup after returning cancellation, deferred-wake adoption, and
|
||||
company-scoped blocker lookup. Hosted-provider behavior is a separate smoke test.
|
||||
|
||||
The browser tests also require the continued provider to complete the task through
|
||||
the agent API. Restoring a session must refresh its run identity, API credential,
|
||||
and scratch environment. The same conversation must not reuse the stopped run's
|
||||
credential. A regression test checks distinct run IDs and token hashes across the
|
||||
restart without logging the credentials themselves.
|
||||
|
||||
On 2026-09-09, all three ACP browser journeys passed. A manual browser walk-through
|
||||
also queued a request, used composer Stop, sent “go” while paused, and selected
|
||||
Resume work. The pause stayed in place during the conversation reply. Resume
|
||||
restored the same provider session, answered the pending request once, and moved
|
||||
the task to Done through the current run's authenticated API call. These tests use
|
||||
a deterministic ACP child process; they do not call Drive or another external app.
|
||||
The manual unfinished-write check also confirmed that file size stayed unchanged
|
||||
for five seconds after Interrupt. Sending “go” displayed the reconciliation reason
|
||||
and did not start another provider prompt.
|
||||
|
||||
A separate live Claude ACP smoke test interrupted a Bash tool writing only to a
|
||||
disposable local file: cancellation settled in 1,167 ms, output remained unchanged
|
||||
for five seconds, and no matching tool process remained. The shell action correctly
|
||||
did not receive automatic replay permission. A second live Claude check interrupted
|
||||
a response with no tools, restored the exact same provider session, and received
|
||||
the requested follow-up answer. These are local-provider observations, not a
|
||||
latency guarantee or proof for every provider and remote sandbox.
|
||||
|
||||
## Quiet task feedback
|
||||
|
||||
The visible task/subtree does not produce duplicate state toasts. Its live
|
||||
|
|
|
|||
|
|
@ -815,6 +815,10 @@ Legacy adapters without a verified resume capability use the same automatic no-r
|
|||
|
||||
The server projection remains available for execution diagnostics. Normal working, finishing, and interaction waits add no badges or cards to task lists or feeds. A retry may briefly change the existing transcript header to Reconnecting; attempts, causes, and recovery decisions belong in the run log. There is no reconciliation dialog. Safe recovery remains automatic. If it cannot continue safely, the source-scoped recovery record resolves with a blocked no-replay disposition and the ordinary task status becomes blocked, preserving its owner. Resolving this record does not grant replay authority: dispatch continues enforcing the durable hold. Replacement history remains inspectable and the composer stays usable.
|
||||
|
||||
An operator Stop reaches embedded ACP execution through its run-owned cancellation signal. The response waits for adapter settlement; acknowledgment requires the local provider to have exited. A deadline or failed cleanup never grants continuation permission. A persistent local ACP session can record an interrupted checkpoint only after acknowledged cancellation, complete tool reporting with settled reads (or no tools), and successful cleanup. Writes, shell commands, incomplete client-operation receipts, forced cancellation, and lost transports retain the ordinary no-replay hold. Continuation must restore the same compatible session; an unavailable checkpoint cannot fall back to a new session. A restored provider receives the current run identity, API credential, and scratch environment. Run-owned scratch paths rotate without changing session identity, while user configuration changes still invalidate compatibility.
|
||||
|
||||
Stop alone does not promote deferred messages. A subsequent explicit wake adopts pending comment IDs atomically in order through the existing queue. The task's ordered continuation history remains authoritative. A subtree pause still requires Resume; the text “go” has no special bypass. Task detail exposes the effective execution blocker, including a recovery record resolved with replay blocked, using the same predicate as dispatch and Resume. A cancelled run that never started says “Couldn't start” instead of claiming successful completion without an answer. Historical ambiguous executions remain held. A queued message or healthy child task cannot clear an execution reconciliation hold during a generic recovery sweep.
|
||||
|
||||
### Codex startup and provider state
|
||||
|
||||
Paperclip trusts the server-selected startup execution root in the isolated
|
||||
|
|
@ -834,6 +838,7 @@ a baseline; do not emit a warning or charge its historical `last` usage to the
|
|||
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 Recovery Action
|
||||
|
||||
Paperclip opens an explicit recovery action when the system can identify a problem but cannot safely complete the work itself.
|
||||
|
|
|
|||
|
|
@ -1436,6 +1436,22 @@ describe("shared ACPX engine runtime behavior", () => {
|
|||
expect(fp(sameEnvNewWake)).toBe(fp(first));
|
||||
});
|
||||
|
||||
it("keeps rotated run scratch paths out of session identity while retaining user temp overrides", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const config = { agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") };
|
||||
async function withScratch(dir: string, userTemp: string) {
|
||||
return runExecutor({ ...config, env: {
|
||||
PAPERCLIP_RUN_SCRATCH_DIR: dir, PAPERCLIP_TASK_SCRATCH_DIR: dir,
|
||||
PAPERCLIP_SCRATCH_DIR: dir, PAPERCLIP_TMPDIR: dir, TEMP: dir, TMP: dir, TMPDIR: userTemp,
|
||||
} }, { context: { taskId: "issue-1", paperclipScratch: { type: "heartbeat_run", dir, tempKeysApplied: ["TEMP", "TMP"] } } });
|
||||
}
|
||||
const first = await withScratch(path.join(root, "run-1"), "/custom/tmp-1");
|
||||
const second = await withScratch(path.join(root, "run-2"), "/custom/tmp-1");
|
||||
const changed = await withScratch(path.join(root, "run-3"), "/custom/tmp-2");
|
||||
expect(second.result.sessionParams?.configFingerprint).toBe(first.result.sessionParams?.configFingerprint);
|
||||
expect(changed.result.sessionParams?.configFingerprint).not.toBe(first.result.sessionParams?.configFingerprint);
|
||||
});
|
||||
|
||||
it("busts the session fingerprint when a stable configured PAPERCLIP_* value rotates", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ import {
|
|||
type ReferencedSourceIgnoreResolution,
|
||||
type SandboxAdditionalSource,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import { captureLocalProcess, capturedProcessExited, killCapturedLocalProcess } from "./local-process-control.js";
|
||||
import type { DuplexLossReason } from "../duplex-observability.js";
|
||||
import { DUPLEX_CHANNEL_LOST_ERROR_CODE } from "../bridge-transport-contract.js";
|
||||
import type { WorkspaceRestoreFailureCode, WorkspaceRestoreOutcome } from "../workspace-restore-merge.js";
|
||||
|
|
@ -87,6 +88,7 @@ import {
|
|||
type AcpRuntimeTurnResult,
|
||||
type AcpRuntimeUsageBreakdown,
|
||||
type AcpRuntimeUsageCost,
|
||||
type AcpSessionStore,
|
||||
} from "acpx/runtime";
|
||||
import {
|
||||
ACPX_HANDSHAKE_TIMEOUT_MS,
|
||||
|
|
@ -1893,6 +1895,12 @@ async function buildRuntime(input: {
|
|||
// `env` above and are never present in shapedEnvConfig, so they inherently
|
||||
// stay out of the hash and don't reset the session every heartbeat.
|
||||
const resolvedAdapterEnv: Record<string, string> = {};
|
||||
const scratch = parseObject(context.paperclipScratch);
|
||||
const scratchKeys = scratch.type === "heartbeat_run" && typeof scratch.dir === "string"
|
||||
? new Set(["PAPERCLIP_RUN_SCRATCH_DIR", "PAPERCLIP_TASK_SCRATCH_DIR", "PAPERCLIP_SCRATCH_DIR", "PAPERCLIP_TMPDIR",
|
||||
...(Array.isArray(scratch.tempKeysApplied) ? scratch.tempKeysApplied.filter((key): key is string =>
|
||||
typeof key === "string" && ["TMPDIR", "TEMP", "TMP"].includes(key)) : [])])
|
||||
: new Set<string>();
|
||||
for (const [key, value] of Object.entries(shapedEnvConfig)) {
|
||||
if (typeof value !== "string") continue;
|
||||
// Runtime PAPERCLIP_* always wins over config: skip a PAPERCLIP_* key that
|
||||
|
|
@ -1903,7 +1911,10 @@ async function buildRuntime(input: {
|
|||
if (isForbiddenConfigEnvKey(key)) continue;
|
||||
if (isPaperclipRuntimeEnvKey(key) && key in env) continue;
|
||||
env[key] = value;
|
||||
resolvedAdapterEnv[key] = value;
|
||||
// The server rotates run-owned scratch paths on every wake. Still forward
|
||||
// them, but only hash actual adapter settings. User-supplied temp overrides
|
||||
// are absent from tempKeysApplied and keep their compatibility protection.
|
||||
if (!scratchKeys.has(key) || value !== scratch.dir) resolvedAdapterEnv[key] = value;
|
||||
}
|
||||
if (authToken) env.PAPERCLIP_API_KEY = authToken;
|
||||
// For the claude agent, set model via ANTHROPIC_MODEL at startup rather than
|
||||
|
|
@ -3778,7 +3789,21 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
// null on the host lane (no staging) and on a build failure (where
|
||||
// `buildRuntime` already released its own partial lease).
|
||||
let releaseStagingLease: (() => void) | null = null;
|
||||
let stopTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let removeStopListener: (() => void) | undefined;
|
||||
let forcedStop = false;
|
||||
let runtimeStopConfirmed = false;
|
||||
let safeInterruptedSession = false;
|
||||
const interruptionTools = new Map<string, { kind?: string; status?: string }>();
|
||||
let incompleteToolInventory = false;
|
||||
try {
|
||||
await ctx.onCancellationReady?.();
|
||||
if (ctx.signal?.aborted) return {
|
||||
exitCode: null, signal: null, timedOut: false,
|
||||
errorCode: "cancelled", errorMessage: "Stopped before provider startup",
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
resultJson: { executionCancellation: { state: "acknowledged", acknowledgedAt: new Date().toISOString(), forced: false } },
|
||||
};
|
||||
// Evict idle staged runtimes BEFORE building the runtime, since buildRuntime
|
||||
// consults the staged cache to decide whether a compatible resume may reuse
|
||||
// an already-staged runtime — an expired entry must not be reused.
|
||||
|
|
@ -4003,6 +4028,9 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
|
||||
const previousParams = parseObject(ctx.runtime.sessionParams);
|
||||
const canResume = isCompatibleSession(previousParams, prepared);
|
||||
if (previousParams.interruptedCheckpoint === true && !canResume) {
|
||||
throw new Error("The interrupted session is no longer compatible. Its action history must be checked before starting a new session.");
|
||||
}
|
||||
const resumeSessionId = canResume ? asString(previousParams.acpSessionId, "") || undefined : undefined;
|
||||
// Borrow the warm entry without removing it, so an overlapping run of the
|
||||
// same session still sees it. The borrow clears the entry's idle timer, so
|
||||
|
|
@ -4020,6 +4048,27 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
processIdentitySink.current = ctx.onSpawn;
|
||||
flushChildStderr(childStderrState);
|
||||
childStderrState.logPath = prepared.childStderrLogPath;
|
||||
const persistedRuntimeStore = createRuntimeStore({ stateDir: prepared.stateDir });
|
||||
const runtimeStore: AcpSessionStore = {
|
||||
async load(id) {
|
||||
const record = await persistedRuntimeStore.load(id);
|
||||
if (!record) return undefined;
|
||||
// ACPX resumes from the stored session options rather than the
|
||||
// options passed to ensureSession. Keep conversation state, but
|
||||
// launch the provider with this run's credentials and scratch paths.
|
||||
return {
|
||||
...record,
|
||||
acpx: {
|
||||
...record.acpx,
|
||||
session_options: {
|
||||
...record.acpx?.session_options,
|
||||
env: { ...prepared.env },
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
save: (record) => persistedRuntimeStore.save(record),
|
||||
};
|
||||
const runtimeOptions: PaperclipAcpRuntimeOptions = {
|
||||
cwd: prepared.cwd,
|
||||
// Host-only spawn cwd for the relay proxy on the remote process-session
|
||||
|
|
@ -4028,7 +4077,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
// fingerprint / compat key are unaffected — this redirects ONLY the host
|
||||
// `spawn()` `chdir`, not the in-sandbox data path.
|
||||
spawnCwd: prepared.hostSpawnCwd,
|
||||
sessionStore: createRuntimeStore({ stateDir: prepared.stateDir }),
|
||||
sessionStore: runtimeStore,
|
||||
agentRegistry: prepared.agentRegistry,
|
||||
permissionMode: prepared.permissionMode,
|
||||
nonInteractivePermissions: prepared.nonInteractivePermissions,
|
||||
|
|
@ -4047,6 +4096,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
: undefined,
|
||||
onAgentSpawn: async (meta) => {
|
||||
processIdentitySink.latest = meta;
|
||||
processIdentitySink.localProcess = prepared.processSessionBridge ? undefined : captureLocalProcess(meta.pid);
|
||||
await processIdentitySink.current?.({
|
||||
pid: meta.pid,
|
||||
processGroupId: null,
|
||||
|
|
@ -4181,7 +4231,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
}),
|
||||
});
|
||||
} catch (err) {
|
||||
if (!resumeSessionId || !isResumeFailure(err)) throw err;
|
||||
if (!resumeSessionId || !isResumeFailure(err) || previousParams.interruptedCheckpoint === true) throw err;
|
||||
clearSession = true;
|
||||
resumedSession = false;
|
||||
await ctx.onLog(
|
||||
|
|
@ -4229,6 +4279,9 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
});
|
||||
}
|
||||
// A compatible warm handle reuses the already-running ACP agent and does
|
||||
if (previousParams.interruptedCheckpoint === true && handle?.backendSessionId !== resumeSessionId) {
|
||||
throw new Error("The provider did not restore the interrupted session; refusing a fresh-session fallback.");
|
||||
}
|
||||
// not emit another spawn event. Persist its known identity on this run
|
||||
// before the next prompt starts so every running heartbeat is adoptable.
|
||||
if (handle && cached && processIdentitySink.latest && ctx.onSpawn) {
|
||||
|
|
@ -4534,6 +4587,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
turnPhaseStart = now();
|
||||
};
|
||||
const stepTurnStart = (signal: AbortSignal, startTimeoutMs: number | undefined): StartedTurn => {
|
||||
ctx.signal?.throwIfAborted();
|
||||
const turn = runtime.startTurn({
|
||||
handle: sessionHandle,
|
||||
text: runPrompt,
|
||||
|
|
@ -4543,6 +4597,25 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
signal,
|
||||
});
|
||||
activeTurn = turn;
|
||||
// ACP can resolve the turn before its provider exits. Keep the Stop
|
||||
// deadline armed through settlement, including provider cleanup.
|
||||
const armStopDeadline = () => {
|
||||
stopTimer = setTimeout(() => {
|
||||
forcedStop = true;
|
||||
const providerPid = processIdentitySink.latest?.pid;
|
||||
if (providerPid && !prepared.processSessionBridge) {
|
||||
// Escalate the owned local process directly. A second ACP close
|
||||
// can wait behind the same hung cleanup (or open a new client).
|
||||
killCapturedLocalProcess(processIdentitySink.localProcess);
|
||||
} else {
|
||||
void runtime.close({ handle: sessionHandle, reason: "operator stop deadline", discardPersistentState: true })
|
||||
.catch(() => {});
|
||||
}
|
||||
}, Math.max(1, asNumber(ctx.config.graceSec, 15)) * 1000);
|
||||
};
|
||||
ctx.signal?.addEventListener("abort", armStopDeadline, { once: true });
|
||||
removeStopListener = () => ctx.signal?.removeEventListener("abort", armStopDeadline);
|
||||
if (ctx.signal?.aborted) armStopDeadline();
|
||||
return {
|
||||
cancel: async (reason: string) => {
|
||||
await turn.cancel({ reason });
|
||||
|
|
@ -4553,6 +4626,19 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
const turn = activeTurn as AcpRuntimeTurn;
|
||||
const toolTitles = new Map<string, string>();
|
||||
for await (const event of turn.events) {
|
||||
// ACPX currently flattens client-side filesystem/terminal receipts
|
||||
// into status text. They cannot establish complete action outcomes.
|
||||
if (event.type === "status" && /^(fs|terminal)\//.test(event.text)) incompleteToolInventory = true;
|
||||
if (event.type === "tool_call") {
|
||||
if (!event.toolCallId) incompleteToolInventory = true;
|
||||
else {
|
||||
const previous = interruptionTools.get(event.toolCallId);
|
||||
interruptionTools.set(event.toolCallId, {
|
||||
kind: event.kind ?? previous?.kind,
|
||||
status: event.status ?? previous?.status,
|
||||
});
|
||||
}
|
||||
}
|
||||
if (event.type === "text_delta" && event.stream !== "thought") {
|
||||
currentOutputChunk.push(event.text);
|
||||
} else if (event.type === "tool_call" && event.tag !== "tool_call_update") {
|
||||
|
|
@ -4627,6 +4713,15 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
eventCostUsd,
|
||||
});
|
||||
const failedTurn = terminal.status === "failed" || terminal.status === "cancelled" || timedOut;
|
||||
// A provider-native command/write has no reliable external outcome
|
||||
// receipt. Only settled reads (or a turn with no tools) can establish
|
||||
// automatic interrupted-session continuity here.
|
||||
safeInterruptedSession = ctx.signal?.aborted === true && !forcedStop && !timedOut && !channelLost
|
||||
&& (terminal.status === "cancelled" || terminal.status === "completed")
|
||||
&& prepared.mode === "persistent" && !prepared.processSessionBridge
|
||||
&& Boolean(sessionHandle.backendSessionId)
|
||||
&& !incompleteToolInventory
|
||||
&& [...interruptionTools.values()].every((tool) => tool.kind === "read" && tool.status === "completed");
|
||||
// Record how the settlement `endSession` step closes the runtime for this
|
||||
// outcome. A clean persistent host turn is save-eligible, but the
|
||||
// Amendment B credential gate fails on the host lane (the run API key is
|
||||
|
|
@ -4646,7 +4741,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
: failedTurn
|
||||
? `paperclip turn ${terminal.status}`
|
||||
: "paperclip completed turn cleanup",
|
||||
discardPersistentState: terminal.status === "cancelled" || timedOut || channelLost,
|
||||
discardPersistentState: (terminal.status === "cancelled" && !safeInterruptedSession) || timedOut || channelLost,
|
||||
dropWarmEntry: false,
|
||||
recordCloseError: false,
|
||||
cancelTurnReason: null,
|
||||
|
|
@ -4826,6 +4921,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
};
|
||||
try {
|
||||
return await runTurnSequence<AcpRuntimeTurnResult>({
|
||||
signal: ctx.signal,
|
||||
timeoutMs: prepared.timeoutSec > 0 ? prepared.timeoutSec * 1000 : undefined,
|
||||
timeoutMessage: formatAdapterExecutionTimeoutErrorMessage(prepared.timeoutResolution),
|
||||
promptBuild: stepPromptBuild,
|
||||
|
|
@ -4960,7 +5056,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
});
|
||||
return;
|
||||
}
|
||||
const onCloseError = settlement.recordCloseError
|
||||
const onCloseError = settlement.recordCloseError || ctx.signal?.aborted
|
||||
? (closeErr: unknown) => recordTeardownError("runtime-close", closeErr)
|
||||
: () => {};
|
||||
await runtime
|
||||
|
|
@ -4969,6 +5065,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
reason: settlement.reason,
|
||||
discardPersistentState: settlement.discardPersistentState,
|
||||
})
|
||||
.then(() => { runtimeStopConfirmed = true; })
|
||||
.catch(onCloseError);
|
||||
if (settlement.dropWarmEntry && warmHandleMatches(existing, runtime, settlement.handle) && existing) {
|
||||
clearWarmHandleTimer(existing);
|
||||
|
|
@ -5041,10 +5138,34 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
recordDispositionReport(report);
|
||||
if (childStderrState) flushChildStderr(childStderrState);
|
||||
},
|
||||
reproduceResult: (): AdapterExecutionResult => {
|
||||
reproduceResult: async (): Promise<AdapterExecutionResult> => {
|
||||
if (!capturedResult) {
|
||||
throw new Error("run coordinator reproduced a result before the run recorded one");
|
||||
}
|
||||
clearTimeout(stopTimer);
|
||||
let providerExited = false;
|
||||
const providerPid = processIdentitySink?.latest?.pid;
|
||||
if (ctx.signal?.aborted && providerPid && !prepared.processSessionBridge) {
|
||||
for (let attempt = 0; attempt < 40; attempt += 1) {
|
||||
providerExited = capturedProcessExited(processIdentitySink.localProcess);
|
||||
if (providerExited || !forcedStop) break;
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
}
|
||||
}
|
||||
// A forced process exit can reject ACP close after the OS has already
|
||||
// confirmed termination. Acknowledge Stop, but never grant replay.
|
||||
if (ctx.signal?.aborted && providerExited && (runtimeStopConfirmed || forcedStop)) {
|
||||
capturedResult = {
|
||||
...capturedResult,
|
||||
...(safeInterruptedSession && !forcedStop && !("workspaceRestoreFailure" in workspaceRestoreFailureField)
|
||||
? { executionRecovery: { kind: "interrupted", providerStopped: true, sessionPreserved: true, actionOutcomes: "settled" },
|
||||
sessionParams: { ...capturedResult.sessionParams, interruptedCheckpoint: true } }
|
||||
: {}),
|
||||
resultJson: { ...capturedResult.resultJson, executionCancellation: {
|
||||
state: "acknowledged", acknowledgedAt: new Date().toISOString(), forced: forcedStop,
|
||||
} },
|
||||
};
|
||||
}
|
||||
// The sync-back settlement step runs before this reproduces the result
|
||||
// (settlement precedes reproduction), so a failed workspace restore is
|
||||
// already recorded by the time we get here. Merge it into `resultJson`
|
||||
|
|
@ -5063,6 +5184,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
|||
};
|
||||
return await runAttempt(plan);
|
||||
} finally {
|
||||
clearTimeout(stopTimer);
|
||||
removeStopListener?.();
|
||||
// End the run root span exactly once, on every return and on a throw.
|
||||
runRootSpan.end(runFailed);
|
||||
// Release the per-session staging lease as the run's final act, AFTER the
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
import { spawn } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { captureLocalProcess, capturedProcessExited, killCapturedLocalProcess } from "./local-process-control.js";
|
||||
|
||||
it("captures and terminates the actual spawned child on the current platform", async () => {
|
||||
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
|
||||
try {
|
||||
await once(child, "spawn");
|
||||
const captured = captureLocalProcess(child.pid!);
|
||||
expect(captured).toBe(child);
|
||||
expect(capturedProcessExited(captured)).toBe(false);
|
||||
const exited = once(child, "exit");
|
||||
expect(killCapturedLocalProcess(captured)).toBe(true);
|
||||
await exited;
|
||||
expect(capturedProcessExited(captured)).toBe(true);
|
||||
expect(captureLocalProcess(child.pid!)).toBeUndefined();
|
||||
} finally {
|
||||
if (!capturedProcessExited(child)) child.kill("SIGKILL");
|
||||
}
|
||||
});
|
||||
|
||||
it("never signals an exited child even if another process has reused its PID", () => {
|
||||
const original = { pid: 123, exitCode: 0, signalCode: null, kill: vi.fn(() => true) };
|
||||
const replacement = { pid: 123, exitCode: null, signalCode: null, kill: vi.fn(() => true) };
|
||||
expect(killCapturedLocalProcess(original)).toBe(false);
|
||||
expect(original.kill).not.toHaveBeenCalled();
|
||||
expect(replacement.kill).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("fails closed when the spawned handle was not captured", () => {
|
||||
expect(killCapturedLocalProcess(undefined)).toBe(false);
|
||||
expect(capturedProcessExited(undefined)).toBe(false);
|
||||
});
|
||||
|
||||
it("reports an unsuccessful signal without throwing", () => {
|
||||
expect(killCapturedLocalProcess({ exitCode: null, signalCode: null, kill: () => false })).toBe(false);
|
||||
expect(killCapturedLocalProcess({ exitCode: null, signalCode: null, kill: () => { throw new Error("gone"); } })).toBe(false);
|
||||
});
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
import { ChildProcess } from "node:child_process";
|
||||
import { channel } from "node:diagnostics_channel";
|
||||
|
||||
// Node publishes the real ChildProcess before its spawn event. ACP's spawn
|
||||
// callback only exposes a PID, so correlate it here while retaining the handle
|
||||
// for all later control. Never resolve a numeric PID again when stopping it.
|
||||
// https://nodejs.org/api/diagnostics_channel.html#event-child_process
|
||||
const children = new Map<number, ChildProcess>();
|
||||
channel("child_process").subscribe((message) => {
|
||||
const child = (message as { process?: unknown }).process;
|
||||
if (!(child instanceof ChildProcess)) return;
|
||||
child.once("spawn", () => {
|
||||
if (child.pid) children.set(child.pid, child);
|
||||
});
|
||||
const remove = () => {
|
||||
if (child.pid && children.get(child.pid) === child) children.delete(child.pid);
|
||||
};
|
||||
child.once("exit", remove);
|
||||
child.once("close", remove);
|
||||
});
|
||||
|
||||
export function captureLocalProcess(pid: number): ChildProcess | undefined {
|
||||
return children.get(pid);
|
||||
}
|
||||
|
||||
export function capturedProcessExited(child: Pick<ChildProcess, "exitCode" | "signalCode"> | undefined): boolean {
|
||||
return Boolean(child && (child.exitCode !== null || child.signalCode !== null));
|
||||
}
|
||||
|
||||
export function killCapturedLocalProcess(child: Pick<ChildProcess, "exitCode" | "signalCode" | "kill"> | undefined): boolean {
|
||||
if (!child || capturedProcessExited(child)) return false;
|
||||
try { return child.kill("SIGKILL"); } catch { return false; }
|
||||
}
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
import fs from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { afterEach, expect, it } from 'vitest';
|
||||
import type { AdapterExecutionContext } from '../types.js';
|
||||
import { createAcpxEngineExecutor } from './execute.js';
|
||||
import { sessionCodec } from './session-codec.js';
|
||||
const fixture = fileURLToPath(new URL('../../../../scripts/mcp-fixtures/servers/acp-stop-agent.mjs', import.meta.url));
|
||||
const roots: string[] = [];
|
||||
afterEach(async () => { await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); });
|
||||
async function setup(tool?: string) {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'paperclip-acp-stop-'));
|
||||
roots.push(root);
|
||||
const abort = new AbortController();
|
||||
let ready!: () => void;
|
||||
const started = new Promise<void>(resolve => { ready = resolve; });
|
||||
const ctx = {
|
||||
runId: 'stop-test', agent: { id: 'agent', companyId: 'company' }, runtime: {},
|
||||
config: { agent: 'custom', agentCommand: `${JSON.stringify(process.execPath)} ${JSON.stringify(fixture)}`, mode: 'persistent',
|
||||
stateDir: path.join(root, 'state'), cwd: root, graceSec: 5,
|
||||
env: { PAPERCLIP_STOP_FIXTURE_ROOT: root, ...(tool ? { PAPERCLIP_STOP_FIXTURE_TOOL: tool } : {}) } },
|
||||
context: {}, signal: abort.signal,
|
||||
onLog: async (_stream: string, text: string) => { await fs.appendFile(path.join(root, 'logs'), text); if (text.includes('Waiting for Stop.')) ready(); },
|
||||
} as unknown as AdapterExecutionContext;
|
||||
return { root, ctx, abort, started, execute: createAcpxEngineExecutor() };
|
||||
}
|
||||
it('stops an actual ACP process and resumes its established session with the new request', async () => {
|
||||
const { root, ctx, abort, started, execute } = await setup('read');
|
||||
ctx.authToken = 'first-run-test-token';
|
||||
const running = execute(ctx);
|
||||
await started;
|
||||
abort.abort(new Error('Operator Stop'));
|
||||
const result = await running;
|
||||
expect(result.resultJson?.executionCancellation).toMatchObject({ state: 'acknowledged', forced: false });
|
||||
expect(result.executionRecovery).toMatchObject({ kind: 'interrupted', sessionPreserved: true });
|
||||
const params = sessionCodec.serialize(result.sessionParams ?? null);
|
||||
expect(params?.interruptedCheckpoint).toBe(true);
|
||||
const next = await execute({ ...ctx, runId: 'follow-up', authToken: 'follow-up-test-token', signal: undefined, context: { prompt: 'List recent Drive files' },
|
||||
runtime: { ...ctx.runtime, sessionParams: params } });
|
||||
expect(next.exitCode).toBe(0);
|
||||
const prompts = (await fs.readFile(path.join(root, 'prompts'), 'utf8')).trim().split('\n').map(line => JSON.parse(line));
|
||||
expect(prompts).toHaveLength(2);
|
||||
expect(prompts[1].sessionId).toBe(prompts[0].sessionId);
|
||||
const launches = (await fs.readFile(path.join(root, 'run-env'), 'utf8')).trim().split('\n').map(line => JSON.parse(line));
|
||||
expect(launches.map(launch => launch.runId)).toEqual(['stop-test', 'follow-up']);
|
||||
expect(launches[1].tokenHash).toBe(createHash('sha256').update('follow-up-test-token').digest('hex'));
|
||||
});
|
||||
it('stops writes but does not authorize replay when the interrupted tool has no outcome', async () => {
|
||||
const { root, ctx, abort, started, execute } = await setup('write');
|
||||
const running = execute(ctx);
|
||||
await started;
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
abort.abort();
|
||||
const result = await running;
|
||||
expect(result.resultJson?.executionCancellation).toMatchObject({ state: 'acknowledged' });
|
||||
expect(result.executionRecovery).toBeUndefined();
|
||||
const before = await fs.readFile(path.join(root, 'writes'), 'utf8');
|
||||
await new Promise(resolve => setTimeout(resolve, 5000));
|
||||
expect(await fs.readFile(path.join(root, 'writes'), 'utf8')).toBe(before);
|
||||
}, 15000);
|
||||
it('does not dispatch a provider when Stop precedes startup', async () => {
|
||||
const { root, ctx, abort, execute } = await setup();
|
||||
abort.abort(new Error('Stopped before startup'));
|
||||
expect(await execute(ctx)).toMatchObject({ errorCode: 'cancelled', executionRecovery: { kind: 'bootstrap', providerWorkStarted: false } });
|
||||
await expect(fs.access(path.join(root, 'prompts'))).rejects.toThrow();
|
||||
});
|
||||
|
||||
it.each(['missing session', 'changed configuration'])('refuses fresh-session fallback after Stop: %s', async (change) => {
|
||||
const { root, ctx, abort, started, execute } = await setup();
|
||||
const running = execute(ctx);
|
||||
await started;
|
||||
abort.abort();
|
||||
const result = await running;
|
||||
expect(result.executionRecovery?.kind).toBe('interrupted');
|
||||
if (change === 'missing session') await fs.rm(path.join(root, 'session'));
|
||||
const next = await execute({ ...ctx, signal: undefined,
|
||||
config: change === 'changed configuration' ? { ...ctx.config, env: { ...(ctx.config.env as object), SETTING: 'changed' } } : ctx.config,
|
||||
runtime: { ...ctx.runtime, sessionParams: sessionCodec.serialize(result.sessionParams ?? null) },
|
||||
});
|
||||
expect(next.exitCode).not.toBe(0);
|
||||
expect((await fs.readFile(path.join(root, 'prompts'), 'utf8')).trim().split('\n')).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('keeps the Stop deadline active after cancellation returns until provider exit', async () => {
|
||||
const { ctx, abort, started, execute } = await setup();
|
||||
ctx.config.graceSec = 1;
|
||||
ctx.config.env = { ...(ctx.config.env as object), PAPERCLIP_STOP_FIXTURE_HANG_ON_CLOSE: '1' };
|
||||
let providerPid: number | undefined;
|
||||
ctx.onSpawn = async ({ pid }) => { providerPid = pid; };
|
||||
const running = execute(ctx);
|
||||
await started;
|
||||
abort.abort();
|
||||
const result = await running;
|
||||
expect(result.resultJson?.executionCancellation).toMatchObject({ state: 'acknowledged', forced: true });
|
||||
expect(result.executionRecovery).toBeUndefined();
|
||||
expect(providerPid).toBeTypeOf('number');
|
||||
expect(() => process.kill(providerPid!, 0)).toThrow(expect.objectContaining({ code: 'ESRCH' }));
|
||||
}, 10000);
|
||||
|
|
@ -11,6 +11,7 @@
|
|||
// operations the host lane always did (host cwd, no staging failures, no
|
||||
// transport, no sync-back). A later phase routes the whole lane through the site.
|
||||
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import type { AdapterExecutionContext } from "@paperclipai/adapter-utils";
|
||||
import type { AcpRuntime, AcpRuntimeHandle } from "acpx/runtime";
|
||||
import type {
|
||||
|
|
@ -36,6 +37,7 @@ export type AcpxAgentProcessIdentity = { pid: number; startedAt: string };
|
|||
export type AcpxProcessIdentitySink = {
|
||||
current: AdapterExecutionContext["onSpawn"];
|
||||
latest: AcpxAgentProcessIdentity | null;
|
||||
localProcess?: ChildProcess;
|
||||
};
|
||||
|
||||
/** The live-line buffer and log path a warm runtime's child stderr carries. */
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ export const sessionCodec: AdapterSessionCodec = {
|
|||
|
||||
return {
|
||||
...(runtimeSessionName ? { runtimeSessionName } : {}),
|
||||
...(record.interruptedCheckpoint === true ? { interruptedCheckpoint: true } : {}),
|
||||
...(readString(record.sessionKey) ? { sessionKey: readString(record.sessionKey) } : {}),
|
||||
...(readString(record.acpxRecordId) ? { acpxRecordId: readString(record.acpxRecordId) } : {}),
|
||||
...(acpSessionId ? { acpSessionId } : {}),
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ export type TurnFinalizeInput<TTerminal> =
|
|||
* rejects.
|
||||
*/
|
||||
export interface TurnSteps<TTerminal> {
|
||||
readonly signal?: AbortSignal;
|
||||
/** The wall-clock timeout, in milliseconds, or undefined for no timeout. */
|
||||
readonly timeoutMs: number | undefined;
|
||||
/** The message the timeout cancel carries. */
|
||||
|
|
@ -72,11 +73,24 @@ export async function runTurn<TTerminal>(steps: TurnSteps<TTerminal>): Promise<T
|
|||
// `turnStarted` separates a pre-turn preparation failure from a running-turn
|
||||
// failure, so `turnFinalize` reports the right phase.
|
||||
let turnStarted = false;
|
||||
let cancellation: Promise<void> | null = null;
|
||||
const cancel = () => {
|
||||
controller.abort(steps.signal?.reason);
|
||||
if (started && !cancellation) {
|
||||
cancellation = started.cancel("Paperclip operator stop");
|
||||
// The result is observed below; do not create an unhandled rejection
|
||||
// while the runtime drains its event stream.
|
||||
void cancellation.catch(() => {});
|
||||
}
|
||||
};
|
||||
steps.signal?.addEventListener("abort", cancel, { once: true });
|
||||
try {
|
||||
steps.signal?.throwIfAborted();
|
||||
// Build the prompt and snapshot the pre-turn usage inside the failure
|
||||
// boundary. A failure here is a `prepare_turn` failure.
|
||||
await steps.promptBuild(controller.signal);
|
||||
await steps.preTurnUsage();
|
||||
steps.signal?.throwIfAborted();
|
||||
// The sequence owns the wall-clock timer. On a timeout it marks the run timed
|
||||
// out, aborts the shared signal, and cancels the started turn. The cancel
|
||||
// no-ops before `turnStart` returns, because `started` is still null.
|
||||
|
|
@ -88,13 +102,17 @@ export async function runTurn<TTerminal>(steps: TurnSteps<TTerminal>): Promise<T
|
|||
}, steps.timeoutMs);
|
||||
}
|
||||
started = steps.turnStart(controller.signal, steps.timeoutMs);
|
||||
if (steps.signal?.aborted) cancel();
|
||||
turnStarted = true;
|
||||
const terminal = await steps.eventRelay(started);
|
||||
if (cancellation) await cancellation;
|
||||
if (timeout) clearTimeout(timeout);
|
||||
return await steps.turnFinalize({ kind: "terminal", terminal, timedOut });
|
||||
} catch (error) {
|
||||
if (timeout) clearTimeout(timeout);
|
||||
const phase: "prepare_turn" | "turn" = turnStarted ? "turn" : "prepare_turn";
|
||||
return await steps.turnFinalize({ kind: "error", error, phase, timedOut });
|
||||
} finally {
|
||||
steps.signal?.removeEventListener("abort", cancel);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -76,7 +76,12 @@ export type AdapterExecutionErrorFamily =
|
|||
|
||||
export interface AdapterExecutionResult {
|
||||
/** Positive evidence for retrying bootstrap; absent evidence never authorizes replay. */
|
||||
executionRecovery?: { kind: "bootstrap"; providerWorkStarted: false };
|
||||
executionRecovery?: { kind: "bootstrap"; providerWorkStarted: false } | {
|
||||
kind: "interrupted";
|
||||
providerStopped: true;
|
||||
sessionPreserved: true;
|
||||
actionOutcomes: "settled";
|
||||
};
|
||||
exitCode: number | null;
|
||||
signal: string | null;
|
||||
timedOut: boolean;
|
||||
|
|
@ -190,6 +195,10 @@ export interface AdapterRuntimeEvent {
|
|||
}
|
||||
|
||||
export interface AdapterExecutionContext {
|
||||
/** Run-scoped operator cancellation; adapters must settle before returning. */
|
||||
signal?: AbortSignal;
|
||||
/** Opt in to signal-based cancellation before starting provider work. */
|
||||
onCancellationReady?: () => Promise<void>;
|
||||
/** Server-owned, actor-attributed snapshot also rendered by legacy wake prompts. */
|
||||
executionContinuation?: ExecutionContinuationEnvelope | null;
|
||||
runId: string;
|
||||
|
|
|
|||
|
|
@ -2755,6 +2755,6 @@ export {
|
|||
} from "./runtime-exposure/loopback-bind.js";
|
||||
export { ACCOUNT_HANDLE_MAX_LENGTH, toAccountHandle } from "./account-handle.js";
|
||||
export type { ExecutionContinuationEnvelope } from "./types/execution-continuation.js";
|
||||
export type { ExecutionProjection, ExecutionReconciliation } from "./types/execution-projection.js";
|
||||
export type { ExecutionProjection, ExecutionReconciliation, ExecutionBlocker } from "./types/execution-projection.js";
|
||||
|
||||
export { EXECUTION_RECONCILIATION_CAUSES, requiresExecutionReconciliation } from "./types/execution-projection.js";
|
||||
|
|
|
|||
|
|
@ -1,3 +1,11 @@
|
|||
export interface ExecutionBlocker {
|
||||
recoveryActionId: string;
|
||||
runId: string | null;
|
||||
agentId: string | null;
|
||||
cause: string;
|
||||
nextAction: string;
|
||||
}
|
||||
|
||||
/** Presentation of existing execution records, not a second task status machine. */
|
||||
export interface ExecutionProjection {
|
||||
phase:
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import type { ExecutionProjection } from "./execution-projection.js";
|
||||
import type { ExecutionProjection, ExecutionBlocker } from "./execution-projection.js";
|
||||
import type {
|
||||
IssueCommentAuthorType,
|
||||
IssueCommentMetadataRowType,
|
||||
|
|
@ -849,6 +849,7 @@ export interface Issue {
|
|||
productivityReview?: IssueProductivityReview | null;
|
||||
activeRecoveryAction?: IssueRecoveryAction | null;
|
||||
successfulRunHandoff?: SuccessfulRunHandoffState | null;
|
||||
executionBlocker?: ExecutionBlocker | null;
|
||||
watchdog?: IssueWatchdogSummary | null;
|
||||
scheduledRetry?: IssueScheduledRetry | null;
|
||||
liveDescendantCount?: number;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,79 @@
|
|||
#!/usr/bin/env node
|
||||
// Deterministic ACP process for interruption and same-session continuation tests.
|
||||
import fs from 'node:fs';
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import { createInterface } from 'node:readline';
|
||||
const root = process.env.PAPERCLIP_STOP_FIXTURE_ROOT ?? process.cwd();
|
||||
const send = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
|
||||
let active;
|
||||
let timer;
|
||||
if (process.env.PAPERCLIP_STOP_FIXTURE_HANG_ON_CLOSE === '1') {
|
||||
process.on('SIGTERM', () => {});
|
||||
setInterval(() => {}, 1000);
|
||||
}
|
||||
const update = (sessionId, value) => send({ jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: value } });
|
||||
async function request(message) {
|
||||
fs.appendFileSync(`${root}/requests`, `${Date.now()} ${message.method}\n`);
|
||||
switch (message.method) {
|
||||
case 'initialize': return { protocolVersion: 1, agentCapabilities: { loadSession: true, sessionCapabilities: { close: {} } }, agentInfo: { name: 'stop-fixture', version: '1' } };
|
||||
case 'session/new': {
|
||||
const sessionId = randomUUID();
|
||||
fs.writeFileSync(`${root}/session`, sessionId);
|
||||
return { sessionId };
|
||||
}
|
||||
case 'session/load':
|
||||
if (fs.readFileSync(`${root}/session`, 'utf8') !== message.params.sessionId) throw Error('Unknown session');
|
||||
return {};
|
||||
case 'session/prompt': {
|
||||
fs.appendFileSync(`${root}/prompts`, `${JSON.stringify(message.params)}\n`);
|
||||
fs.appendFileSync(`${root}/run-env`, `${JSON.stringify({ runId: process.env.PAPERCLIP_RUN_ID, tokenHash: createHash('sha256').update(process.env.PAPERCLIP_API_KEY ?? '').digest('hex'), scratchDir: process.env.PAPERCLIP_RUN_SCRATCH_DIR })}\n`);
|
||||
if (fs.existsSync(`${root}/continued`)) {
|
||||
const paused = JSON.stringify(message.params.prompt).includes('tree-hold interaction: yes');
|
||||
if (!paused) {
|
||||
fs.appendFileSync(`${root}/completed`, 'follow-up\n');
|
||||
// Browser journeys finish the task through the normal agent API so
|
||||
// the scheduler does not need a separate successful-run handoff.
|
||||
if (process.env.PAPERCLIP_STOP_FIXTURE_FINISH_TASK === '1') {
|
||||
const base = process.env.PAPERCLIP_API_URL.replace(/\/api\/?$/, '').replace(/\/$/, '');
|
||||
const response = await fetch(`${base}/api/issues/${process.env.PAPERCLIP_TASK_ID}`, {
|
||||
method: 'PATCH',
|
||||
headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.PAPERCLIP_API_KEY}`, 'X-Paperclip-Run-Id': process.env.PAPERCLIP_RUN_ID },
|
||||
body: JSON.stringify({ status: 'done' }),
|
||||
});
|
||||
if (!response.ok) throw new Error(`Task completion failed: ${response.status} ${await response.text()}`);
|
||||
}
|
||||
}
|
||||
update(message.params.sessionId, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: paused ? 'Task remains paused. Use Resume work to continue.' : 'Answered the pending follow-up once.' } });
|
||||
return { stopReason: 'end_turn' };
|
||||
}
|
||||
active = message;
|
||||
if (process.env.PAPERCLIP_STOP_FIXTURE_TOOL === 'read') {
|
||||
update(message.params.sessionId, { sessionUpdate: 'tool_call', toolCallId: 'read-1', title: 'Read local file', kind: 'read', status: 'completed' });
|
||||
}
|
||||
if (process.env.PAPERCLIP_STOP_FIXTURE_TOOL === 'write') {
|
||||
update(message.params.sessionId, { sessionUpdate: 'tool_call', toolCallId: 'write-1', title: 'Write local file', kind: 'edit', status: 'in_progress' });
|
||||
timer = setInterval(() => fs.appendFileSync(`${root}/writes`, 'tick\n'), 20);
|
||||
}
|
||||
update(message.params.sessionId, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Waiting for Stop.' } });
|
||||
return undefined;
|
||||
}
|
||||
case 'session/cancel':
|
||||
clearInterval(timer);
|
||||
fs.writeFileSync(`${root}/continued`, 'ready');
|
||||
if (active) send({ jsonrpc: '2.0', id: active.id, result: { stopReason: 'cancelled' } });
|
||||
active = undefined;
|
||||
return undefined;
|
||||
case 'session/close': clearInterval(timer); return {};
|
||||
case 'session/set_mode': case 'session/set_config_option': return {};
|
||||
default: throw Error(`Unsupported method: ${message.method}`);
|
||||
}
|
||||
}
|
||||
createInterface({ input: process.stdin }).on('line', async (line) => {
|
||||
const message = JSON.parse(line);
|
||||
try {
|
||||
const result = await request(message);
|
||||
if (message.id !== undefined && result !== undefined) send({ jsonrpc: '2.0', id: message.id, result });
|
||||
} catch (error) {
|
||||
if (message.id !== undefined) send({ jsonrpc: '2.0', id: message.id, error: { code: -32603, message: error.message } });
|
||||
}
|
||||
});
|
||||
|
|
@ -1,4 +1,7 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { adapterExecutionControls, createAdapterExecutionControl } from "../services/adapter-execution-control.js";
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
|
|
@ -1365,6 +1368,27 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
expect(recoveryRuns).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("keeps an unsafe Stop blocked when recovery sees a deferred human comment", async () => {
|
||||
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress", runStatus: "cancelled",
|
||||
resultJson: { executionCancellation: { state: "acknowledged" } },
|
||||
});
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
await terminalizeLegacyExecution({ db, run, status: "cancelled" });
|
||||
const wakeId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: wakeId, companyId, agentId, source: "on_demand", triggerDetail: "manual",
|
||||
reason: "issue_commented", payload: { issueId }, status: "deferred_issue_execution",
|
||||
});
|
||||
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId });
|
||||
await heartbeatService(db).reconcileStrandedAssignedIssues();
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId });
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId));
|
||||
expect(wake.status).toBe("deferred_issue_execution");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("leaves hidden issues out of stranded-issue reconciliation", async () => {
|
||||
const { issueId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
|
|
@ -5569,6 +5593,56 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
expect(repairWakeups).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("preserves deferred input on a clean Stop and adopts it once on the next explicit comment", async () => {
|
||||
const { companyId, agentId, issueId, runId } = await seedRunFixture({ runtimeMode: "legacy", agentStatus: "running" });
|
||||
const heartbeat = heartbeatService(db);
|
||||
const [pending] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "responsible-user", body: "List recent Drive files" }).returning();
|
||||
const [deferred] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution",
|
||||
payload: { issueId, commentId: pending!.id, _paperclipWakeContext: { issueId, wakeReason: "issue_commented", wakeCommentIds: [pending!.id] } },
|
||||
}).returning();
|
||||
await heartbeat.cancelRun(runId, "Operator Stop", { resultJson: {
|
||||
executionCancellation: { state: "acknowledged" },
|
||||
executionRecovery: { kind: "interrupted", providerStopped: true, sessionPreserved: true, actionOutcomes: "settled" },
|
||||
} });
|
||||
await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId))).toHaveLength(1);
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, deferred!.id)))[0]?.status).toBe("deferred_issue_execution");
|
||||
const [go] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "responsible-user", body: "go" }).returning();
|
||||
const next = await heartbeat.wakeup(agentId, { source: "automation", reason: "issue_commented", requestedByActorType: "user", requestedByActorId: "responsible-user",
|
||||
payload: { issueId, commentId: go!.id }, contextSnapshot: { issueId, commentId: go!.id, wakeReason: "issue_commented" },
|
||||
});
|
||||
expect(next?.contextSnapshot?.wakeCommentIds).toEqual([pending!.id, go!.id]);
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, deferred!.id)))[0]).toMatchObject({ status: "coalesced", runId: next!.id });
|
||||
await vi.waitFor(async () => expect((await heartbeat.getRun(next!.id))?.status).not.toBe("running"));
|
||||
});
|
||||
|
||||
it("signals an embedded adapter and waits for its cleanup before returning Stop", async () => {
|
||||
const { runId } = await seedRunFixture({ runtimeMode: "legacy", includeIssue: false });
|
||||
const control = createAdapterExecutionControl();
|
||||
adapterExecutionControls.set(runId, control);
|
||||
try {
|
||||
const heartbeat = heartbeatService(db);
|
||||
let returned = false;
|
||||
const stopping = heartbeat.cancelRun(runId).then((run) => { returned = true; return run; });
|
||||
await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true));
|
||||
const repeatedStop = heartbeat.cancelRun(runId);
|
||||
// Let the duplicate request observe the still-running execution.
|
||||
await new Promise(resolve => setTimeout(resolve, 25));
|
||||
expect(returned).toBe(false);
|
||||
expect((await heartbeat.getRun(runId))?.status).toBe("running");
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: {
|
||||
executionCancellation: { state: "acknowledged" },
|
||||
} }).where(eq(heartbeatRuns.id, runId));
|
||||
control.finish();
|
||||
expect(await stopping).toMatchObject({ status: "cancelled", resultJson: { executionCancellation: { state: "acknowledged" } } });
|
||||
expect(await repeatedStop).toMatchObject({ status: "cancelled", resultJson: { executionCancellation: { state: "acknowledged" } } });
|
||||
} finally {
|
||||
control.finish();
|
||||
adapterExecutionControls.delete(runId);
|
||||
}
|
||||
});
|
||||
|
||||
it("clears the detached warning when the run reports activity again", async () => {
|
||||
const { runId } = await seedRunFixture({
|
||||
includeIssue: false,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
|
|
@ -52,17 +52,9 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
|||
}, 30_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.update(issues).set({ executionRunId: null }).catch(() => undefined);
|
||||
await db.update(agentWakeupRequests).set({ runId: null }).catch(() => undefined);
|
||||
await db.delete(activityLog).catch(() => undefined);
|
||||
await db.delete(issueComments).catch(() => undefined);
|
||||
await db.delete(heartbeatRunEvents).catch(() => undefined);
|
||||
await db.delete(heartbeatRuns).catch(() => undefined);
|
||||
await db.delete(agentWakeupRequests).catch(() => undefined);
|
||||
await db.delete(issues).catch(() => undefined);
|
||||
await db.delete(companyMemberships).catch(() => undefined);
|
||||
await db.delete(agents).catch(() => undefined);
|
||||
await db.delete(companies).catch(() => undefined);
|
||||
// Each case owns the entire disposable database. Clear the full company
|
||||
// graph, including attribution rows and constraints added by migrations.
|
||||
await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import {
|
|||
startEmbeddedPostgresTestDatabase,
|
||||
} from "../../../__tests__/helpers/embedded-postgres.js";
|
||||
import { createPostgresRunDispatchAdapter } from "./postgres.js";
|
||||
import { getExecutionBlocker } from "../../../services/execution-blocker.js";
|
||||
|
||||
// Proves the DB-to-facts mapping this adapter owns for each state the two
|
||||
// run-dispatch gates decide on. `application/use-cases.test.ts` and
|
||||
|
|
@ -701,8 +702,31 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
|
|||
await db.insert(issues).values({ id: issueId, companyId, title: "Uncertain email", status: "in_progress", assigneeAgentId: agentId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "queued", contextSnapshot: { issueId, wakeReason: "retry_failed_run" } });
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", cause: "uncertain_external_action", status, evidence: status === "resolved" ? { automaticRecovery: { replay: "blocked" } } : {}, fingerprint: runId, nextAction: "Verify whether email-1 was sent before continuing." });
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ cause: "uncertain_external_action", nextAction: "Verify whether email-1 was sent before continuing." });
|
||||
expect(await getExecutionBlocker(db, randomUUID(), issueId)).toBeNull();
|
||||
const adapter = createPostgresRunDispatchAdapter(db);
|
||||
await expect(adapter.cancelStaleQueuedRun({ companyId, runId, expectedStatus: "queued", now: new Date() })).resolves.toMatchObject({ outcome: "cancelled", errorCode: "execution_reconciliation_required" });
|
||||
});
|
||||
|
||||
it("links the stopped run's agent instead of its return owner, within the same company", async () => {
|
||||
const { companyId, agentId: ownerId } = await seedCompanyAndAgent();
|
||||
const reviewerId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
await seedAgent({ id: reviewerId, companyId, name: "Reviewer" });
|
||||
await seedIssue({ companyId, issueId, status: "in_review", assigneeAgentId: ownerId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId: reviewerId, status: "cancelled" });
|
||||
const [action] = await db.insert(issueRecoveryActions).values({
|
||||
companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board",
|
||||
returnOwnerAgentId: ownerId, cause: "legacy_execution_requires_reconciliation", status: "active",
|
||||
evidence: { runId }, fingerprint: runId, nextAction: "Inspect the stopped reviewer.",
|
||||
}).returning();
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId, agentId: reviewerId });
|
||||
const other = await seedCompanyAndAgent();
|
||||
const otherRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: otherRunId, companyId: other.companyId, agentId: other.agentId, status: "cancelled" });
|
||||
await db.update(issueRecoveryActions).set({ evidence: { runId: otherRunId } }).where(eq(issueRecoveryActions.id, action!.id));
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ agentId: null });
|
||||
await db.update(issueRecoveryActions).set({ evidence: { runId: "invalid" } }).where(eq(issueRecoveryActions.id, action!.id));
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId: null, agentId: null });
|
||||
});
|
||||
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import { EXECUTION_RECONCILIATION_CAUSES } from "@paperclipai/shared";
|
||||
import { getExecutionBlocker } from "../../../services/execution-blocker.js";
|
||||
import { and, asc, eq, gte, inArray, lte, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
|
|
@ -870,16 +870,10 @@ export function createPostgresRunDispatchAdapter(
|
|||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
if (!issueId) return { issueId: null, decision: { stale: false as const } };
|
||||
const [recovery] = await tx.select({ id: issueRecoveryActions.id, nextAction: issueRecoveryActions.nextAction })
|
||||
.from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
or(inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`),
|
||||
inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]),
|
||||
)).limit(1);
|
||||
const recovery = await getExecutionBlocker(tx, run.companyId, issueId);
|
||||
if (recovery) return { issueId, decision: { stale: true as const,
|
||||
errorCode: "execution_reconciliation_required" as const, reason: recovery.nextAction,
|
||||
details: { issueId, recoveryActionId: recovery.id },
|
||||
details: { issueId, recoveryActionId: recovery.recoveryActionId },
|
||||
} };
|
||||
const facts = await loadStalenessFacts(
|
||||
{
|
||||
|
|
|
|||
|
|
@ -5,10 +5,10 @@ import {
|
|||
issues as issueRows,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
|
||||
import { and, eq, inArray, isNotNull } from "drizzle-orm";
|
||||
import { executionBlockerPredicate } from "../services/execution-blocker.js";
|
||||
import { conflict } from "../errors.js";
|
||||
import {
|
||||
EXECUTION_RECONCILIATION_CAUSES,
|
||||
createIssueTreeHoldSchema,
|
||||
isUuidLike,
|
||||
previewIssueTreeControlSchema,
|
||||
|
|
@ -403,13 +403,7 @@ export function issueTreeControlRoutes(db: Db) {
|
|||
inArray(issueRecoveryActions.sourceIssueId, issueIds),
|
||||
inArray(issueRows.status, RESUME_EXECUTABLE_STATUSES),
|
||||
isNotNull(issueRows.assigneeAgentId),
|
||||
inArray(issueRecoveryActions.cause, [
|
||||
...EXECUTION_RECONCILIATION_CAUSES,
|
||||
]),
|
||||
or(
|
||||
inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
|
||||
),
|
||||
executionBlockerPredicate(),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { issueRecoveryActionReadModel } from "../services/issue-recovery-actions.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { requiresExecutionReconciliation } from "@paperclipai/shared";
|
||||
import { validateExecutionReconciliation, markExecutionReconciliation } from "../services/execution-recovery-resolution.js";
|
||||
import { storedSteeringAcknowledgement, reconcileSteeredIdentity, reserveSteeredIdentity, acceptSteeredIdentity, rejectSteeredIdentity } from "../services/run-identity.js";
|
||||
|
|
@ -7446,6 +7447,7 @@ export function issueRoutes(
|
|||
...(reviewAttention ? { reviewAttention } : {}),
|
||||
productivityReview,
|
||||
successfulRunHandoff: successfulRunHandoffStates.get(issue.id) ?? null,
|
||||
executionBlocker: await getExecutionBlocker(db, issue.companyId, issue.id),
|
||||
scheduledRetry,
|
||||
activeRecoveryAction: revalidatedActiveRecoveryAction,
|
||||
blockedBy: relationsWithRecoveryActions.blockedBy,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { createAdapterExecutionControl, waitForAdapterStop } from "./adapter-execution-control.js";
|
||||
|
||||
afterEach(() => vi.useRealTimers());
|
||||
|
||||
it("does not acknowledge abort until execution and cleanup settle", async () => {
|
||||
const control = createAdapterExecutionControl();
|
||||
const finished = vi.fn();
|
||||
const waiting = waitForAdapterStop(control.settled).then(finished);
|
||||
control.controller.abort();
|
||||
await Promise.resolve();
|
||||
expect(finished).not.toHaveBeenCalled();
|
||||
control.finish();
|
||||
await waiting;
|
||||
expect(finished).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("bounds Stop when an adapter does not settle", async () => {
|
||||
vi.useFakeTimers();
|
||||
const control = createAdapterExecutionControl();
|
||||
const assertion = expect(waitForAdapterStop(control.settled, 1000)).rejects.toThrow("termination has not been verified");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await assertion;
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
/** Live adapter ownership shared by routes and scheduler service instances. */
|
||||
export function createAdapterExecutionControl() {
|
||||
const controller = new AbortController();
|
||||
let finish!: () => void;
|
||||
const settled = new Promise<void>((resolve) => { finish = resolve; });
|
||||
return { controller, settled, finish };
|
||||
}
|
||||
|
||||
export const adapterExecutionControls = new Map<string, ReturnType<typeof createAdapterExecutionControl>>();
|
||||
|
||||
export async function waitForAdapterStop(settled: Promise<void>, timeoutMs = 60_000) {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
settled,
|
||||
new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("Execution is still stopping; termination has not been verified.")), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
import { and, desc, eq, inArray, or, sql } from "drizzle-orm";
|
||||
import { z } from "zod";
|
||||
import { heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { EXECUTION_RECONCILIATION_CAUSES, type ExecutionBlocker } from "@paperclipai/shared";
|
||||
|
||||
/** Resolved recovery bookkeeping can still carry an effective no-replay hold. */
|
||||
export function executionBlockerPredicate() {
|
||||
return and(
|
||||
inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]),
|
||||
or(inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`),
|
||||
);
|
||||
}
|
||||
|
||||
export async function getExecutionBlocker(db: Db, companyId: string, issueId: string): Promise<ExecutionBlocker | null> {
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
executionBlockerPredicate(),
|
||||
)).orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id)).limit(1);
|
||||
if (!action) return null;
|
||||
const parsedRunId = z.string().guid().safeParse(action.evidence.runId ?? action.evidence.sourceRunId);
|
||||
const runId = parsedRunId.success ? parsedRunId.data : null;
|
||||
const [run] = runId ? await db.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId),
|
||||
)).limit(1) : [];
|
||||
|
||||
return {
|
||||
recoveryActionId: action.id,
|
||||
runId,
|
||||
// A stopped reviewer can differ from the task owner who receives the work back.
|
||||
agentId: run?.agentId ?? null,
|
||||
cause: action.cause,
|
||||
nextAction: action.nextAction,
|
||||
};
|
||||
}
|
||||
|
|
@ -1,4 +1,6 @@
|
|||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { legacyExecutionNeedsReconciliation, terminalizeLegacyExecution } from "./legacy-execution-recovery.js";
|
||||
import { adapterExecutionControls, createAdapterExecutionControl, waitForAdapterStop } from "./adapter-execution-control.js";
|
||||
import { executionFailureRetryCount } from "./execution-recovery-attempt.js";
|
||||
import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js";
|
||||
import { issueRecoveryActionService } from "./issue-recovery-actions.js";
|
||||
|
|
@ -17488,6 +17490,7 @@ export function heartbeatService(
|
|||
}
|
||||
|
||||
activeRunExecutions.add(run.id);
|
||||
const executionControl = createAdapterExecutionControl();
|
||||
let runScratch: HeartbeatRunScratch | null = null;
|
||||
let githubLauncherLocation: Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
|
||||
let nativeSessionResumeScheduled = false;
|
||||
|
|
@ -21101,6 +21104,14 @@ export function heartbeatService(
|
|||
);
|
||||
},
|
||||
onDispatch: markDispatchStarted,
|
||||
signal: executionControl.controller.signal,
|
||||
onCancellationReady: async () => {
|
||||
adapterExecutionControls.set(run.id, executionControl);
|
||||
const current = await getRun(run.id);
|
||||
if (!current || isHeartbeatRunTerminalStatus(current.status)) {
|
||||
executionControl.controller.abort(new Error("Run stopped before provider startup"));
|
||||
}
|
||||
},
|
||||
onSpawn: async (meta) => {
|
||||
markDispatchStarted();
|
||||
await persistRunProcessMetadata(run.id, {
|
||||
|
|
@ -21387,6 +21398,8 @@ export function heartbeatService(
|
|||
const latestRun = await getRun(run.id);
|
||||
if (isHeartbeatRunTerminalStatus(latestRun?.status)) {
|
||||
outcome = latestRun.status;
|
||||
} else if (executionControl.controller.signal.aborted) {
|
||||
outcome = "cancelled";
|
||||
} else if (adapterResult.nativeFinalization) {
|
||||
const nativeTerminal =
|
||||
adapterResult.nativeFinalization.terminal.runTerminalState;
|
||||
|
|
@ -21547,7 +21560,7 @@ export function heartbeatService(
|
|||
mergeRunStopMetadataForAgent(agent, outcome, {
|
||||
resultJson: mergeAdapterRecoveryMetadata({
|
||||
resultJson: {
|
||||
...(adapterResult.nativeFinalization
|
||||
...(adapterResult.nativeFinalization || outcome === "cancelled"
|
||||
? parseObject(latestRun?.resultJson)
|
||||
: {}),
|
||||
...parseObject(adapterResult.resultJson),
|
||||
|
|
@ -22110,14 +22123,18 @@ export function heartbeatService(
|
|||
);
|
||||
});
|
||||
|
||||
const failedRunWrite = await setRunStatusIfRunning(run.id, "failed", {
|
||||
const stoppedDuringFailure = executionControl.controller.signal.aborted;
|
||||
const stopSnapshot = stoppedDuringFailure ? await getRun(run.id) : null;
|
||||
const failureOutcome = stoppedDuringFailure ? "cancelled" : "failed";
|
||||
const failedRunWrite = await setRunStatusIfRunning(run.id, failureOutcome, {
|
||||
error: message,
|
||||
errorCode: failureErrorCode,
|
||||
errorCode: stopSnapshot?.errorCode ?? failureErrorCode,
|
||||
finishedAt: new Date(),
|
||||
resultJson: mergeRunStopMetadataForAgent(agent, "failed", {
|
||||
resultJson: mergeRunStopMetadataForAgent(agent, failureOutcome, {
|
||||
errorCode: failureErrorCode,
|
||||
errorMessage: message,
|
||||
resultJson: {
|
||||
...parseObject(stopSnapshot?.resultJson),
|
||||
...(workspaceValidationFailure?.resultJson ?? configurationIncompleteFailure?.resultJson ?? {}),
|
||||
...(!legacyAdapterEntered && run.runtimeMode !== "native" ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } : {}),
|
||||
},
|
||||
|
|
@ -22569,6 +22586,8 @@ export function heartbeatService(
|
|||
}
|
||||
}
|
||||
activeRunExecutions.delete(run.id);
|
||||
executionControl.finish();
|
||||
if (adapterExecutionControls.get(run.id) === executionControl) adapterExecutionControls.delete(run.id);
|
||||
if (
|
||||
!nativeSessionResumeScheduled &&
|
||||
!nativeWorkspaceFinalizeScheduled &&
|
||||
|
|
@ -22732,6 +22751,11 @@ export function heartbeatService(
|
|||
}
|
||||
|
||||
if (legacyExecutionNeedsReconciliation(run)) return { kind: "released" as const };
|
||||
// An operator stop never promotes old queued work by itself. The next
|
||||
// explicit wake adopts those messages atomically when it queues a run.
|
||||
if (run.status === "cancelled" && parseObject(run.resultJson?.executionCancellation).state === "acknowledged") {
|
||||
return { kind: "released" as const };
|
||||
}
|
||||
|
||||
// Native recovery owns its entire incident. The legacy stranded-work
|
||||
// fallback must not create a fresh run and reset a failed native budget,
|
||||
|
|
@ -24969,6 +24993,21 @@ export function heartbeatService(
|
|||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
const pendingComments = !await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id)
|
||||
? await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, issue.companyId),
|
||||
eq(agentWakeupRequests.agentId, agentId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
|
||||
)).orderBy(asc(agentWakeupRequests.requestedAt))
|
||||
: [];
|
||||
const adoptedComments = pendingComments.filter((wake) =>
|
||||
(parseObject(parseObject(wake.payload)[DEFERRED_WAKE_CONTEXT_KEY]).wakeReason ?? wake.reason) === "issue_commented"
|
||||
&& queuedCommentIdsFromWakePayload(wake.payload).length > 0);
|
||||
const adoptedCommentIds = [...new Set([
|
||||
...adoptedComments.flatMap((wake) => queuedCommentIdsFromWakePayload(wake.payload)),
|
||||
...queuedCommentIdsFromRunContext(enrichedContextSnapshot),
|
||||
])];
|
||||
const newRun = await tx
|
||||
.insert(heartbeatRuns)
|
||||
.values({
|
||||
|
|
@ -24979,7 +25018,7 @@ export function heartbeatService(
|
|||
status: "queued",
|
||||
responsibleUserId: await resolveQueuedResponsibleUserId(),
|
||||
wakeupRequestId: wakeupRequest.id,
|
||||
contextSnapshot: enrichedContextSnapshot,
|
||||
contextSnapshot: adoptedComments.length ? withQueuedCommentIdsInRunContext(enrichedContextSnapshot, adoptedCommentIds) : enrichedContextSnapshot,
|
||||
sessionIdBefore: sessionBefore,
|
||||
continuationAttempt,
|
||||
...(reconciledSourceRunId
|
||||
|
|
@ -24989,6 +25028,13 @@ export function heartbeatService(
|
|||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
if (adoptedComments.length) {
|
||||
await tx.update(agentWakeupRequests).set({ status: "coalesced", runId: newRun.id, finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(inArray(agentWakeupRequests.id, adoptedComments.map((wake) => wake.id)));
|
||||
await tx.update(agentWakeupRequests).set({ payload: withQueuedCommentIdsInWakePayload(payload, adoptedCommentIds) })
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
|
||||
}
|
||||
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
|
|
@ -25679,6 +25725,17 @@ export function heartbeatService(
|
|||
: options.resultJson;
|
||||
|
||||
const running = runningProcesses.get(run.id);
|
||||
const control = run.runtimeMode !== "native" ? adapterExecutionControls.get(run.id) : undefined;
|
||||
if (control) {
|
||||
await db.update(heartbeatRuns).set({
|
||||
error: reason,
|
||||
errorCode,
|
||||
resultJson: { ...parseObject(run.resultJson), ...resultJson,
|
||||
...(!running ? { executionCancellation: { state: "requested", requestedAt: new Date().toISOString() } } : {}) },
|
||||
updatedAt: new Date(),
|
||||
}).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running")));
|
||||
control.controller.abort(new Error(reason));
|
||||
}
|
||||
try {
|
||||
await cancelHeartbeatNativeRun({
|
||||
db,
|
||||
|
|
@ -25697,6 +25754,17 @@ export function heartbeatService(
|
|||
runningProcesses.delete(run.id);
|
||||
}
|
||||
|
||||
if (control) {
|
||||
await waitForAdapterStop(control.settled);
|
||||
const stopped = await getRun(run.id);
|
||||
if (stopped && isHeartbeatRunTerminalStatus(stopped.status)) {
|
||||
if (parseObject(stopped.resultJson?.executionCancellation).state !== "acknowledged") {
|
||||
throw conflict("Execution ended, but provider termination could not be verified. Inspect the stopped run before continuing.");
|
||||
}
|
||||
return stopped;
|
||||
}
|
||||
}
|
||||
|
||||
const finishedAt = new Date();
|
||||
const persistedCancellationResult =
|
||||
run.runtimeMode === "native"
|
||||
|
|
@ -25773,6 +25841,10 @@ export function heartbeatService(
|
|||
);
|
||||
|
||||
for (const run of runs) {
|
||||
if (run.runtimeMode !== "native" && adapterExecutionControls.has(run.id)) {
|
||||
await cancelRunInternal(run.id, reason, { errorCode });
|
||||
continue;
|
||||
}
|
||||
if (run.runtimeMode === "native") {
|
||||
await cancelHeartbeatNativeRun({
|
||||
db,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,31 @@
|
|||
import { expect, it } from "vitest";
|
||||
import { legacyExecutionNeedsReconciliation } from "./legacy-execution-recovery.js";
|
||||
|
||||
const stopped = {
|
||||
runtimeMode: "legacy", status: "cancelled", errorCode: "cancelled",
|
||||
resultJson: {
|
||||
executionCancellation: { state: "acknowledged" },
|
||||
executionRecovery: { kind: "interrupted", providerStopped: true, sessionPreserved: true, actionOutcomes: "settled" },
|
||||
},
|
||||
};
|
||||
|
||||
it("allows a confirmed interrupted checkpoint without treating ordinary cancellation as replay permission", () => {
|
||||
expect(legacyExecutionNeedsReconciliation(stopped)).toBe(false);
|
||||
expect(legacyExecutionNeedsReconciliation({ ...stopped, resultJson: {} })).toBe(true);
|
||||
expect(legacyExecutionNeedsReconciliation({ ...stopped, status: "failed" })).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ providerStopped: false }, { sessionPreserved: false }, { actionOutcomes: "unknown" },
|
||||
])("retains the hold for incomplete interruption evidence: %j", (missing) => {
|
||||
expect(legacyExecutionNeedsReconciliation({ ...stopped, resultJson: {
|
||||
...stopped.resultJson,
|
||||
executionRecovery: { ...stopped.resultJson.executionRecovery, ...missing },
|
||||
} })).toBe(true);
|
||||
});
|
||||
|
||||
it("retains the hold until the provider actually acknowledges cancellation", () => {
|
||||
expect(legacyExecutionNeedsReconciliation({ ...stopped, resultJson: {
|
||||
...stopped.resultJson, executionCancellation: { state: "requested" },
|
||||
} })).toBe(true);
|
||||
});
|
||||
|
|
@ -22,6 +22,10 @@ export function legacyExecutionNeedsReconciliation(
|
|||
if (normalizeMaxTurnStopReason(run.resultJson?.stopReason) ?? normalizeMaxTurnStopReason(run.errorCode)) return false;
|
||||
const evidence = run.resultJson?.executionRecovery as
|
||||
Record<string, unknown> | undefined;
|
||||
if (run.status === "cancelled" && evidence?.kind === "interrupted"
|
||||
&& evidence.providerStopped === true && evidence.sessionPreserved === true
|
||||
&& evidence.actionOutcomes === "settled"
|
||||
&& (run.resultJson?.executionCancellation as Record<string, unknown> | undefined)?.state === "acknowledged") return false;
|
||||
// Waiting for a live workspace holder precedes provider execution. It is a
|
||||
// resource wait, not a failed provider attempt or permission to replay work.
|
||||
if (run.status === "cancelled" && run.errorCode === "workspace_busy" &&
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import {
|
|||
ONBOARDING_FIRST_TASK_ORIGIN_KIND,
|
||||
PROVIDER_QUOTA_MONITOR_SERVICE_NAME,
|
||||
ISSUE_DISPOSITION_REPAIR_RETRY_REASON,
|
||||
requiresExecutionReconciliation,
|
||||
type IssueCommentMetadata,
|
||||
type IssueCommentPresentation,
|
||||
} from "@paperclipai/shared";
|
||||
|
|
@ -2289,6 +2290,13 @@ export function recoveryService(
|
|||
continue;
|
||||
}
|
||||
|
||||
// A queued comment or healthy child cannot establish what the stopped
|
||||
// provider already did. Only execution reconciliation can clear this hold.
|
||||
if (requiresExecutionReconciliation(action.cause)) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const [sourceState, healthyChildren, hasNewSourcePath] = await Promise.all([
|
||||
collectDispositionRepairSourceState(db, { issue }),
|
||||
healthyOpenChildIssues(issue),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,101 @@
|
|||
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { test, expect, type APIResponse } from "@playwright/test";
|
||||
|
||||
async function json(response: APIResponse) {
|
||||
const body = await response.text();
|
||||
expect(response.ok(), `${response.url()}: ${response.status()} ${body}`).toBe(true);
|
||||
return JSON.parse(body);
|
||||
}
|
||||
|
||||
for (const { unfinishedWrite, pause } of [{ unfinishedWrite: false, pause: false }, { unfinishedWrite: true, pause: false }, { unfinishedWrite: false, pause: true }]) {
|
||||
test(`embedded ACP Stop: ${unfinishedWrite ? "unknown action stays visibly blocked" : pause ? "composer pause requires Resume before continuation" : "go continues the same session with queued input"}`, async ({ page, request }) => {
|
||||
test.setTimeout(120_000);
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-stop-browser-"));
|
||||
const company = await json(await request.post("/api/companies", { data: { name: `ACP Stop ${Date.now()}` } }));
|
||||
const originalSettings = await json(await request.get("/api/instance/settings/experimental"));
|
||||
try {
|
||||
await json(await request.patch("/api/instance/settings/experimental", { data: { enableClassicTaskInterface: false } }));
|
||||
const owner = await json(await request.post(`/api/companies/${company.id}/agents`, { data: {
|
||||
name: "ACP Stop 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", ...(unfinishedWrite ? { PAPERCLIP_STOP_FIXTURE_TOOL: "write" } : {}) },
|
||||
}, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true } },
|
||||
} }));
|
||||
const issue = await json(await request.post(`/api/companies/${company.id}/issues`, { data: {
|
||||
title: "ACP Stop continuation", status: "backlog", assigneeAgentId: owner.id,
|
||||
} }));
|
||||
await json(await request.patch(`/api/issues/${issue.id}`, { data: { status: "todo" } }));
|
||||
await expect.poll(async () => (await readFile(path.join(root, "prompts"), "utf8").catch(() => "")).trim().split("\n").filter(Boolean).length, { timeout: 45_000 }).toBe(1);
|
||||
const [active] = await json(await request.get(`/api/issues/${issue.id}/live-runs`));
|
||||
expect(active).toBeTruthy();
|
||||
await page.goto(`/${company.issuePrefix}/issues/${issue.identifier}`);
|
||||
const editor = page.getByRole("textbox", { name: "editable markdown" });
|
||||
await editor.fill("List my recent Drive files.");
|
||||
await page.getByRole("button", { name: "Send", exact: true }).click();
|
||||
await expect.poll(async () => JSON.stringify(await json(await request.get(`/api/issues/${issue.id}/queued-comments`))))
|
||||
.toContain("List my recent Drive files.");
|
||||
|
||||
// Run-level Stop leaves the task unpaused; composer Stop additionally pauses the task.
|
||||
let stopped;
|
||||
if (pause) {
|
||||
await page.getByRole("button", { name: "Stop", exact: true }).click();
|
||||
} else {
|
||||
await page.getByRole("button", { name: "Interrupt", exact: true }).click();
|
||||
}
|
||||
await expect.poll(async () => {
|
||||
stopped = await json(await request.get(`/api/heartbeat-runs/${active.id}`));
|
||||
return stopped.resultJson?.executionCancellation?.state;
|
||||
}, { timeout: 30_000 }).toBe("acknowledged");
|
||||
expect(stopped.status).toBe("cancelled");
|
||||
expect(stopped.resultJson.executionCancellation.state).toBe("acknowledged");
|
||||
const writesAtStop = unfinishedWrite ? await readFile(path.join(root, "writes"), "utf8") : null;
|
||||
await page.reload();
|
||||
if (unfinishedWrite) await expect(page.getByText("Work cannot start.", { exact: false })).toBeVisible();
|
||||
await editor.fill("go");
|
||||
await page.getByRole("button", { name: "Send", exact: true }).click();
|
||||
if (pause) {
|
||||
await expect(page.getByText("Task is paused.", { exact: true })).toBeVisible();
|
||||
await expect(page.getByText("Task remains paused. Use Resume work to continue.", { exact: false })).toBeVisible();
|
||||
await expect.poll(async () => (await json(await request.get(`/api/issues/${issue.id}/live-runs`))).length).toBe(0);
|
||||
const pausedPrompts = (await readFile(path.join(root, "prompts"), "utf8")).trim().split("\n").map(line => JSON.parse(line));
|
||||
expect(pausedPrompts).toHaveLength(2);
|
||||
expect(JSON.stringify(pausedPrompts[1])).toContain("execution scope: respond or triage the human comment");
|
||||
expect(await readFile(path.join(root, "completed"), "utf8").catch(() => "")).toBe("");
|
||||
await page.getByRole("button", { name: "Resume work", exact: true }).click();
|
||||
const dialog = page.getByRole("dialog");
|
||||
await dialog.getByRole("checkbox").check();
|
||||
await dialog.getByRole("button", { name: "Resume work", exact: true }).click();
|
||||
}
|
||||
if (unfinishedWrite) {
|
||||
await expect(page.getByText("Couldn't start", { exact: false })).toBeVisible();
|
||||
expect((await json(await request.get(`/api/issues/${issue.id}`))).executionBlocker).toBeTruthy();
|
||||
await page.waitForTimeout(1000);
|
||||
expect(await readFile(path.join(root, "writes"), "utf8")).toBe(writesAtStop);
|
||||
expect((await readFile(path.join(root, "prompts"), "utf8")).trim().split("\n")).toHaveLength(1);
|
||||
} else {
|
||||
await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible({ timeout: 30_000 });
|
||||
await expect.poll(async () => (await json(await request.get(`/api/issues/${issue.id}/live-runs`))).length).toBe(0);
|
||||
const prompts = (await readFile(path.join(root, "prompts"), "utf8")).trim().split("\n").map(line => JSON.parse(line));
|
||||
expect(prompts).toHaveLength(pause ? 3 : 2);
|
||||
expect(new Set(prompts.map(prompt => prompt.sessionId)).size).toBe(1);
|
||||
// Paused conversation already delivered the request into this same
|
||||
// provider session; Resume legitimately sends only its next delta.
|
||||
const continuationPrompts = pause ? prompts.slice(1) : [prompts.at(-1)];
|
||||
expect(JSON.stringify(continuationPrompts)).toContain("List my recent Drive files.");
|
||||
expect(JSON.stringify(continuationPrompts)).toContain("go");
|
||||
expect(await readFile(path.join(root, "completed"), "utf8")).toBe("follow-up\n");
|
||||
const completedIssue = await json(await request.get(`/api/issues/${issue.id}`));
|
||||
expect(completedIssue.executionBlocker).toBeNull();
|
||||
expect(completedIssue.status).toBe("done");
|
||||
}
|
||||
await expect(page.getByRole("dialog")).toHaveCount(0);
|
||||
} finally {
|
||||
await request.patch(`/api/companies/${company.id}`, { data: { status: "archived" } });
|
||||
await request.patch("/api/instance/settings/experimental", { data: { enableClassicTaskInterface: originalSettings.enableClassicTaskInterface } });
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
// @vitest-environment jsdom
|
||||
|
||||
import { act } from "react";
|
||||
import { flushSync } from "react-dom";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
|
@ -28,18 +29,11 @@ function Harness({ onRoute }: { onRoute: boolean }) {
|
|||
);
|
||||
}
|
||||
|
||||
async function flushReact() {
|
||||
await Promise.resolve();
|
||||
await new Promise((resolve) => window.setTimeout(resolve, 0));
|
||||
flushSync(() => {});
|
||||
}
|
||||
|
||||
async function render(onRoute: boolean): Promise<{ root: Root; host: HTMLDivElement }> {
|
||||
const host = document.createElement("div");
|
||||
document.body.appendChild(host);
|
||||
const root = createRoot(host);
|
||||
flushSync(() => root.render(<Harness onRoute={onRoute} />));
|
||||
await flushReact();
|
||||
await act(async () => root.render(<Harness onRoute={onRoute} />));
|
||||
return { root, host };
|
||||
}
|
||||
|
||||
|
|
@ -71,9 +65,9 @@ describe("RequestCollapsedSidebar", () => {
|
|||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
afterEach(async () => {
|
||||
if (active) {
|
||||
flushSync(() => active!.root.unmount());
|
||||
await act(async () => active!.root.unmount());
|
||||
active.host.remove();
|
||||
active = null;
|
||||
}
|
||||
|
|
@ -100,8 +94,7 @@ describe("RequestCollapsedSidebar", () => {
|
|||
expect(capturedValue?.collapsed).toBe(false);
|
||||
|
||||
// Navigate away: the route (and its <RequestCollapsedSidebar/>) unmounts.
|
||||
flushSync(() => active!.root.render(<Harness onRoute={false} />));
|
||||
await flushReact();
|
||||
await act(async () => active!.root.render(<Harness onRoute={false} />));
|
||||
expect(capturedValue?.routeRequestsCollapsed).toBe(false);
|
||||
expect(capturedValue?.collapsed).toBe(false);
|
||||
});
|
||||
|
|
@ -111,8 +104,7 @@ describe("RequestCollapsedSidebar", () => {
|
|||
flushSync(() => capturedValue?.setCollapsed(true));
|
||||
expect(localStorage.getItem("paperclip.sidebar.collapsed")).toBeNull();
|
||||
|
||||
flushSync(() => active!.root.render(<Harness onRoute={false} />));
|
||||
await flushReact();
|
||||
await act(async () => active!.root.render(<Harness onRoute={false} />));
|
||||
expect(capturedValue?.routeRequestsCollapsed).toBe(false);
|
||||
expect(capturedValue?.collapsed).toBe(false);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1026,6 +1026,19 @@ describe("TaskChatThread runtime transcript selection", () => {
|
|||
expect(onRetryFailedRun).toHaveBeenCalledWith("native-failed");
|
||||
});
|
||||
|
||||
it("explains a legacy run prevented from starting by a reconciliation hold", () => {
|
||||
render(<TaskChatThread comments={[]} onAdd={async () => {}} linkedRuns={[{
|
||||
runId: "blocked-legacy", runtimeMode: "legacy", status: "cancelled",
|
||||
errorCode: "execution_reconciliation_required", agentId: "agent-1", agentName: "Runner",
|
||||
adapterType: "claude_local", createdAt: "2026-08-25T18:00:00.000Z",
|
||||
startedAt: null, finishedAt: "2026-08-25T18:00:00.012Z",
|
||||
}]} />);
|
||||
expect(container.textContent).toContain("Couldn't start");
|
||||
expect(container.textContent).not.toContain("No user-facing response");
|
||||
expect(container.textContent).not.toContain("Run completed");
|
||||
expect(container.querySelector(".text-destructive")).toBeNull();
|
||||
});
|
||||
|
||||
it("shows cancellation after native progress without offering a retry", () => {
|
||||
nativeTranscriptState.transcriptByRun.set("native-cancelled", [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -1529,7 +1529,7 @@ export function TaskChatThread(props: TaskChatThreadProps) {
|
|||
settledRunIds.add(source.id);
|
||||
} else if (
|
||||
!sourceIsPaperclipRunner &&
|
||||
(source.status === "failed" || source.status === "timed_out")
|
||||
(source.status === "failed" || source.status === "timed_out" || source.status === "cancelled")
|
||||
) {
|
||||
settledRunIds.add(source.id);
|
||||
const code = meta?.errorCode ?? "native_runner_process_exited";
|
||||
|
|
@ -1537,7 +1537,11 @@ export function TaskChatThread(props: TaskChatThreadProps) {
|
|||
? "Retry scheduled automatically."
|
||||
: "You can retry this message now.";
|
||||
const detail =
|
||||
code === "provider_frame_too_large"
|
||||
source.status === "cancelled"
|
||||
? code === "execution_reconciliation_required"
|
||||
? "The previous execution must be checked before this task can continue. Your message is preserved. View the stopped run for details."
|
||||
: "Execution was stopped before returning an answer."
|
||||
: code === "provider_frame_too_large"
|
||||
? `Provider output exceeded the safe limit. ${retryDetail}`
|
||||
: `The runner stopped before returning an answer (${code}). ${retryDetail}`;
|
||||
const id = `${source.id}:failure`;
|
||||
|
|
@ -1549,7 +1553,8 @@ export function TaskChatThread(props: TaskChatThreadProps) {
|
|||
id,
|
||||
kind: "marker",
|
||||
variant: "interrupted",
|
||||
label: "Run failed",
|
||||
label: source.status === "cancelled" ? (meta?.startedAt ? "Stopped" : "Couldn't start") : "Run failed",
|
||||
tone: source.status === "cancelled" ? "neutral" : "error",
|
||||
detail,
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// @vitest-environment jsdom
|
||||
|
||||
import { StrictMode, useState, type ReactElement } from "react";
|
||||
import { act, StrictMode, useState, type ReactElement } from "react";
|
||||
import { flushSync } from "react-dom";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
|
@ -202,7 +202,9 @@ function render(ui: ReactElement) {
|
|||
}
|
||||
|
||||
async function flushAsync() {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
await act(async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
});
|
||||
}
|
||||
|
||||
function editable() {
|
||||
|
|
|
|||
|
|
@ -2425,6 +2425,8 @@ span.paperclip-mention-chip[data-mention-kind="external-object"] {
|
|||
for the full site list).
|
||||
*/
|
||||
:root {
|
||||
--sz-execution-blocker-inline: calc(var(--spacing) * 4);
|
||||
--sz-execution-blocker-block: calc(var(--spacing) * 2);
|
||||
/* Ten single-line repository rows, including the gaps between them. */
|
||||
--sz-github-repository-list: calc(10lh + 9 * var(--spacing) * 2);
|
||||
--sz-320px: 320px; /* Extracted from ui/src/components/ActiveAgentsPanel.tsx (h-[320px]). */
|
||||
|
|
|
|||
|
|
@ -7,6 +7,19 @@ function run(id: string, status: HeartbeatRun["status"]) {
|
|||
}
|
||||
afterEach(() => vi.useRealTimers());
|
||||
describe("stop confirmation", () => {
|
||||
it("waits for embedded ACP acknowledgment even after terminal status", async () => {
|
||||
vi.useFakeTimers();
|
||||
const getRun = vi.fn()
|
||||
.mockResolvedValueOnce({ ...run("acp", "cancelled"), resultJson: { executionCancellation: { state: "requested" } } })
|
||||
.mockResolvedValueOnce({ ...run("acp", "cancelled"), resultJson: { executionCancellation: { state: "acknowledged" } } });
|
||||
const finished = vi.fn();
|
||||
const result = waitForStoppedRuns(["acp"], { getRun }).then(finished);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(finished).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
await result;
|
||||
expect(finished).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("waits for native cancellation acknowledgment after the run becomes terminal", async () => {
|
||||
vi.useFakeTimers();
|
||||
const native = { ...run("native", "cancelled"), runtimeMode: "native" };
|
||||
|
|
|
|||
|
|
@ -37,6 +37,9 @@ export async function waitForStoppedRuns(
|
|||
remaining = states
|
||||
.filter((run) => {
|
||||
if (LIVE_STATUSES.has(run.status)) return true;
|
||||
const adapterCancellation = run.resultJson?.executionCancellation;
|
||||
if (adapterCancellation && typeof adapterCancellation === "object"
|
||||
&& "state" in adapterCancellation && adapterCancellation.state !== "acknowledged") return true;
|
||||
if (!("runtimeMode" in run) || run.runtimeMode !== "native" || run.status !== "cancelled")
|
||||
return false;
|
||||
const cancellation = run.resultJson?.nativeCancellation;
|
||||
|
|
|
|||
|
|
@ -7409,6 +7409,16 @@ export function IssueDetail() {
|
|||
: undefined
|
||||
}
|
||||
>
|
||||
{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>
|
||||
)}
|
||||
{resolvedDetailTab === "chat" ? (
|
||||
<IssueDetailChatTab
|
||||
threadHeader={taskChatThreadHeader}
|
||||
|
|
|
|||
Loading…
Reference in New Issue