diff --git a/packages/plugins/sdk/README.md b/packages/plugins/sdk/README.md index 29da17228ed..07bf77b16c5 100644 --- a/packages/plugins/sdk/README.md +++ b/packages/plugins/sdk/README.md @@ -1566,3 +1566,26 @@ The return value is transient raw credential material with the Connector's existing permissions. This capability is for trusted execution infrastructure, not agent-facing tools or UI. Never expose or persist it. See [GitHub identity](../../../doc/execution-github-identity.md#project-authorization-for-background-repository-work). + +## Login on an existing computer + +A driver can use the existing login PTY hooks for a login on a saved agent's +computer. Core sends `purpose: "login"`, a closed `loginCommandKey`, a stable +`creationOperationId`, and an absolute `requestedExpiresAt` to lease acquisition. +Release and destroy must retire only that login allocation and preserve the +computer. Omitted purpose retains the existing execution behavior. + +Core supplies `expiresAt` to the PTY open and an `inputId` for each input write. +Input, stop, and close retain the opening company context so provider callbacks +can use company-scoped host services during cleanup. +A provider that retries delivery must retain the input identity and bytes, and +reject reuse of the identity with different input. A failed input call ends the +Core login; Core does not silently replay an unconfirmed write. + +Providers can implement `onLoginPtyReadCredential` to export the fixed Codex +credential artifact for a bound login lease. The result is +`{ kind: "codex-auth-json", authJson }`; the request accepts no path or command. +The worker advertises this optional method only when the hook exists. The host +must check that method before using this read path, enforce a size limit, and +keep credential bytes out of logs and public session responses. Existing sandbox +providers can continue to use their descriptor-bound credential read. diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index 18edbc1db9d..eb6530c0067 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -94,6 +94,8 @@ import type { PluginLoginPtyOpenParams, PluginLoginPtyOpenResult, PluginLoginPtyInputParams, + PluginLoginPtyReadCredentialParams, + PluginLoginPtyReadCredentialResult, PluginLoginPtyStopParams, PluginLoginPtyCloseParams, PluginLoginPtyCloseResult, @@ -523,6 +525,9 @@ export interface PluginDefinition { /** Called to write delayed input to an open login pseudo-terminal, keyed by the worker session identifier. */ onLoginPtyInput?(params: PluginLoginPtyInputParams): Promise; + /** Optional fixed credential read, advertised through supported-method discovery. */ + onLoginPtyReadCredential?(params: PluginLoginPtyReadCredentialParams): Promise; + /** Called to stop an open login pseudo-terminal child, keyed by the worker session identifier. */ onLoginPtyStop?(params: PluginLoginPtyStopParams): Promise; diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 1cf41d3241f..20e52d9e4ad 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -214,6 +214,15 @@ export type { PluginEnvironmentCancelInteractiveSetupResult, PluginEnvironmentDeleteTemplateParams, PluginEnvironmentDeleteTemplateResult, + PluginLoginCommandKey, + PluginLoginPtyOpenParams, + PluginLoginPtyOpenResult, + PluginLoginPtyInputParams, + PluginLoginPtyReadCredentialParams, + PluginLoginPtyReadCredentialResult, + PluginLoginPtyStopParams, + PluginLoginPtyCloseParams, + PluginLoginPtyCloseResult, PluginModalBoundsRequest, PluginRenderCloseEvent, PluginLauncherRenderContextSnapshot, diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 0ad0814b2fd..ff3a43d51a8 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -646,6 +646,11 @@ export interface PluginEnvironmentRunnerIngressEndpointParams } export interface PluginEnvironmentAcquireLeaseParams extends PluginEnvironmentDriverBaseParams { + /** Omitted means ordinary execution. A login allocation targets a saved agent, + * requires requestedExpiresAt, and releases only that login, never its computer. */ + purpose?: "run" | "login"; + /** Host-resolved fixed command identity for a login allocation. */ + loginCommandKey?: PluginLoginCommandKey; runId: string; creationOperationId?: string; /** Explicit platform probes may request a provider-owned immutable snapshot @@ -1086,6 +1091,8 @@ export type PluginLoginCommandKey = "claude" | "codex" | "grok"; /** The open request for one live login pseudo-terminal. The worker registers the terminal by `hostRouteId`. */ export interface PluginLoginPtyOpenParams { + /** Independent absolute deadline, required by providers hosting login on an existing computer. */ + expiresAt?: string; /** The host-owned opaque route identifier. The worker registers the terminal by it. */ hostRouteId: string; /** The environment driver key, for the worker sandbox scope. It routes the worker; it confers no command authority. */ @@ -1117,20 +1124,42 @@ export interface PluginLoginPtyOpenResult { /** The input request. It carries the worker session identifier and the raw input bytes. */ export interface PluginLoginPtyInputParams { + /** Host-bound company context for provider callbacks during this lifecycle call. */ + companyId?: string; + /** Stable identity of this input write. Retries retain the identity and bytes. */ + inputId?: string; /** The worker session identifier that the open reply returned. */ workerSessionId: string; /** The raw input bytes to write to the terminal. */ data: string; } +/** Read the fixed Codex result of a bound login allocation; never an arbitrary path. */ +export interface PluginLoginPtyReadCredentialParams { + driverKey: string; + companyId: string; + environmentId: string; + providerLeaseId: string; + loginCommandKey: "codex"; +} + +export interface PluginLoginPtyReadCredentialResult { + kind: "codex-auth-json"; + authJson: string; +} + /** The stop request. It carries the worker session identifier. */ export interface PluginLoginPtyStopParams { + /** Host-bound company context for provider callbacks during this lifecycle call. */ + companyId?: string; /** The worker session identifier that the open reply returned. */ workerSessionId: string; } /** The close request. The host route identifier is the authoritative key. */ export interface PluginLoginPtyCloseParams { + /** Host-bound company context for provider callbacks during this lifecycle call. */ + companyId?: string; /** * The host-owned opaque route identifier. This is the authoritative close key, * so the host closes the terminal even when no worker session identifier @@ -1498,6 +1527,8 @@ export interface HostToWorkerMethods { ]; /** Write delayed input to a live login pseudo-terminal, keyed by the worker session identifier. */ loginPtyInput: [params: PluginLoginPtyInputParams, result: void]; + /** Read the fixed credential artifact after the bound login process succeeds. */ + loginPtyReadCredential: [params: PluginLoginPtyReadCredentialParams, result: PluginLoginPtyReadCredentialResult]; /** Stop a live login pseudo-terminal child, keyed by the worker session identifier. */ loginPtyStop: [params: PluginLoginPtyStopParams, result: void]; /** Close a live login pseudo-terminal by the host route identifier and return a bound acknowledgement. */ @@ -1572,6 +1603,7 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "environmentDeleteTemplate", "loginPtyOpen", "loginPtyInput", + "loginPtyReadCredential", "loginPtyStop", "loginPtyClose", "duplexChannelOpen", diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index ec3f1ff2aa5..f6b801fc085 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -121,6 +121,7 @@ import type { PluginEnvironmentDeleteTemplateParams, PluginLoginPtyOpenParams, PluginLoginPtyInputParams, + PluginLoginPtyReadCredentialParams, PluginLoginPtyStopParams, PluginLoginPtyCloseParams, PluginDuplexChannelOpenParams, @@ -1807,6 +1808,9 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost case "loginPtyInput": return handleLoginPtyInput(params as PluginLoginPtyInputParams); + case "loginPtyReadCredential": + return handleLoginPtyReadCredential(params as PluginLoginPtyReadCredentialParams); + case "loginPtyStop": return handleLoginPtyStop(params as PluginLoginPtyStopParams); @@ -1892,6 +1896,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onEnvironmentDeleteTemplate) supportedMethods.push("environmentDeleteTemplate"); if (plugin.definition.onLoginPtyOpen) supportedMethods.push("loginPtyOpen"); if (plugin.definition.onLoginPtyInput) supportedMethods.push("loginPtyInput"); + if (plugin.definition.onLoginPtyReadCredential) supportedMethods.push("loginPtyReadCredential"); if (plugin.definition.onLoginPtyStop) supportedMethods.push("loginPtyStop"); if (plugin.definition.onLoginPtyClose) supportedMethods.push("loginPtyClose"); if (plugin.definition.onDuplexChannelOpen) supportedMethods.push("duplexChannelOpen"); @@ -2269,6 +2274,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost return plugin.definition.onLoginPtyInput(params); } + async function handleLoginPtyReadCredential(params: PluginLoginPtyReadCredentialParams) { + if (!plugin.definition.onLoginPtyReadCredential) throw methodNotImplemented("loginPtyReadCredential"); + return plugin.definition.onLoginPtyReadCredential(params); + } + async function handleLoginPtyStop(params: PluginLoginPtyStopParams) { if (!plugin.definition.onLoginPtyStop) { throw methodNotImplemented("loginPtyStop"); diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index 9aafcf5a26b..723020a812a 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -403,6 +403,23 @@ describe("worker configChanged cross-tenant guard", () => { return { callWorker, initialize, stop }; } + it.each([false, true])("advertises a fixed login credential read only with an explicit hook: %s", async supported => { + const params = { driverKey: "computer", companyId: "company", environmentId: "environment", providerLeaseId: "login", loginCommandKey: "codex" as const }; + const worker = makeWorker(definePlugin({ + async setup() {}, + ...(supported ? { async onLoginPtyReadCredential(input: typeof params) { + expect(input).toEqual(params); + return { kind: "codex-auth-json" as const, authJson: '{"fixture":true}' }; + } } : {}), + })); + try { + const initialized = await worker.initialize() as { supportedMethods: string[] }; + expect(initialized.supportedMethods.includes("loginPtyReadCredential")).toBe(supported); + if (supported) expect(await worker.callWorker("loginPtyReadCredential", params)).toEqual({ kind: "codex-auth-json", authJson: '{"fixture":true}' }); + else await expect(worker.callWorker("loginPtyReadCredential", params)).rejects.toMatchObject({ code: PLUGIN_RPC_ERROR_CODES.METHOD_NOT_IMPLEMENTED }); + } finally { worker.stop(); } + }); + it.each([false, true])("advertises and dispatches stop-only only with an explicit hook: %s", async supported => { let releases = 0, stops = 0; const worker = makeWorker(definePlugin({ async setup() {}, @@ -900,6 +917,7 @@ describe("worker setup-token pseudo-terminal dispatch", () => { return { workerSessionId: "ws-1" }; }, async onLoginPtyInput(params) { + expect(params.inputId).toBe("input-1"); inputs.push(params.data); }, async onLoginPtyStop() { @@ -986,7 +1004,7 @@ describe("worker setup-token pseudo-terminal dispatch", () => { // The worker streams output as a notification bound to the worker session id. emitOutput?.("prompt output"); - await callWorker("loginPtyInput", { workerSessionId: "ws-1", data: "browser-code" }); + await callWorker("loginPtyInput", { workerSessionId: "ws-1", data: "browser-code", inputId: "input-1" }); await callWorker("loginPtyStop", { workerSessionId: "ws-1" }); resolveWait?.({ exitCode: 0 }); await expect( diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index d2513eca283..99639d300a7 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -6288,11 +6288,13 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { it("forwards a requested expiry to the acquire RPC and records only a provider-attested expiry", async () => { const pluginId = randomUUID(); let receivedRequestedExpiresAt: unknown; + let receivedPurpose: unknown, receivedCommand: unknown; const workerManager = { isRunning: vi.fn(() => true), call: vi.fn(async (_pluginId: string, method: string, params: Record) => { if (method === "environmentAcquireLease") { receivedRequestedExpiresAt = params.requestedExpiresAt; + receivedPurpose = params.purpose; receivedCommand = params.loginCommandKey; // The provider grants no expiry. return { providerLeaseId: "plugin-lease-no-expiry", metadata: { remoteCwd: "/workspace" } }; } @@ -6310,10 +6312,12 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { heartbeatRunId: runId, persistedExecutionWorkspace: null, requestedExpiresAt, + purpose: "login", loginCommandKey: "codex", }); // The acquire RPC receives the requested expiry as an ISO 8601 string. expect(receivedRequestedExpiresAt).toBe(requestedExpiresAt.toISOString()); + expect(receivedPurpose).toBe("login"); expect(receivedCommand).toBe("codex"); // The provider gave no expiry, so the lease row records no expiry. The runtime // never synthesizes the requested deadline as evidence of provider enforcement. expect(acquired.lease.expiresAt ?? null).toBeNull(); diff --git a/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs b/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs index b8ffe6d0c1a..87383b2aefb 100644 --- a/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs @@ -146,6 +146,16 @@ function scriptedOutputLines(directive, hostRouteId, workerSessionId) { // The registered terminals, keyed by the host route id. Each entry records the // bound worker session id and the close directive. const routes = new Map(); +const pendingHost = new Map(); +let nextHostCall = 0; +function probeCompanyScope(message, entry) { + const id = `scope-${++nextHostCall}`; + return new Promise((resolve, reject) => { + pendingHost.set(id, { resolve, reject }); + send({ jsonrpc: "2.0", id, method: "companies.get", params: { companyId: entry.companyId, stage: message.method }, + ...(message.paperclipInvocation ? { paperclipInvocationId: message.paperclipInvocation.id } : {}) }); + }); +} function parseDirective(raw) { try { @@ -158,12 +168,24 @@ function parseDirective(raw) { const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); -rl.on("line", (line) => { +rl.on("line", async (line) => { if (!line.trim()) return; const message = JSON.parse(line); + if (pendingHost.has(message.id)) { + const pending = pendingHost.get(message.id); pendingHost.delete(message.id); + if (message.error) pending.reject(new Error("Host company scope denied")); else pending.resolve(message.result); + return; + } const method = message && typeof message.method === "string" ? message.method : null; const params = message.params ?? {}; + if (["loginPtyInput", "loginPtyStop", "loginPtyClose"].includes(method)) { + const entry = method === "loginPtyClose" ? routes.get(params.hostRouteId) : [...routes.values()].find(value => value.workerSessionId === params.workerSessionId); + if (entry?.probeLifecycle) { + try { await probeCompanyScope(message, entry); } + catch { send({ jsonrpc: "2.0", id: message.id, error: { code: -32000, message: "Host company scope denied" } }); return; } + } + } if (method === "initialize") { send({ jsonrpc: "2.0", @@ -188,7 +210,9 @@ rl.on("line", (line) => { const closeMode = directive.closeMode ?? "ack"; routes.set(params.hostRouteId, { workerSessionId, + companyId: params.companyId, probeLifecycle: directive.probeLifecycle === true, closeMode, + rejectInput: directive.rejectInput === true, pendingOutput: directive.emitOnInput === true ? scriptedOutputLines(directive, params.hostRouteId, workerSessionId) : null, @@ -253,6 +277,14 @@ rl.on("line", (line) => { // test proves the input reaches the worker and the output routes back. for (const [hostRouteId, entry] of routes.entries()) { if (entry.workerSessionId === params.workerSessionId) { + if (entry.rejectInput) { + send({ jsonrpc: "2.0", id: message.id, error: { code: -32000, message: "input delivery unconfirmed" } }); + return; + } + if (typeof params.inputId !== "string" || params.inputId.length === 0) { + send({ jsonrpc: "2.0", id: message.id, error: { code: -32000, message: "input identity missing" } }); + return; + } if (entry.pendingOutput !== null) { process.stdout.write(entry.pendingOutput); entry.pendingOutput = null; diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index 5f8ee1d604c..fbc6bda25ec 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -1336,6 +1336,64 @@ describe("plugin worker manager setup-token pty route gate", () => { } }); + it("retains the opening company for scoped host calls during input, stop, and close", async () => { + const stages: string[] = []; + const handle = makeLoginPtyHandle({ hostHandlers: { "companies.get": async ( + params: { companyId: string; stage: string }, context?: { invocationScope?: { companyId?: string } }, + ) => { + if (context?.invocationScope?.companyId !== params.companyId || params.companyId !== "company-1") throw new Error("Host company scope denied"); + stages.push(params.stage); + return { id: params.companyId }; + } } }); + try { + await handle.start(); + const session = await handle.openLoginPtySession(ptyOpenInput({ probeLifecycle: true })); + session.write("browser-code"); + await vi.waitFor(() => expect(stages).toEqual(["loginPtyInput"])); + session.kill(); + await vi.waitFor(() => expect(stages).toEqual(["loginPtyInput", "loginPtyStop"])); + await session.close(); + expect(stages).toEqual(["loginPtyInput", "loginPtyStop", "loginPtyClose"]); + } finally { await handle.stop().catch(() => undefined); } + }); + + it("fails the login when input delivery is unconfirmed", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + const session = await handle.openLoginPtySession(ptyOpenInput({ rejectInput: true })); + session.write("browser-code"); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + await session.close(); + } finally { await handle.stop().catch(() => undefined); } + }); + + it("joins an in-flight close after input failure before releasing the login", async () => { + let releaseClose!: () => void; + const heldClose = new Promise(resolve => { releaseClose = resolve; }); + let closeCalls = 0; + const handle = makeLoginPtyHandle({ hostHandlers: { "companies.get": async (params: { stage: string }) => { + if (params.stage === "loginPtyClose") { closeCalls++; await heldClose; } + return { id: "company-1" }; + } } }); + try { + await handle.start(); + const session = await handle.openLoginPtySession(ptyOpenInput({ rejectInput: true, probeLifecycle: true })); + session.write("browser-code"); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + await vi.waitFor(() => expect(closeCalls).toBe(1)); + let completed = 0; + const first = session.close().then(() => { completed++; }); + const second = session.close().then(() => { completed++; }); + await new Promise(resolve => setImmediate(resolve)); + expect(completed).toBe(0); + releaseClose(); + await Promise.all([first, second]); + expect(completed).toBe(2); + expect(closeCalls).toBe(1); + } finally { releaseClose(); await handle.stop().catch(() => undefined); } + }); + it("terminalizes the route when the cumulative output passes the per-route bound", async () => { const handle = makeLoginPtyHandle({ loginPtyLimits: { maxTotalChars: 10 }, diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 76b590c0d8f..2d5dd44d074 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -454,6 +454,8 @@ function stripSecretRefValuesFromPluginLeaseMetadata(input: { } export interface EnvironmentDriverAcquireInput { + purpose?: "run" | "login"; + loginCommandKey?: import("@paperclipai/plugin-sdk").PluginLoginCommandKey; companyId: string; environment: Environment; /** Saved environment before run-only lease retention settings. */ @@ -3364,7 +3366,7 @@ function createPluginEnvironmentDriver( driver.warmTaskVersion === 1 && !!input.issueId && !!input.agentId && !!input.heartbeatRunId && workerManager.getWorker(plugin.id)?.supportedMethods.includes("environmentResumeLease"); const lifecyclePolicy = warm ? { mode: "warm", idleTimeoutMs: parsed.config.runnerIdleTimeoutMs ?? 300_000 } : { mode: "per_turn", idleTimeoutMs: null }; - const fingerprint = createHash("sha256").update(JSON.stringify([parsed.config, input.agentId, input.runnerTaskProjectIds ?? []])).digest("hex"); + const fingerprint = createHash("sha256").update(JSON.stringify([parsed.config, input.agentId, input.runnerTaskProjectIds ?? [], ...(input.purpose ? [input.purpose, input.loginCommandKey] : [])])).digest("hex"); const candidates = warm ? await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environment.id), eq(environmentLeases.status, "retained"), eq(environmentLeases.issueId, input.issueId!), @@ -3405,6 +3407,7 @@ function createPluginEnvironmentDriver( config: parsed.config.driverConfig, runId: identity.runId, creationOperationId: identity.id, + ...(input.purpose ? { purpose: input.purpose, loginCommandKey: input.loginCommandKey } : {}), workspaceMode: input.executionWorkspaceMode ?? undefined, agentId: input.agentId ?? undefined, executionWorkspaceId: input.executionWorkspaceId ?? undefined, @@ -3436,8 +3439,9 @@ function createPluginEnvironmentDriver( ...(warm ? {runnerTaskProjectIds:[...input.runnerTaskProjectIds ?? []]} : {}),remoteCwd:".", ...(input.agentId ? { agentId: input.agentId } : {}), providerMetadata: providerLease.metadata ?? {}, + ...(input.purpose ? { purpose: input.purpose, loginCommandKey: input.loginCommandKey } : {}), environmentId: input.environment.id, - ...(input.adapterType === "paperclip_runner" && driver.supportsTasks === true ? { runnerTaskLease: true } : {}), + ...(input.purpose !== "login" && input.adapterType === "paperclip_runner" && driver.supportsTasks === true ? { runnerTaskLease: true } : {}), driver: input.environment.driver, executionWorkspaceMode: input.executionWorkspaceMode, pluginId: plugin.id, @@ -3845,6 +3849,8 @@ export function environmentRuntimeService( }, async acquireRunLease(input: { + purpose?: "run" | "login"; + loginCommandKey?: import("@paperclipai/plugin-sdk").PluginLoginCommandKey; companyId: string; environment: Environment; /** Saved environment before run-only lease retention settings. */ @@ -3901,6 +3907,8 @@ export function environmentRuntimeService( executionWorkspaceMode: leaseContext.executionWorkspaceMode, executionWorkspaceSettings: input.executionWorkspaceSettings ?? null, runnerTaskProjectIds: input.runnerTaskProjectIds, + purpose: input.purpose, + loginCommandKey: input.loginCommandKey, adapterType: input.adapterType ?? null, resolveSnapshotIdentity: input.resolveSnapshotIdentity, applyCustomImageTemplate: input.applyCustomImageTemplate ?? false, diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 723eb327485..38ed65d8a53 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -590,6 +590,7 @@ export type ExecuteLogSink = ( * carries no command string, so a caller cannot select the command. */ export interface LoginPtyOpenInput { + expiresAt?: string; /** The environment driver key. It routes the worker; it confers no command authority. */ driverKey: string; companyId: string; @@ -1564,7 +1565,9 @@ export function createPluginWorkerHandle( type LoginPtyPreBindRecord = LoginPtyPreBindOutputRecord | LoginPtyPreBindExitRecord; interface LoginPtyRoute { + companyId: string; hostRouteId: string; + closePromise: Promise | null; state: LoginPtyRouteState; workerSessionId: string | null; listener: ((chunk: string) => void) | null; @@ -1626,11 +1629,11 @@ export function createPluginWorkerHandle( // that carries the exact host route identifier. An absent, malformed, // mismatched, or timed-out acknowledgement returns false, so the caller fails // closed. - async function closeLoginPtyTerminal(hostRouteId: string): Promise { + async function closeLoginPtyTerminal(hostRouteId: string, companyId: string): Promise { try { const ack = await callInternal( "loginPtyClose", - { hostRouteId }, + { hostRouteId, companyId }, loginPtyCloseTimeoutMs, ); return isRecord(ack) && readNonEmptyString(ack.hostRouteId) === hostRouteId; @@ -1647,6 +1650,9 @@ export function createPluginWorkerHandle( // every other route this worker still holds settles through the worker-exit // path that follows. async function terminalizeLoginPtyRoute(route: LoginPtyRoute): Promise { + // A notification may already have started close. Every later disposer must + // join that acknowledgement before its owner can release the allocation. + if (route.closePromise) return route.closePromise; if (route.terminalized) return; route.terminalized = true; route.state = "closed"; @@ -1660,20 +1666,22 @@ export function createPluginWorkerHandle( loginPtyRoutesByWorkerSessionId.delete(route.workerSessionId); } releaseLoginPtyRouteSlot(route); - // A terminalized route reports a null exit code, which the runner treats as a - // failure. + route.closePromise = (async () => { + const confirmed = await closeLoginPtyTerminal(route.hostRouteId, route.companyId); + if (!confirmed) { + // The worker did not acknowledge the close, so the host cannot prove the + // terminal is gone. Retire the worker before any reuse. + log.error( + { pluginId }, + "login pseudo-terminal close not acknowledged; retiring worker", + ); + retireWorkerProcess(); + } + })(); + // Report failure promptly, but retain the shared close barrier so the + // runner's following dispose waits for confirmed stop (or worker retirement). settleRouteWait(route, { exitCode: null }); - const confirmed = await closeLoginPtyTerminal(route.hostRouteId); - if (!confirmed) { - // The worker did not acknowledge the close, so the host cannot prove the - // terminal is gone. Fail closed: retire the worker before any reuse. This - // also settles and clears every other route the worker still holds. - log.error( - { pluginId }, - "login pseudo-terminal close not acknowledged; retiring worker", - ); - retireWorkerProcess(); - } + return route.closePromise; } // Resolve one login pseudo-terminal notification to its route. Return null @@ -1920,7 +1928,9 @@ export function createPluginWorkerHandle( settleWait = resolve; }); const route: LoginPtyRoute = { + companyId: input.companyId, hostRouteId, + closePromise: null, state: "reserved", workerSessionId: null, listener: null, @@ -1957,6 +1967,7 @@ export function createPluginWorkerHandle( providerLeaseId: input.providerLeaseId, loginCommandKey: input.loginCommandKey, sessionHome: input.sessionHome, + ...(input.expiresAt ? { expiresAt: input.expiresAt } : {}), }, loginPtyOpenTimeoutMs, ); @@ -2001,9 +2012,9 @@ export function createPluginWorkerHandle( if (route.state !== "open" || !sid) return; void callInternal( "loginPtyInput", - { workerSessionId: sid, data }, + { companyId: route.companyId, workerSessionId: sid, data, inputId: randomUUID() }, loginPtyOpenTimeoutMs, - ).catch(() => {}); + ).catch(() => terminalizeLoginPtyRoute(route)); }, wait(): Promise<{ exitCode: number | null }> { return waitPromise; @@ -2013,7 +2024,7 @@ export function createPluginWorkerHandle( if (!sid) return; void callInternal( "loginPtyStop", - { workerSessionId: sid }, + { companyId: route.companyId, workerSessionId: sid }, loginPtyOpenTimeoutMs, ).catch(() => {}); },