Preserve safe ACPX admission stages on startup timeout

This commit is contained in:
Dotta committed 2026-10-10 23:38:48 -05:00
1 parent b68c68566c
commit b698f54966
4 files changed
+165 -2

No files matched your search

@@ -84,6 +84,7 @@ pub struct AcpxSidecarTransport {
buffered_events: VecDeque<AcpxSidecarEvent>,
stderr_tail: BoundedLogBuffer,
stderr_categories: BTreeSet<&'static str>,
admission_diagnostic: Option<(&'static str, u64)>,
poisoned: bool,
}
@@ -322,6 +323,7 @@ impl AcpxSidecarTransport {
buffered_events: VecDeque::new(),
stderr_tail: BoundedLogBuffer::new(32, 8 * 1024),
stderr_categories: BTreeSet::new(),
admission_diagnostic: None,
poisoned: false,
})
}
@@ -601,6 +603,7 @@ impl AcpxSidecarTransport {
fn diagnostic_suffix(&self) -> String {
let diagnostics = self.stderr_tail.snapshot().lines.join("\n");
let admission = admission_diagnostic_suffix(self.admission_diagnostic);
let categories = if self.stderr_categories.is_empty() {
String::new()
} else {
@@ -614,15 +617,22 @@ impl AcpxSidecarTransport {
)
};
if diagnostics.is_empty() {
categories
format!("{admission}{categories}")
} else {
format!("{categories} stderrTail={diagnostics:?}")
format!("{admission}{categories} stderrTail={diagnostics:?}")
}
}
fn record_stderr(&mut self, line: &str) {
// Only fixed categories cross this boundary. Raw errors, stack paths,
// identifiers, and credential-bearing strings remain fully redacted.
if let Some(progress) = parse_admission_diagnostic(line) {
// Cleanup can race the outer deadline; preserve the admission step
// that failed rather than replacing it with generic teardown.
if progress.0 != "cleanup" || self.admission_diagnostic.is_none() {
self.admission_diagnostic = Some(progress);
}
}
self.stderr_categories
.extend(stderr_diagnostic_categories(line));
self.stderr_tail.push(redact_diagnostic(line));
@@ -787,6 +797,40 @@ fn nullable_identifier(
Ok(Some(value.to_owned()))
}
// Reuse the sidecar's diagnostic channel, retaining only a closed stage and
// bounded integer. Unknown lines remain fully redacted and never grant authority.
fn parse_admission_diagnostic(line: &str) -> Option<(&'static str, u64)> {
let (stage, elapsed) = line
.strip_prefix("[paperclip-acpx-sidecar] admission_")?
.split_once(": elapsedMs=")?;
let stage = match stage {
"binding" => "binding",
"installation" => "installation",
"sandbox" => "sandbox",
"lifetime" => "lifetime",
"agent_files" => "agent_files",
"skills" => "skills",
"command" => "command",
"tool_bridge" => "tool_bridge",
"handshake" => "handshake",
"verification" => "verification",
"ready" => "ready",
"cleanup" => "cleanup",
_ => return None,
};
if elapsed.is_empty() || !elapsed.bytes().all(|byte| byte.is_ascii_digit()) {
return None;
}
let elapsed = elapsed.parse::<u64>().ok()?;
(elapsed <= 3_600_000).then_some((stage, elapsed))
}
fn admission_diagnostic_suffix(progress: Option<(&'static str, u64)>) -> String {
progress.map_or_else(String::new, |(stage, elapsed)| {
format!(" admissionStage={stage} admissionElapsedMs={elapsed}")
})
}
fn redact_diagnostic(value: &str) -> String {
if value.is_empty() {
String::new()
@@ -991,6 +1035,60 @@ fn response_error_classification(error: &ResponseError) -> &'static str {
#[cfg(test)]
mod tests {
#[test]
fn admission_diagnostics_accept_only_closed_stages_and_bounded_integer_timing() {
assert_eq!(
parse_admission_diagnostic(
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=321"
),
Some(("handshake", 321))
);
for line in [
"[paperclip-acpx-sidecar] admission_private-path: elapsedMs=321",
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=321 private-token",
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=-1",
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=NaN",
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=3600001",
"[paperclip-acpx-sidecar] admission_handshake: elapsedMs=18446744073709551616",
"private-prefix [paperclip-acpx-sidecar] admission_handshake: elapsedMs=321",
] {
assert_eq!(parse_admission_diagnostic(line), None);
}
}
#[cfg(unix)]
#[test]
fn admission_timeout_retains_last_safe_stage_without_untrusted_stderr() {
let config = AcpxSidecarTransportConfig {
command: PathBuf::from("/bin/sh"),
args: vec!["-c".into(), "printf '%s\\n' '[paperclip-acpx-sidecar] admission_sandbox: elapsedMs=1' '[paperclip-acpx-sidecar] admission_handshake: elapsedMs=24' '[paperclip-acpx-sidecar] admission_handshake: elapsedMs=25 /private/token-canary' '[paperclip-acpx-sidecar] admission_cleanup: elapsedMs=26' >&2; sleep 2".into()],
verified_launch: None,
request_timeout: Duration::from_millis(200),
shutdown_grace: Duration::from_millis(10),
};
let mut transport = AcpxSidecarTransport::start(&config).unwrap();
let error = transport
.request(GeneratedAcpxSidecarCommand::SessionOpen, json!({}))
.err()
.unwrap()
.to_string();
assert!(
error.contains("request timed out at session.open"),
"{error}"
);
assert!(
error.contains("admissionStage=handshake admissionElapsedMs=24"),
"{error}"
);
assert!(
!error.contains("private") && !error.contains("token-canary"),
"{error}"
);
assert!(transport
.request(GeneratedAcpxSidecarCommand::SessionOpen, json!({}))
.is_err());
}
#[test]
fn pi_credentials_require_a_bounded_binding_without_process_control_variables() {
assert!(pi_credential_environment_keys(None).unwrap().is_empty());
@@ -315,6 +315,7 @@ async function dispatch(
},
{
retainAdmissionCleanup: retainFailedAdmissionCleanup,
onAdmissionStage: (stage, elapsedMs) => diagnostic(`admission_${stage}`, `elapsedMs=${elapsedMs}`),
reportRetainedCleanupFailure: reportRetainedAcpxCleanupFailure,
openRuntime: (options) =>
openCodexAcpxRuntime(options, {
@@ -201,6 +201,45 @@ afterEach(async () => {
});
describe("ACPX runtime host", () => {
it.each(["throw", "reject"] as const)("keeps admission diagnostics observational when the observer can %s", async (failure) => {
const fixture = await hostFixture();
const port = runtimePort({ getStatus: async () => ({ models: { currentModelId: "claude-sonnet-5" } }) });
const gate = deferred<AcpxRuntimePort>();
const stages: Array<{ stage: string; elapsedMs: number }> = [];
const openRuntime = vi.fn(() => gate.promise);
const opening = trackAdmissionOpening(AcpxRuntimeHost.open({
...fixture.options, agent: "claude", model: "claude-sonnet-5",
semanticTools: { tools: [], handler: async () => ({}) },
}, {
...fixture.dependencies({ openRuntime }),
onAdmissionStage: (stage, elapsedMs) => {
stages.push({ stage, elapsedMs });
if (failure === "throw") throw new Error("private-observer-canary");
return Promise.reject(new Error("private-observer-canary"));
},
}));
try {
await vi.waitFor(() => expect(openRuntime).toHaveBeenCalledOnce());
expect(stages.at(-1)?.stage).toBe("handshake");
} finally {
gate.resolve(port);
}
const host = await opening;
expect(stages.map(({ stage }) => stage)).toEqual([
"binding", "installation", "sandbox", "lifetime", "agent_files", "skills",
"command", "tool_bridge", "handshake", "verification", "ready",
]);
for (const [index, { elapsedMs }] of stages.entries()) {
expect(Number.isInteger(elapsedMs)).toBe(true);
expect(elapsedMs).toBeGreaterThanOrEqual(stages[index - 1]?.elapsedMs ?? 0);
expect(elapsedMs).toBeLessThanOrEqual(3_600_000);
}
expect(host.identity().acpxRecordId).toBe("record-1");
await host.close({ reason: "diagnostic observer test complete" });
expect(port.close).toHaveBeenCalledOnce();
expect(fixture.commandClose).toHaveBeenCalledOnce();
});
it.each([undefined, "plan", "ask"] as const)("requires observed Cursor mode %s in the host identity", async selected => {
const fixture = await hostFixture();
const mode = selected ?? "agent";
@@ -199,7 +199,12 @@ export interface AcpxRetainedCleanupFailure {
error: unknown;
}
export type AcpxAdmissionStage = "binding" | "installation" | "sandbox" | "lifetime"
| "agent_files" | "skills" | "command" | "tool_bridge" | "handshake" | "verification" | "ready" | "cleanup";
export interface AcpxRuntimeHostDependencies {
/** Closed diagnostic progress only; observer failures cannot affect ownership. */
onAdmissionStage?: (stage: AcpxAdmissionStage, elapsedMs: number) => void;
verifyInstallation?: (
profile: QualifiedAcpxProfile,
) => Promise<VerifiedAcpxInstallation>;
@@ -325,6 +330,15 @@ export class AcpxRuntimeHost {
options: OpenAcpxRuntimeHostOptions,
dependencies: AcpxRuntimeHostDependencies,
): Promise<AcpxRuntimeHost> {
const admissionStarted = performance.now();
const reportStage = (stage: AcpxAdmissionStage): void => {
try {
void Promise.resolve(dependencies.onAdmissionStage?.(stage,
Math.max(0, Math.min(3_600_000, Math.round(performance.now() - admissionStarted))),
)).catch(() => undefined);
} catch { /* Diagnostics never own admission or cleanup. */ }
};
reportStage("binding");
options.signal?.throwIfAborted();
if (options.agent === "pi" && typeof options.providerPolicy?.readOnly !== "boolean") {
throw new Error("Pi admission requires an explicit task execution policy");
@@ -368,6 +382,7 @@ export class AcpxRuntimeHost {
);
}
reportStage("installation");
const installation = await runAbortableAdmissionStage(
options.signal,
() =>
@@ -422,6 +437,7 @@ export class AcpxRuntimeHost {
retainRuntimeHostCleanup(ownedCleanup);
};
try {
reportStage("sandbox");
const sandbox = await runAbortableAdmissionStage(
options.signal,
() =>
@@ -443,6 +459,7 @@ export class AcpxRuntimeHost {
}),
dependencies.retainAdmissionCleanup,
);
reportStage("lifetime");
if (options.agent === "codex") {
credential = await acquireAbortableAdmissionResource({
signal: options.signal,
@@ -483,6 +500,7 @@ export class AcpxRuntimeHost {
// agent and run. Environment/config values cannot widen the grant. Each
// resumed process receives the newly registered copy; collection already
// requires verified provider shutdown in the native executor.
reportStage("agent_files");
const agentFiles = options.runtimeContext?.persistentAgentHome || ["cursor", "copilot", "pi"].includes(options.agent)
? bindAcpxAgentFiles(options.runtimeContext, [sandbox.root,
...(installation.agentServerPackageJsonPath === null ? [] : [dirname(installation.agentServerPackageJsonPath)]),
@@ -522,6 +540,7 @@ export class AcpxRuntimeHost {
PAPERCLIP_PI_READ_ROOTS: JSON.stringify([...(options.providerPolicy?.readRoots ?? []), ...skills.readRoots]),
});
}
reportStage("skills");
if (options.agent === "claude" || options.agent === "grok") {
// The lifetime lease proves the previous provider has stopped. Refresh
// the assigned snapshot before every launch, including durable resume;
@@ -534,6 +553,7 @@ export class AcpxRuntimeHost {
);
options.signal?.throwIfAborted();
}
reportStage("command");
command = await acquireAbortableAdmissionResource({
signal: options.signal,
acquire: () => installation.openCommand({ signal: options.signal }),
@@ -547,6 +567,7 @@ export class AcpxRuntimeHost {
signal => installation.openCommand({ signal }),
);
command = commandOwner.command;
reportStage("tool_bridge");
toolBridge = options.semanticTools
? await acquireAbortableAdmissionResource({
signal: options.signal,
@@ -561,6 +582,7 @@ export class AcpxRuntimeHost {
if (admittedLifetime === null) {
throw new Error("ACPX provider lifetime lease is unavailable");
}
reportStage("handshake");
runtime = await acquireAbortableAdmissionResource({
signal: options.signal,
acquire: async () => {
@@ -628,6 +650,7 @@ export class AcpxRuntimeHost {
});
},
});
reportStage("verification");
const runtimeIdentity = await runAbortableAdmissionStage(
options.signal,
() =>
@@ -661,6 +684,7 @@ export class AcpxRuntimeHost {
verifyExpectedAcpxIdentity(options.expectedIdentity, binding, identity);
}
options.signal?.throwIfAborted();
reportStage("ready");
admissionSucceeded = true;
return new AcpxRuntimeHost({
runtime,
@@ -676,6 +700,7 @@ export class AcpxRuntimeHost {
: [],
});
} catch (error) {
reportStage("cleanup");
error = classifyAcpxProfileError(options.agent, error) ?? error;
const cleanup = cleanupRuntimeResources(
runtime,