diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 677a14969a..180bd820b6 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -834,6 +834,11 @@ apply. Stream closure without a turn terminal is not proof of success. Event replay uses the existing source receipts and never repeats provider work merely to recover recorded output. +If runnerd synthesizes a result when the provider stops, it publishes that result +before the provider-turn terminal and publishes the run terminal last. The +adapter can therefore retain the result while the matching turn still has +authority. A late result must not reopen an already finalized turn. + Routine task completion and human-input requests must work under Conservative runner permissions. The isolated Claude runtime grants only the narrow task tools on the runner-owned bridge; it does not change general tool permissions. @@ -854,6 +859,11 @@ neutral even if teardown releases the run lease or returns no semantic result. **Pause work** separately controls future execution. A crash preventing progress is **Blocked**; **In Review** requires a concrete human decision. +Subtree pause and cancel record the authenticated board actor on each run they +interrupt. A verified native stop must not become an unexplained failure simply +because it came from a subtree action. The explicit pause hold still prevents +future execution until Resume, and missing stop proof still blocks continuation. + ### Provider continuity and bounded finalization A permanently unusable native runner session may be replaced only with evidence that its predecessor is stopped and fenced, completed results and workspace state are preserved, required task history is available, and pending effects have been reconciled. A provider-native shell command or external write without a reliable outcome receipt is unknown. Unknown effects, integrity failures, and unverified process ownership never authorize speculative replay. Once automatic recovery is ruled out, Paperclip selects a conservative default: preserve recorded work, stop the affected task, and retain a durable no-replay hold. Unknown action outcomes remain unknown. No reconciliation form or user diagnosis is required. @@ -1158,3 +1168,17 @@ and final dispatch gates. Queued and final native replacement dispatch also re-read dependency readiness, since new dependencies need not change the displayed task status. Old blocked rows without a receipt remain held; no historical status backfill is performed. + +### Queued input after a native Stop + +A run-only Stop ends the current response. It does not discard queued user +messages or require a recovery incident. After the controller releases ownership +and the old local process or remote environment has a verified stop record, +Paperclip submits saved input through normal task admission, once, with the +original user's authority. Pauses, task ownership, budgets, approvals, and +execution recovery holds still apply. Unconfirmed cleanup does not start work. + +The active session advertises steering only when its driver supports it. A +transport method that rejects steering does not grant that capability. The +queued-message control remains mounted until the server accepts a steer request, +so a rejected last-row action keeps its message and visible error. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs index 2b8212c794..2c9dedeb26 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs @@ -32,7 +32,9 @@ use crate::provider_bridge::{ authorized_tool_catalog_digest, semantic_value_digest, AuthorizedTool, AuthorizedToolSet, PendingToolCall, ToolResult, MAX_PENDING_CALLS, TOOL_SET_SCHEMA, }; -use crate::provider_events::{normalize_codex_notification, NormalizedProviderEvent}; +use crate::provider_events::{ + normalize_codex_notification, with_terminal_outcome, NormalizedProviderEvent, +}; pub const MANAGED_PROVIDER_STATE_FILE: &str = "managed-provider-state.json"; const MANAGED_PROVIDER_STATE_SCHEMA: &str = "paperclip.runner.managed-provider-state.v1"; @@ -765,7 +767,7 @@ impl ManagedProviderCommandExecutor { .expect("managed state remains present during recovery"); let prior_turn = state.active_turn_id.take(); state.lifecycle = "failed".to_owned(); - state.push(NormalizedProviderEvent { + let provider_terminal = NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -775,9 +777,9 @@ impl ManagedProviderCommandExecutor { "providerTerminalObserved": false, "code": "agentcore_active_turn_recovery_requires_review", }), - })?; + }; let terminal = terminal_events(state, "turn.failed"); - for event in terminal { + for event in with_terminal_outcome(vec![provider_terminal], terminal) { state.push(event)?; } self.save_state()?; @@ -1483,7 +1485,7 @@ impl ManagedProviderCommandExecutor { })?; let prior_turn = state.active_turn_id.take(); state.lifecycle = "session_open".to_owned(); - state.push(NormalizedProviderEvent { + let provider_terminal = NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -1493,8 +1495,11 @@ impl ManagedProviderCommandExecutor { "stopReason": params.get("stopReason"), "code": "provider_limit_reached", }), - })?; - for event in terminal_events(state, "turn.failed") { + }; + for event in with_terminal_outcome( + vec![provider_terminal], + terminal_events(state, "turn.failed"), + ) { state.push(event)?; } return Ok(()); @@ -1532,15 +1537,14 @@ impl ManagedProviderCommandExecutor { ); } } - for event in normalized { - state.push(event)?; - } if let Some(event_type) = terminal { state.active_turn_id = None; state.lifecycle = "session_open".to_owned(); - for event in terminal_events(state, &event_type) { - state.push(event)?; - } + normalized = + with_terminal_outcome(normalized, terminal_events(state, &event_type)); + } + for event in normalized { + state.push(event)?; } } ProviderEvent::SemanticResult { result, .. } => { @@ -1594,7 +1598,7 @@ impl ManagedProviderCommandExecutor { .expect("managed state exists while failing provider"); let active = state.active_turn_id.take(); state.lifecycle = "failed".to_owned(); - state.push(NormalizedProviderEvent { + let mut failures = vec![NormalizedProviderEvent { event_type: "session.failed".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -1602,9 +1606,10 @@ impl ManagedProviderCommandExecutor { "code": "managed_provider_failed", "message": message, }), - })?; + }]; + let mut outcome = Vec::new(); if active.is_some() { - state.push(NormalizedProviderEvent { + failures.push(NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -1613,10 +1618,11 @@ impl ManagedProviderCommandExecutor { "status": "failed", "code": "managed_provider_failed", }), - })?; - for event in terminal_events(state, "turn.failed") { - state.push(event)?; - } + }); + outcome = terminal_events(state, "turn.failed"); + } + for event in with_terminal_outcome(failures, outcome) { + state.push(event)?; } self.save_state() } @@ -2442,6 +2448,76 @@ mod tests { }) } + fn managed_failure_event_types(recovery: bool) -> Vec { + let directory = std::env::temp_dir().join(format!( + "paperclip-managed-terminal-test-{}", + Uuid::new_v4() + )); + fs::create_dir_all(&directory).unwrap(); + #[cfg(unix)] + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).unwrap(); + let config = test_config(&directory); + let mut executor = ManagedProviderCommandExecutor::with_runner_config(&directory, &config); + let mut payload = agentcore_prepare_payload(); + payload["completionContract"] = json!({ + "revision": "contract-1", + "criterionIds": ["requested-work"], + }); + executor.prepare(&payload).unwrap(); + let state = executor.state.as_mut().unwrap(); + state.lifecycle = "turn_active".to_owned(); + state.active_turn_id = Some("provider-turn-1".to_owned()); + state.provider_session_id = Some("provider-session-1".to_owned()); + state.provider_usage = Some(json!({ + "inputTokens": 0, + "outputTokens": 0, + "cacheReadInputTokens": 0, + "cacheWriteInputTokens": 0, + "requestCount": 1, + "estimatedCostUsd": 0.0, + "costSource": "paperclip_estimate", + })); + state.pending_events.clear(); + if recovery { + executor.restore_provider_if_needed().unwrap(); + } else { + executor + .fail_provider("synthetic provider crash".to_owned()) + .unwrap(); + } + let events = executor + .state + .as_ref() + .unwrap() + .pending_events + .iter() + .map(|event| event.event_type.clone()) + .collect(); + fs::remove_dir_all(directory).unwrap(); + events + } + + #[test] + fn managed_crash_preserves_result_before_failure_closes_authority() { + assert_eq!( + managed_failure_event_types(false), + vec![ + "run.result.proposed", + "session.failed", + "turn.failed", + "run.terminal", + ] + ); + } + + #[test] + fn managed_active_turn_recovery_preserves_result_before_failure() { + assert_eq!( + managed_failure_event_types(true), + vec!["run.result.proposed", "turn.failed", "run.terminal",] + ); + } + #[test] fn claude_usage_maps_nested_cache_creation_token_buckets() { let descriptor = ManagedProviderDescriptor::ClaudeManaged(ClaudeManagedProviderConfig { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index fe6301308f..6f9a9533fe 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -28,7 +28,8 @@ use crate::provider_bridge::{ TOOL_SET_SCHEMA, }; use crate::provider_events::{ - normalize_codex_notification, normalized_codex_terminal_event_type, NormalizedProviderEvent, + normalize_codex_notification, normalized_codex_terminal_event_type, with_terminal_outcome, + NormalizedProviderEvent, }; use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; @@ -2093,7 +2094,7 @@ impl CodexCommandExecutor { // trustworthy success notification to replay. Terminate it // conservatively so the controller cannot wait forever or // mistake an unknown outcome for success. - state.push_terminal_event(NormalizedProviderEvent { + let provider_terminal = NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, payload: json!({ @@ -2102,11 +2103,15 @@ impl CodexCommandExecutor { "status": "failed", "providerTerminalObserved": false, }), - })?; - state.extend_terminal_events(terminal_events( + }; + let outcome = terminal_events( state, "turn.failed", state.goal.as_ref().map(|goal| goal.status.as_str()), + ); + state.extend_terminal_events(with_terminal_outcome( + vec![provider_terminal], + outcome, ))?; } } else { @@ -3533,7 +3538,7 @@ impl CodexCommandExecutor { } else { "Codex" }; - state.push_terminal_event(NormalizedProviderEvent { + let provider_terminal = NormalizedProviderEvent { event_type: terminal_event_type.to_owned(), priority: EventPriority::P0, payload: json!({ @@ -3552,9 +3557,9 @@ impl CodexCommandExecutor { "providerTerminalObserved": false, "providerShutdownFailed": provider_shutdown_failed, }), - })?; + }; let terminal = terminal_events(state, terminal_event_type, None); - state.extend_terminal_events(terminal)?; + state.extend_terminal_events(with_terminal_outcome(vec![provider_terminal], terminal))?; self.save_state() } @@ -3919,14 +3924,18 @@ impl CodexCommandExecutor { priority: EventPriority::P0, payload: diagnostic.clone(), })?; - state.push_terminal_event(NormalizedProviderEvent { + let provider_terminal = NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, payload: json!({ "provider": state.config.provider, "status": "failed", "code": diagnostic["code"], "recoverable": false, "message": diagnostic["message"], "error": diagnostic }), - })?; - state.extend_terminal_events(terminal_events(state, "turn.failed", None))?; + }; + let outcome = terminal_events(state, "turn.failed", None); + state.extend_terminal_events(with_terminal_outcome( + vec![provider_terminal], + outcome, + ))?; // Commit the authoritative failure before best-effort provider cleanup. self.save_state()?; if let Some(mut provider) = self.provider.take() { @@ -4197,8 +4206,10 @@ impl CodexCommandExecutor { } } let trace_first_event_sequence = state.next_provider_event_seq; - if terminal_event_type.is_some() { - state.extend_terminal_events(normalized)?; + if let Some(ref event_type) = terminal_event_type { + let goal_status = state.goal.as_ref().map(|goal| goal.status.as_str()); + let outcome = terminal_events(state, event_type, goal_status); + state.extend_terminal_events(with_terminal_outcome(normalized, outcome))?; } else if receipt_limit_terminal_poll { for event in normalized { state.push_receipt_limit_cleanup_event(event)?; @@ -4207,14 +4218,6 @@ impl CodexCommandExecutor { state.extend_events(normalized)?; } let trace_last_event_sequence = state.next_provider_event_seq; - if let Some(event_type) = terminal_event_type { - let goal_status = state.goal.as_ref().map(|goal| goal.status.as_str()); - state.extend_terminal_events(terminal_events( - state, - &event_type, - goal_status, - ))?; - } let trace_emitted_event_ids = identity .as_ref() .map(|identity| { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index 317f4980ad..4c5e313ba0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -19,6 +19,24 @@ pub struct NormalizedProviderEvent { pub payload: Value, } +/// The facade closes provider-turn authority on its terminal notification. +/// Commit any result synthesized at that boundary before the turn terminal, +/// then publish the run terminal. Already committed semantic results and +/// active goals supply no new result and keep their existing event order. +pub(crate) fn with_terminal_outcome( + provider_events: Vec, + outcome_events: Vec, +) -> Vec { + let (results, terminals): (Vec<_>, Vec<_>) = outcome_events + .into_iter() + .partition(|event| event.event_type == "run.result.proposed"); + results + .into_iter() + .chain(provider_events) + .chain(terminals) + .collect() +} + pub(crate) fn normalized_codex_terminal_event_type( method: &str, params: &Value, diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index 1a46ea7d73..b8a1ae0211 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -97,6 +97,25 @@ const driver: HarnessDriver = { }; describe("HarnessDriverBackend", () => { + it.each([false, true])("honors driver steering support even when the transport exposes a steer method (%s)", async supported => { + const steer = vi.fn(async () => { throw new Error("provider does not support steering"); }); + const session = Object.assign(new FakeHarnessSession(), { steer }); + const backend = new HarnessDriverBackend({ ...driver, + descriptor: async () => ({ ...(await driver.descriptor()), capabilities: { + ...(await driver.descriptor()).capabilities, steering: supported, + } }), + openSession: async () => session, + recoverSession: async () => ({ recovered: true, session }), + }); + const opened = await backend.openSession({ identity: { + runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1", + } }); + expect((await opened.capabilities()).steering).toBe(supported); + const recovered = await backend.recoverSession(await opened.snapshot(), { signal: new AbortController().signal }); + expect(recovered.recovered).toBe(true); + expect((await recovered.session!.capabilities()).steering).toBe(supported); + expect(steer).not.toHaveBeenCalled(); + }); it("retains Codex accounting and startup state through a serialized native checkpoint", async () => { const fields = { workingDirectory: "/workspace/selected", codexUsageBaseline: { baseline: { inputTokens: 100, outputTokens: 20 }, diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index 403d0b9954..3669e58c43 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -73,7 +73,7 @@ export class HarnessDriverBackend implements NativeSessionBackend { .catch(() => undefined); throw error; } - return new HarnessNativeSession(input, session); + return new HarnessNativeSession(input, session, undefined, (await this.#driver.descriptor()).capabilities.steering); } async recoverSession( @@ -183,6 +183,7 @@ export class HarnessDriverBackend implements NativeSessionBackend { { identity: snapshot.identity }, recovered.session, recoveredTerminal, + (await this.#driver.descriptor()).capabilities.steering, ), }; } @@ -430,11 +431,15 @@ class HarnessNativeSession implements NativeSession { } } + readonly #steeringSupported: boolean; + constructor( input: OpenNativeSessionInput, session: HarnessSession, terminal?: PrpTerminalState | null, + steeringSupported = false, ) { + this.#steeringSupported = steeringSupported; this.#input = structuredClone(input); this.#session = session; this.#terminal = terminal === undefined ? null : structuredClone(terminal); @@ -448,7 +453,7 @@ class HarnessNativeSession implements NativeSession { return { resume: true, typedEvents: true, - steering: this.#session.steer !== undefined, + steering: this.#steeringSupported && this.#session.steer !== undefined, interruption: this.#session.interrupt !== undefined, structuredResult: true, read: this.#session.read !== undefined, @@ -678,7 +683,7 @@ class HarnessNativeSession implements NativeSession { correlationId?: string; }) { this.#assertProtocolIntegrity(); - if (this.#session.steer === undefined) + if (!this.#steeringSupported || this.#session.steer === undefined) throw new Error("steering is unavailable"); return this.#withProtocolIntegrity(() => this.#session.steer!(input)); } diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 054d17a0a7..c559d58218 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -5317,6 +5317,93 @@ it.each(["not_suspended", "wrong_identity"] as const)( 10_000, ); +it.each([ + { + label: "interrupted", + args: ["--hold-turn"], + terminal: "turn.interrupted", + disposition: "needs_review", + }, + { + label: "failed", + args: ["--fail-turn-immediately"], + terminal: "turn.failed", + disposition: "needs_review", + }, + { + label: "completed", + args: [], + terminal: "turn.completed", + disposition: "done", + }, +])("retains a $label runner's result before closing its provider turn", async ({ + label, args, terminal, disposition, +}) => { + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-stop-result-")); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory, ...args), + stateDirectory, + }); + const driver = new CodexAppServerDriver({ + taskEnvelope: createCodexTaskEnvelope({ + objective: "Stop and retain unfinished work.", + }), + environment: { + PATH: process.env.PATH, + HOME: join(tmpdir(), "runnerd-stop-host-home"), + PAPERCLIP_WORKSPACE_CWD: stateDirectory, + }, + approvalPolicy: "never", + transportFactory: () => bundle.transport, + }); + const session = await driver.openSession({ + runId: "run-stop-result", + normalizedSessionId: "session-stop-result", + workingDirectory: stateDirectory, + }); + try { + const turn = await session.startTurn({ + message: { role: "user", text: "Keep working until interrupted." }, + }); + if (label === "interrupted") { + await session.interrupt({ + turnId: turn.turnId, + reason: "Stopped by the user", + }); + } + const events: PrpEvent[] = []; + for await (const event of session.events()) { + events.push(event); + if (event.eventType === "session.failed" || event.eventType === terminal) { + break; + } + } + // Let already committed durable suffix events reach the facade as well. + await bundle.transport.request("thread/read", {}); + const snapshot = await session.snapshot(); + expect(events.some((event) => event.eventType === "session.failed")).toBe(false); + expect(events.map((event) => event.eventType)).toContain("run.result.proposed"); + expect(events.at(-1)?.eventType).toBe(terminal); + expect(snapshot).toMatchObject({ + activeTurnId: null, + semanticResult: { + turnId: turn.turnId, + result: { reportedWorkDisposition: disposition }, + }, + }); + } finally { + await session.close().catch(() => undefined); + await rm(stateDirectory, { + recursive: true, + force: true, + maxRetries: 5, + retryDelay: 25, + }); + } +}, 30_000); + it("binds an immediately failed durable turn before exposing its terminal", async () => { const stateDirectory = await mkdtemp( join(tmpdir(), "runnerd-fast-terminal-"), diff --git a/server/src/__tests__/issue-tree-control-routes.test.ts b/server/src/__tests__/issue-tree-control-routes.test.ts index 3b22b9d985..babfff8dc3 100644 --- a/server/src/__tests__/issue-tree-control-routes.test.ts +++ b/server/src/__tests__/issue-tree-control-routes.test.ts @@ -340,7 +340,11 @@ describe("issue tree control routes", () => { .send({ mode: "pause", reason: "pause subtree" }); expect(res.status).toBe(201); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("44444444-4444-4444-8444-444444444444"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + "44444444-4444-4444-8444-444444444444", + "Cancelled by a board operator's subtree pause", + { resultJson: { cancelledByActorType: "user", cancelledByUserId: "user-1" } }, + ); expect(mockTreeControlService.cancelUnclaimedWakeupsForTree).toHaveBeenCalledWith( "company-2", "11111111-1111-4111-8111-111111111111", @@ -440,7 +444,11 @@ describe("issue tree control routes", () => { .send({ mode: "cancel", reason: "cancel subtree" }); expect(res.status).toBe(201); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("44444444-4444-4444-8444-444444444444"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + "44444444-4444-4444-8444-444444444444", + "Cancelled by a board operator's subtree cancel", + { resultJson: { cancelledByActorType: "user", cancelledByUserId: "user-1" } }, + ); expect(mockTreeControlService.cancelIssueStatusesForHold).toHaveBeenCalledWith( "company-2", "11111111-1111-4111-8111-111111111111", diff --git a/server/src/routes/issue-tree-control.ts b/server/src/routes/issue-tree-control.ts index 5a2acae573..c8c4327a4b 100644 --- a/server/src/routes/issue-tree-control.ts +++ b/server/src/routes/issue-tree-control.ts @@ -124,7 +124,19 @@ export function issueTreeControlRoutes(db: Db) { for (const heartbeatRunId of interruptedRunIds) { const cancellationTask = (async () => { try { - await heartbeat.cancelRun(heartbeatRunId); + // This board-only operation is an intentional interruption, just + // like composer Stop. Preserve its actor so verified native stops + // do not manufacture recovery incidents while the hold is active. + await heartbeat.cancelRun( + heartbeatRunId, + `Cancelled by a board operator's subtree ${result.hold.mode}`, + { + resultJson: { + cancelledByActorType: "user", + cancelledByUserId: req.actor.userId ?? null, + }, + }, + ); await logActivity(db, { companyId: root.companyId, actorType: actor.actorType, diff --git a/server/src/services/acknowledged-native-stop.ts b/server/src/services/acknowledged-native-stop.ts index 17b55693ac..37efd1c84e 100644 --- a/server/src/services/acknowledged-native-stop.ts +++ b/server/src/services/acknowledged-native-stop.ts @@ -1,3 +1,8 @@ +import { and, eq } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, type Db } from "@paperclipai/db"; +import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; +import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; + /** A server-recorded run-only Stop must not manufacture a recovery incident. */ export function hasAcknowledgedNativeStopIntent(run: { id: string; companyId: string; status: string; nativeIssueId: string | null; @@ -17,3 +22,20 @@ export function hasAcknowledgedNativeStopIntent(run: { export function isAcknowledgedNativeStop(run: Parameters[0]): boolean { return run.status === "cancelled" && hasAcknowledgedNativeStopIntent(run); } + +/** A Stop acknowledgement alone does not prove provider cleanup completed. */ +export async function acknowledgedNativeStopExecutionHasStopped(db: Db, run: typeof heartbeatRuns.$inferSelect) { + if (!isAcknowledgedNativeStop(run)) return false; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + // Provider PIDs are meaningful only on their own host. + if (leases.some(lease => lease.provider !== "local")) return leases.every(hasRemoteTerminationReceipt); + if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return false; + if (!run.processPid && !run.processGroupId) return hasNativeLocalProcessStop(db, run.companyId, run.id); + const absent = (pid: number) => { + try { process.kill(pid, 0); return false; } + catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; } + }; + return (!run.processPid || absent(run.processPid)) && (!run.processGroupId || absent(-run.processGroupId)); +} diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 6a7176bf01..7c235d6fc5 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -42,6 +42,72 @@ const support = await getEmbeddedPostgresTestSupport(); actorType: "user", actorId: "board", reason: "issue_commented" }; } type Fixture = Awaited>; + it.each(["ready", "unacknowledged", "pause", "recovery", "controller", "process_running", "identity_missing", "remote_pending", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"])("delivers a saved native message after run-only Stop exactly once (%s)", async gate => { + const f = await seed(); + if (gate !== "recovery") await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issueId)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureDetail: null, leaseOwner: gate === "controller" ? "still-cleaning" : null }) + .where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:00:00Z") }) + .where(eq(issueComments.id, f.commentId)); + const [source] = await db.update(heartbeatRuns).set({ status: "cancelled", + processPid: gate === "process_running" ? process.pid : gate === "identity_missing" ? null : 999999999, resultJson: { + cancelledByActorType: "user", cancelledByUserId: "board", nativeCancellation: { + schema: "paperclip.native-cancellation.v1", runId: f.sourceRunId, companyId: f.companyId, + issueId: f.issueId, scope: "run", reasonCode: "cancellation_run_only", dispatched: true, + dispatchState: gate === "unacknowledged" ? "requested" : "acknowledged", + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID(), + }, + } }).where(eq(heartbeatRuns.id, f.sourceRunId)).returning(); + // Occupy the agent slot: admission is real, but no provider should launch. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + if (gate === "pause") { + const holdId = randomUUID(); + await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" }); + await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "in_progress" }); + } + let queuedIds = [f.commentId]; + let expectedIds = queuedIds; + if (["first_delivered", "last_delivered", "mixed_authors"].includes(gate)) { + const secondId = randomUUID(); + await db.insert(issueComments).values({ id: secondId, companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: gate === "mixed_authors" ? "second-author" : "board", + body: "Keep the earlier direction too.", createdAt: new Date("2026-09-11T09:01:00Z") }); + queuedIds = [f.commentId, secondId]; + expectedIds = gate === "first_delivered" ? [secondId] : gate === "last_delivered" ? [f.commentId] : queuedIds; + if (gate !== "mixed_authors") await db.update(heartbeatRuns).set({ startedAt: new Date("2026-09-11T09:02:00Z"), + contextSnapshot: { issueId: f.issueId, wakeCommentIds: gate === "first_delivered" ? [f.commentId] : [secondId] }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + } + const queueId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", triggerDetail: "system", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "board", payload: { + issueId: f.issueId, commentId: queuedIds.at(-1), _paperclipWakeContext: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: queuedIds.at(-1), wakeCommentIds: queuedIds }, + }, + }); + if (gate.startsWith("remote_")) { + const identity = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId, + provider: "daytona", providerLeaseId: "owned-sandbox" }; + await db.insert(environmentLeases).values({ ...identity, status: gate === "remote_stopped" ? "released" : "active", + leasePolicy: "ephemeral", releasedAt: gate === "remote_stopped" ? new Date() : null, + cleanupStatus: gate === "remote_stopped" ? "success" : null, + metadata: gate === "remote_stopped" ? { remoteExecutionTermination: remoteTerminationReceipt(identity, + { providerLeaseId: identity.providerLeaseId, state: "stopped" }) } : {}, + }); + } + const heartbeat = heartbeatService(db); + await heartbeat.resumeRemoteStopComments(source); + await heartbeat.resumeRemoteStopComments(source); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"].includes(gate) ? 1 : 0); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + if (["ready", "remote_stopped", "first_delivered", "last_delivered", "mixed_authors"].includes(gate)) { + expect(wake).toMatchObject({ status: "coalesced", runId: successors[0].id, requestedByActorId: "board" }); + expect(successors[0].contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + } else expect(wake.status).toBe("deferred_issue_execution"); + }); it.each(["pending", "failed", "historical", "shared", "retained"])("an explicit queued interrupt retries only its stopped sandbox, without granting automatic retries (%s)", async scenario => { const fails = scenario === "failed"; const protectedLease = scenario === "shared" || scenario === "retained"; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9c682d24a5..3de2209b3c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,6 +1,6 @@ import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversationExecutionWake, isWaitingConversation, prepareConversationTurn, settleConversationTurn } from "./agent-conversations.js"; import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; -import { hasAcknowledgedNativeStopIntent } from "./acknowledged-native-stop.js"; +import { hasAcknowledgedNativeStopIntent, isAcknowledgedNativeStop, acknowledgedNativeStopExecutionHasStopped } from "./acknowledged-native-stop.js"; import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; @@ -10051,6 +10051,20 @@ export function heartbeatService( !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return; const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); if (!issueId) return; + const currentRun = run.runtimeMode === "native" ? await getRun(run.id) : null; + const [coordinator] = currentRun ? await db.select({ phase: nativeRunFinalizations.phase, + leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), + )) : []; + // Stop ends one response. Saved user input can enter ordinary admission + // once its executor settles; it does not need a manufactured crash incident. + // The ordinary path still enforces task holds, ownership, and native session + // cleanup before starting a provider. + const stoppedNativeContinuation = currentRun && isAcknowledgedNativeStop(currentRun) && + !activeRunExecutions.has(run.id) && !coordinator?.leaseOwner && + ["terminal_failure", "applied"].includes(coordinator?.phase ?? "") && + await acknowledgedNativeStopExecutionHasStopped(db, currentRun) && + !(await getExecutionBlocker(db, run.companyId, issueId)); const legacyContinuation = run.runtimeMode === "legacy" && hasConversationContinuationPolicy((await getRun(run.id))?.resultJson) && !(await getExecutionBlocker(db, run.companyId, issueId)); @@ -10064,28 +10078,45 @@ export function heartbeatService( )).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50); for (const wake of pending) { if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue; - const payload = parseObject(wake.payload); + let payload = parseObject(wake.payload); if (payload.queuedCommentInterrupt) { await resumeQueuedCommentInterrupt(wake.companyId, wake.id); continue; } - const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]); - const commentId = deriveCommentId(context, payload); - if (legacyContinuation) { - if (!commentId || !run.finishedAt || !wake.requestedByActorId || - !["issue_commented", "issue_reopened_via_comment"].includes(wake.reason ?? "")) continue; + let context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]); + let commentId = deriveCommentId(context, payload); + let requestedByActorId = wake.requestedByActorId; + const reason = readNonEmptyString(context.wakeReason) ?? wake.reason; + if (stoppedNativeContinuation) { + const ids = await undeliveredLegacyUserCommentIds(db, run.companyId, issueId, run.agentId, + queuedCommentIdsFromWakePayload(payload)); + if (!ids.length) continue; + payload = withQueuedCommentIdsInWakePayload(payload, ids); + context = withQueuedCommentIdsInRunContext(context, ids); + commentId = ids.at(-1)!; + // A coalesced queue can contain several authors. Its saved comments, + // not the outer wake's first author, authorize the remaining input. + const [author] = await db.select({ id: issueComments.authorUserId }).from(issueComments).where(and( + eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, commentId), + )); + requestedByActorId = author?.id ?? null; + } + if (legacyContinuation || stoppedNativeContinuation) { + if (!commentId || !run.finishedAt || !requestedByActorId || + !["issue_commented", "issue_reopened_via_comment"].includes(reason ?? "")) continue; const [comment] = await db.select().from(issueComments).where(and( eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), - eq(issueComments.authorUserId, wake.requestedByActorId), isNull(issueComments.deletedAt), - isNull(issueComments.createdByRunId), gt(issueComments.createdAt, run.finishedAt), + eq(issueComments.authorUserId, requestedByActorId), isNull(issueComments.deletedAt), + isNull(issueComments.createdByRunId), + stoppedNativeContinuation ? undefined : gt(issueComments.createdAt, run.finishedAt), )); if (!comment?.body.trim()) continue; } else { let wait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." }; const admitted = await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId, - agentId: run.agentId, actorType: wake.requestedByActorType, actorId: wake.requestedByActorId, - reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true, + agentId: run.agentId, actorType: wake.requestedByActorType, actorId: requestedByActorId, + reason, commentId, successorRunId: randomUUID(), dryRun: true, onBlocked: (reason, message) => { wait = { reason, message }; }, }); if (!admitted) { @@ -10101,8 +10132,9 @@ export function heartbeatService( // Re-enter ordinary admission with the original user's authority. It // atomically adopts the deferred comments and still applies every gate. await enqueueWakeup(run.agentId, { source: wake.source as WakeupOptions["source"], triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"], - reason: wake.reason, payload, contextSnapshot: context, - requestedByActorType: "user", requestedByActorId: wake.requestedByActorId, + reason, payload, contextSnapshot: context, + requestedByActorType: "user", requestedByActorId, + ...(stoppedNativeContinuation ? { queuedCommentRequestId: wake.id } : {}), idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }, wake.id); break; } diff --git a/tests/e2e/composer-stop.spec.ts b/tests/e2e/composer-stop.spec.ts index d443ec1a46..e5030d6f7b 100644 --- a/tests/e2e/composer-stop.spec.ts +++ b/tests/e2e/composer-stop.spec.ts @@ -397,6 +397,13 @@ for (const adapter of ["process", "paperclip_runner"] as const) { const resumedChildRun = await running(request, child.id, adapter); expect(resumedParentRun.id).not.toBe(parentRun.id); expect(resumedChildRun.id).not.toBe(childRun.id); + if (adapter === "paperclip_runner") { + await expect.poll(async () => { + const calls = await readFile(process.env.PAPERCLIP_STOP_CODEX_LOG!, "utf8"); + return calls.split("turn/start").length - 1; + }, { timeout: 30_000 }).toBeGreaterThanOrEqual(5); + await page.screenshot({ path: testInfo.outputPath("native-resumed.png"), fullPage: true }); + } await menu(page, "Pause subtree"); await expect(page.getByRole("dialog")).toHaveCount(0); await expect( diff --git a/ui/src/pages/IssueDetail.test.tsx b/ui/src/pages/IssueDetail.test.tsx index 57ed3bad90..6fa0a8f79f 100644 --- a/ui/src/pages/IssueDetail.test.tsx +++ b/ui/src/pages/IssueDetail.test.tsx @@ -5629,6 +5629,40 @@ describe("IssueDetail", () => { }, ); + it("keeps the last queued message mounted until steering is acknowledged so rejection stays visible", async () => { + const queue = createQueuedCommentQueue(); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", executionRunId: "run-active-1" })); + mockAgentsApi.list.mockResolvedValue([createAgent({ adapterType: "paperclip_runner" })]); + mockIssuesApi.listComments.mockResolvedValue([queue.entries[0].comment]); + mockIssuesApi.getQueuedComments.mockResolvedValue(queue); + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue({ + id: "run-active-1", runtimeMode: "native", status: "running", invocationSource: "issue", + triggerDetail: null, contextCommentId: null, contextWakeCommentId: null, + startedAt: "2026-04-21T00:00:00.000Z", finishedAt: null, createdAt: "2026-04-21T00:00:00.000Z", + agentId: "agent-1", agentName: "Runner", adapterType: "paperclip_runner", issueId: "issue-1", + }); + let rejectSteer!: (error: Error) => void; + mockIssuesApi.steerQueuedComment.mockReturnValue(new Promise((_, reject) => { rejectSteer = reject; })); + await act(async () => { root.render(); }); + type Props = { onSteerQueuedComment: (id: string, revision: string) => Promise; queuedCommentQueue: IssueQueuedCommentQueue | null }; + let props!: Props; + await waitForAssertion(() => { + props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + expect(props.queuedCommentQueue?.entries).toHaveLength(1); + expect(props.queuedCommentQueue?.queueId).toBe("wake-queue-1"); + expect(props.queuedCommentQueue?.targetRunId).toBe("run-active-1"); + }); + let pending!: Promise; + await act(async () => { pending = props.onSteerQueuedComment("queued-comment-1", queue.revision).catch(error => error); }); + // The child owns its pending/error state. Unmounting it here loses any later error. + const whilePending = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + expect(whilePending.queuedCommentQueue?.entries).toHaveLength(1); + const failure = new ApiError("This runner does not support steering", 409, { code: "steering_unsupported" }); + await act(async () => { rejectSteer(failure); await pending; }); + expect(await pending).toBe(failure); + expect((mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props).queuedCommentQueue?.entries).toHaveLength(1); + }); + it("promotes a steered message immediately while its durable timeline position refreshes", async () => { const queue = createQueuedCommentQueue(); const steeredQueue = createQueuedCommentQueue({ @@ -5702,6 +5736,7 @@ describe("IssueDetail", () => { await waitForAssertion(() => { expect(mockIssuesApi.steerQueuedComment).toHaveBeenCalled(); expect(mockActivityApi.forIssue.mock.calls.length).toBeGreaterThan(1); + expect(mockIssueChatThreadRender.mock.calls.at(-1)?.[0]?.queuedCommentQueue).toBeNull(); }); const whileRefreshing = mockIssueChatThreadRender.mock.calls.at( diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index 41188f7c98..1a2b4979b4 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -2128,16 +2128,6 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ throw new Error( "The queued message no longer has an active run target.", ); - const anchorAt = new Date().toISOString(); - setLocalSteeringPlacements((current) => { - const next = new Map(current); - const sequence = [...current.values()].filter( - (placement) => placement.targetRunId === targetRunId, - ).length; - next.set(commentId, { targetRunId, anchorAt, sequence }); - return next; - }); - setConsumedQueuedCommentIds((current) => new Set(current).add(commentId)); try { const nextQueue = await issuesApi.steerQueuedComment( issueId, @@ -2148,6 +2138,18 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ revision, }, ); + // Keep the queue component mounted until the server accepts steering: + // its pending/error state must survive a rejected last-row action. + const anchorAt = new Date().toISOString(); + setLocalSteeringPlacements((current) => { + const next = new Map(current); + const sequence = [...current.values()].filter( + (placement) => placement.targetRunId === targetRunId, + ).length; + next.set(commentId, { targetRunId, anchorAt, sequence }); + return next; + }); + setConsumedQueuedCommentIds((current) => new Set(current).add(commentId)); // The local steering placement already promoted the message into the // active turn. Refresh its durable acknowledgement before publishing // the returned queue so the local and server anchors hand off without a