fix: preserve queued work after native Stop and honor steering support (#13354)
## Thinking Path > - Paperclip lets people manage AI agents and their tasks. > - The runner owns execution, while the task keeps user instructions and status. > - Stop must stop the current response without losing instructions that the user already sent. > - The queue stored its original reason inside saved context. Recovery checked the outer deferred reason and left the message waiting. > - Claude also exposed Steer through a shared method even though its driver did not support it. A rejected request could remove its own error row. > - This pull request keeps queued work until execution has stopped, uses the driver's real capability, and preserves completion event order. > - Users can continue work without repairing task state or repeating messages. ## Linked Issues or Issue Description Refs #13338. Related recovery work: #13353. **What happened?** A message sent during a native run stayed queued after Stop. Claude exposed an unsupported Steer action. A steering failure could hide the queued row and its error. A terminal event could also precede the final provider result, and subtree Stop omitted the board actor. **Expected behavior** Stop ends the current execution. Once Paperclip proves that execution has stopped, it delivers the saved instruction once through normal admission. Pause and recovery holds still prevent dispatch. Unsupported controls stay disabled, and a rejected action leaves an actionable error visible. Final results precede terminal events. **Steps to reproduce** 1. Start a Claude or Codex task that writes a file and then waits. 2. Send a follow-up instruction while it runs. 3. Press Stop. Check that the queued instruction runs once and preserves the file. 4. Check Claude's Steer control and simulate a server rejection on the only queued message. ## What Changed - Recover saved native comments after acknowledged Stop using their original wake reason. - Require durable remote termination receipts or verified local process termination before dispatch. - Preserve actor identity, queued-message deduplication, Pause, and recovery gates. - Derive steering support from the driver descriptor and reject unsupported calls. - Keep the queue mounted until a steering request succeeds so its error remains visible. - Emit provider results before terminal events and pass the board actor into subtree Stop. - Document Stop and steering behavior. ## Verification - Final-head continuation suite: 104 passed, after failing regressions for saved wake reasons, cleanup proof, and deduplication of every queued message. Steering UI: 121 passed. Driver capability: 26 passed. Runner backend/transport: 205 passed; Rust library: 285 passed. - Real Claude and Codex browser journeys both preserved the saved file, delivered the queued instruction once after Stop, and reached Done with exactly two total runs. The process Stop browser fixture also passed. - Local full repository typecheck and build passed on `afaa35139`; server typecheck and the affected 104-test suite passed after the final queue changes. Token gates passed. Final-head CI verifies the complete integrated source. - Local aggregate evidence has explicit limits: the general-server invocation overlapped the queue fixes and finished with 11,989 passed, 2 failed, and 80 skipped; both failures are covered by the final 104-test pass. The UI and CLI then passed all 6,184 and 485 tests; the complete 145-file serialized rerun passed all 2470 tests. The shared-package lock fixture passed unchanged on rerun, but the package phase subsequently stopped at an embedded-Postgres bootstrap resource failure. No single pristine green local full aggregate is claimed. - Greptile reviewed `fa66e2bd5` at [5/5 with no unresolved findings](https://github.com/paperclipai/paperclip/pull/13354#issuecomment-5650334242). [Final-head CI completed successfully](https://github.com/paperclipai/paperclip/actions/runs/34733781888/attempts/2): 33 successful checks, 2 conditional skips, including all server, package, UI, browser, runner, typecheck, and build gates. The first attempt hit a preview-readiness/port-collision fixture; its unchanged local control passed 25 tests with 3 skips, and one supported unchanged CI retry passed the affected shard and aggregate gates. ## Risks - Queue recovery must never overlap an old execution. Unknown cleanup state remains blocked. - Driver descriptors are now authoritative for steering; a wrong descriptor disables the action instead of attempting it. - No schema migration or historical status reconciliation is included. ## Model Used OpenAI GPT-6 through Codex, with reasoning, code execution, browser testing, and tool use. The exact hosted model ID and context-window size are not exposed in this task. ## 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 #` 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
6809314a3f
commit
c9e3bb7ca4
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<String> {
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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| {
|
||||
|
|
|
|||
|
|
@ -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<NormalizedProviderEvent>,
|
||||
outcome_events: Vec<NormalizedProviderEvent>,
|
||||
) -> Vec<NormalizedProviderEvent> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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-"),
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<typeof hasAcknowledgedNativeStopIntent>[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));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,6 +42,72 @@ const support = await getEmbeddedPostgresTestSupport();
|
|||
actorType: "user", actorId: "board", reason: "issue_commented" };
|
||||
}
|
||||
type Fixture = Awaited<ReturnType<typeof seed>>;
|
||||
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";
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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(<QueryClientProvider client={queryClient}><IssueDetail /></QueryClientProvider>); });
|
||||
type Props = { onSteerQueuedComment: (id: string, revision: string) => Promise<void>; 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<unknown>;
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue