Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .changeset/6261-stale-lifecycle-ready-resume.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
### Fixed

- `session.resume` of a session whose detached-idle host already exited no longer fails with `EEXIST` on `<id>.lifecycle.ready.json`: the host revokes its own ready marker on graceful exit, including the global SIGTERM path, and a launch retires the same id's leftover ready/marker pair when the recorded owner is proven exited (a live or unknown owner still wins) (#6261).
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
### Fixed

- `session.resume` of a session whose detached-idle host already exited no longer fails with `EEXIST` on `<id>.lifecycle.ready.json`: the host revokes its own ready marker on graceful exit, and a launch retires the same id's leftover ready/marker pair when the recorded owner is proven exited (a live or unknown owner still wins) (#6261).
133 changes: 112 additions & 21 deletions packages/coding-agent/src/commands/sdk.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,15 @@
import { type ChildProcess, spawn } from "node:child_process";
import { createHash } from "node:crypto";
import { createHash, randomUUID } from "node:crypto";
import * as nodeFs from "node:fs";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { exactRemoveDirectoryTree, type NativeDirectoryTreeSnapshot, snapshotDirectoryTree } from "@gajae-code/natives";
import {
exactRemoveDirectoryTree,
exactUnlinkDirect,
type NativeDirectoryTreeSnapshot,
snapshotDirectoryTree,
} from "@gajae-code/natives";
import { logger, postmortem } from "@gajae-code/utils";
import { CliParseError, Command } from "@gajae-code/utils/cli";
import type { Args as ParsedArgs } from "../cli/args";
Expand All @@ -24,6 +29,7 @@ import {
writeBrokerStartupExitRecordBounded,
} from "../sdk/broker/broker-exit";
import { readBrokerDiscovery } from "../sdk/broker/discovery";
import { type EndpointFileRead, readEndpointFile } from "../sdk/broker/endpoint-authority";
import {
emitBrokerStartupTestSignal,
reconcileBrokerGenerationForStartup,
Expand Down Expand Up @@ -653,6 +659,7 @@ export async function runSessionHost(
let startupComplete = false;
let readinessPublicationCleanupComplete = true;
let revokePendingReadinessMarker: (() => Promise<boolean>) | undefined;
let revokePublishedReadinessMarker: (() => Promise<boolean>) | undefined;
let readinessRevocation: Promise<boolean> | undefined;
let startupInterruption: SdkStartupFailure | undefined;
const interrupted = Promise.withResolvers<SdkStartupFailure>();
Expand Down Expand Up @@ -1267,48 +1274,134 @@ export async function runSessionHost(
return failureRollback;
};
const sessionEndpointPath = path.join(request.stateRoot, "sdk", `${request.sessionId}.json`);
const exitAfterSessionDisposal = async (reason?: "detached_idle"): Promise<void> => {
await disposeSession(reason);
const removeOwnedSessionEndpoint = async (): Promise<boolean> => {
try {
await fs.stat(sessionEndpointPath);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return true;
return false;
}
const endpoint: EndpointFileRead | undefined = await readEndpointFile(sessionEndpointPath);
if (!endpoint) return false;
let parsed: { pid?: unknown; sessionId?: unknown };
try {
parsed = JSON.parse(endpoint.source) as { pid?: unknown; sessionId?: unknown };
} catch {
return false;
}
if (parsed.pid !== process.pid || parsed.sessionId !== request.sessionId) {
return false;
}
const removal = (() => {
try {
return exactUnlinkDirect(sessionEndpointPath, {
dev: endpoint.dev,
ino: endpoint.ino,
nlink: endpoint.nlink,
size: endpoint.size,
mtimeNs: endpoint.mtimeNs,
quarantineName: `.gjc-delete-endpoint-${request.sessionId}-${randomUUID()}`,
sha256: createHash("sha256").update(endpoint.source).digest("hex"),
allowHardLink: true,
});
} catch (error) {
logger.warn("SDK host endpoint cleanup failed", { error: String(error) });
return undefined;
}
})();
if (!removal) return false;
return (
removal.ok ||
removal.code === "not_found" ||
(removal.code === "cleanup_pending" &&
removal.payloadDurable === true &&
!removal.retainedSuccessorPath &&
!removal.retainedUnknownPath)
);
};
let unregisterHostPostmortem: (() => void) | undefined;
let sessionDisposalExit: Promise<void> | undefined;
const exitAfterSessionDisposal = async (reason?: "detached_idle", exitProcess = true): Promise<void> => {
let failure: SdkStartupFailure | undefined;
let ownsEndpoint = false;
const endpointCleanup = (async (): Promise<boolean> => {
let complete = await removeOwnedSessionEndpoint();
for (let attempt = 0; !complete && attempt < 20; attempt += 1) {
await Bun.sleep(25);
complete = await removeOwnedSessionEndpoint();
}
return complete;
})();
let [endpointCleanupComplete, readinessRevoked] = await Promise.all([
endpointCleanup,
revokePublishedReadinessMarker?.() ?? Promise.resolve(true),
]);
await disposeSession(reason);
endpointCleanupComplete = await removeOwnedSessionEndpoint();
try {
const endpoint = JSON.parse(await fs.readFile(sessionEndpointPath, "utf8")) as {
pid?: unknown;
sessionId?: unknown;
};
if (endpoint.pid === process.pid && endpoint.sessionId === request.sessionId)
failure = {
phase: "startup",
reason: "failed",
message: `SDK host endpoint remained after graceful shutdown: ${request.sessionId}`,
};
ownsEndpoint = endpoint.pid === process.pid && endpoint.sessionId === request.sessionId;
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
failure = {
phase: "startup",
reason: "failed",
message: `SDK host endpoint cleanup could not be verified: ${request.sessionId}`,
};
}
} else endpointCleanupComplete = true;
}
if (!ownsEndpoint && !failure && !readinessRevoked && revokePublishedReadinessMarker) {
failure = {
phase: "startup",
reason: "failed",
message: `SDK host readiness marker cleanup could not be verified: ${request.sessionId}`,
};
}
if (ownsEndpoint && !endpointCleanupComplete && !failure)
failure = {
phase: "startup",
reason: "failed",
message: `SDK host endpoint remained after graceful shutdown: ${request.sessionId}`,
};
if (failure) {
process.exitCode = 1;
process.stderr.write(`${failure.message}\n`);
await writeFailure(failure, rollback.result).catch(() => {});
}
process.exit(process.exitCode ?? 0);
unregisterHostPostmortem?.();
unregisterHostPostmortem = undefined;
if (exitProcess) process.exit(process.exitCode ?? 0);
};
let stopping = false;
const requestSessionStop = (reason?: "detached_idle", exitProcess = true): Promise<void> => {
if (stopping) return sessionDisposalExit ?? Promise.resolve();
stopping = true;
sessionDisposalExit = exitAfterSessionDisposal(reason, exitProcess);
return sessionDisposalExit;
};
const stop = (reason?: "detached_idle") => {
if (startupComplete) {
if (stopping) return;
stopping = true;
void exitAfterSessionDisposal(reason);
void requestSessionStop(reason);
return;
}
const failure = capability.normalizeFailure("startup", "failed", "SDK lifecycle host terminated.");
interruptStartup(failure);
};
const onReadySignal = (): void => stop();
const registerHostPostmortem = (): void => {
if (unregisterHostPostmortem) return;
unregisterHostPostmortem = postmortem.register("sdk-session-host:exit", async reason => {
if (reason !== postmortem.Reason.SIGTERM && reason !== postmortem.Reason.SIGINT) return;
// The postmortem signal authority defaults to 143/130; a session host
// signal is graceful and must preserve its cleanup result instead.
await requestSessionStop(undefined, false);
process.exitCode = 0;
process.exit(0);
});
};
registerHostPostmortem();

try {
const startupThinkingLevel = request.modelId ? parseModelString(request.modelId)?.thinkingLevel : undefined;
Expand Down Expand Up @@ -1356,6 +1449,7 @@ export async function runSessionHost(
effectMarker,
() => startupInterruption === undefined && now() < request.semanticReadyDeadlineAt,
revoke => {
revokePublishedReadinessMarker = revoke;
revokePendingReadinessMarker = revoke;
if (startupInterruption !== undefined) startReadinessRevocation(revoke);
else if (now() >= request.semanticReadyDeadlineAt) {
Expand Down Expand Up @@ -1390,8 +1484,6 @@ export async function runSessionHost(
revokePendingReadinessMarker = undefined;
readinessRevocation = undefined;
removeStartupSignalHandlers();
process.once("SIGTERM", onReadySignal);
process.once("SIGINT", onReadySignal);
} catch (error) {
if (error instanceof LifecycleReadinessCleanupError) constructionCleanupComplete = false;
const failure =
Expand All @@ -1404,8 +1496,6 @@ export async function runSessionHost(
await failAfterRollback(durableFailure);
} finally {
removeStartupSignalHandlers();
process.removeListener("SIGTERM", onReadySignal);
process.removeListener("SIGINT", onReadySignal);
}
throw error;
}
Expand Down Expand Up @@ -1655,7 +1745,8 @@ export default class Sdk extends Command {
child.unref();
return;
}
if (internal.action === "broker-internal") usePostmortemSignalExitAuthority();
if (internal.action === "broker-internal" || internal.action === "session-host-internal")
usePostmortemSignalExitAuthority();
if (internal.action === "session-host-internal") {
await runSessionHost();
return;
Expand Down
105 changes: 104 additions & 1 deletion packages/coding-agent/src/sdk/broker/lifecycle.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1667,9 +1667,9 @@ export async function reapDeadLifecycleMarkers(
let inspected = 0;
for await (const entry of directoryHandle) {
if (inspected >= inspectionLimit) break;
inspected += 1;
const name = entry.name;
if (!entry.isFile() || !name.endsWith(".lifecycle.json")) continue;
inspected += 1;
const id = name.slice(0, -".lifecycle.json".length);
if (!isCanonicalSessionId(id)) continue;
const markerPath = path.join(directory, name);
Expand Down Expand Up @@ -1741,6 +1741,108 @@ export async function reapDeadLifecycleMarkers(
return reaped;
}

/**
* Retires one launch id's lifecycle marker pair after its recorded owner has
* exited. Unlike the bounded background sweep, this launch-local repair is
* not age-gated and does not require the ready sibling to contain the same
* effect marker: the primary marker is the owner authority for this id.
*/
type RetireExitedLifecycleMarkerPairHook = () => void | Promise<void>;

async function retireExitedLifecycleMarkerPairImpl(
root: string,
id: string,
afterObservation?: RetireExitedLifecycleMarkerPairHook,
): Promise<boolean> {
if (!isCanonicalSessionId(id)) return false;
let directory: string;
let directoryIdentity: { dev: bigint; ino: bigint };
try {
const canonicalRoot = fsSync.realpathSync(root);
directory = path.join(canonicalRoot, "sdk");
const directoryStat = fsSync.lstatSync(directory, { bigint: true });
if (!directoryStat.isDirectory() || directoryStat.isSymbolicLink()) return false;
directoryIdentity = { dev: directoryStat.dev, ino: directoryStat.ino };
} catch {
return false;
}
const markerPath = lifecycleMarkerPath(path.dirname(directory), id);
const primary = captureLifecycleFile(markerPath, true, true);
if (!primary) return false;
let marker: EffectMarker;
try {
const parsed = parseLifecycleJson(primary.bytes);
if (!isExactEffectMarker(parsed)) return false;
marker = parsed;
} catch {
return false;
}
if (observeProcess(marker.pid, marker.incarnation) !== "exited") return false;
await afterObservation?.();
const readyPath = lifecycleReadyPath(path.dirname(directory), id);
const ready = captureLifecycleFile(readyPath, true, true);
try {
const currentParent = lifecycleParentIdentity(directory);
if (
!currentParent ||
BigInt(currentParent.dev) !== directoryIdentity.dev ||
BigInt(currentParent.ino) !== directoryIdentity.ino
)
return false;
const currentPrimary = captureLifecycleFile(markerPath, true, true);
await afterObservation?.();
if (
!currentPrimary ||
!sameLifecycleCleanupIdentity(
currentPrimary.identity,
serializeCleanupIdentity({ ...primary.identity, size: Number(primary.identity.size) }),
)
)
return false;
const currentMarker = parseLifecycleJson(currentPrimary.bytes);
if (!isExactEffectMarker(currentMarker) || !sameEffectMarker(currentMarker, marker)) return false;
if (ready) {
const currentReady = captureLifecycleFile(readyPath, true, true);
if (
!currentReady ||
!sameLifecycleCleanupIdentity(
currentReady.identity,
serializeCleanupIdentity({ ...ready.identity, size: Number(ready.identity.size) }),
)
)
return false;
const removedReady = nativeLifecycle().exactUnlinkDirect(readyPath, {
...ready.identity,
parentDev: directoryIdentity.dev,
parentIno: directoryIdentity.ino,
quarantineName: `.gjc-reap-${randomUUID()}-${path.basename(readyPath)}`,
});
if (!removedReady.ok && removedReady.code !== "not_found") return false;
}
const removedMarker = nativeLifecycle().exactUnlinkDirect(markerPath, {
...primary.identity,
parentDev: directoryIdentity.dev,
parentIno: directoryIdentity.ino,
quarantineName: `.gjc-reap-${randomUUID()}-${path.basename(markerPath)}`,
});
return removedMarker.ok || removedMarker.code === "not_found";
} catch {
return false;
}
}

export async function retireExitedLifecycleMarkerPair(root: string, id: string): Promise<boolean> {
return retireExitedLifecycleMarkerPairImpl(root, id);
}

export async function retireExitedLifecycleMarkerPairForTest(
root: string,
id: string,
afterObservation: RetireExitedLifecycleMarkerPairHook,
): Promise<boolean> {
return retireExitedLifecycleMarkerPairImpl(root, id, afterObservation);
}

export async function writeEffectMarker(root: string, id: string, marker: EffectMarker): Promise<void> {
const directory = path.join(root, "sdk");
await fs.mkdir(directory, { recursive: true, mode: 0o700 });
Expand Down Expand Up @@ -6650,6 +6752,7 @@ async function executeLifecycleResponse(
terminationStartDeadline = childDeadlines.terminationStartDeadlineAt;
lifecycleDeadline = childDeadlines.lifecycleCleanupDeadlineAt;
}
await retireExitedLifecycleMarkerPair(launch.root, launch.id);
await reapDeadLifecycleMarkers(launch.root);
} catch (error) {
if (launch.worktreePlan && worktreeReceipt?.created && !worktreeReceipt.reused) {
Expand Down
11 changes: 6 additions & 5 deletions packages/coding-agent/test/sdk-broker-lifecycle-e2e.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -813,13 +813,14 @@ test("shipped session host exits promptly after publishing a cutoff receipt", as
});
if (!child.pid) throw new Error("session host has no pid");
const childIncarnation = await incarnation(child.pid);
const cutoffStartedAt = performance.now();
await fs.writeFile(
path.join(stateRoot, "sdk", `${sessionId}.lifecycle.json`),
JSON.stringify({ pid: child.pid, effectMarker, incarnation: childIncarnation }),
);
await waitFor(async () => ((await Bun.file(failurePath).exists()) ? true : undefined), "cutoff receipt");
const receiptPublishedAt = performance.now();
const outcome = await Promise.race([
child.exited.then(code => ({ code, latencyMs: performance.now() - cutoffStartedAt })),
child.exited.then(code => ({ code, latencyMs: performance.now() - receiptPublishedAt })),
Bun.sleep(1_500).then(() => undefined),
]);
expect(outcome).toBeDefined();
Expand Down Expand Up @@ -4387,7 +4388,7 @@ test("broker directly resumes and forks a canonical cold saved session with scop
() => true,
() => false,
),
).toBe(true);
).toBe(false);
expect(await broker.handleRequest("session.get_endpoint", { sessionId: sourceId })).toMatchObject({
ok: false,
error: { code: "resource_gone" },
Expand Down Expand Up @@ -4455,7 +4456,7 @@ test("broker directly resumes and forks a canonical cold saved session with scop
() => true,
() => false,
),
).toBe(true);
).toBe(false);
expect(await broker.handleRequest("session.get_endpoint", { sessionId: forkId })).toMatchObject({
ok: false,
error: { code: "resource_gone" },
Expand Down Expand Up @@ -4588,7 +4589,7 @@ test("broker replays one identity-bound lifecycle metadata cleanup plan after th
const markerPath = path.join(stateRoot, "sdk", `${sessionId}.lifecycle.json`);
const readyPath = path.join(stateRoot, "sdk", `${sessionId}.lifecycle.ready.json`);
await expect(fs.stat(markerPath)).resolves.toBeDefined();
await expect(fs.stat(readyPath)).resolves.toBeDefined();
await expect(fs.stat(readyPath)).rejects.toThrow();
setLifecycleCleanupHookForTest(crashing, () => {});
const deleteInput = { cwd: root, stateRoot, sessionId, sessionPath };
await expect(
Expand Down
Loading
Loading