fix(daytona): retain duplex close ownership after lease cleanup (#19150)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Remote agents use provider plugins to manage sandbox leases and
duplex channels.
> - The host verifies the exact route and session when a bound channel
closes.
> - Daytona lease teardown removed that binding before a later host
close.
> - The incomplete reply made the host retire the shared worker. A later
command then failed during setup.
> - This pull request retains bounded close receipts and joins the
original cleanup.
> - Later work can use the same worker while invalid or unconfirmed
closes still fail closed.

## Linked Issues or Issue Description

**What happened?**

A normal lease teardown can close a duplex channel before the host
closes its bound route. Daytona then replies with only the route ID. The
host requires the original session ID too. It logs `duplex channel close
not acknowledged; retiring worker`, stops the worker, and rejects a
later `environmentExecute` call as stopped.

**Expected behavior**

A later close must identify the original route and session and join its
existing cleanup. It must not revive the channel or restart the worker.
Unknown ownership and an unconfirmed close must retain the existing
failure behavior.

**Steps to reproduce**

1. Acquire two leases through one Daytona provider worker.
2. Open a duplex channel on the first lease.
3. Release that lease, then close the bound channel through the host.
4. Execute a command on the second lease.

**Paperclip version or commit**

The real host/worker regression fails on
`69dea878c0b920b335594debb0286443ba9f6c35`. This branch includes master
`3a72efebe5b050e7594bb309e71a0d5f3072db00`.

**Deployment mode**

Managed Daytona sandbox execution in Cloud. The regression uses a local
real host and SDK worker. Only the external Daytona SDK and PTY are
simulated.

Related work: Refs #19004, Refs #10429, Refs #18065, Refs #19050. These
cover task Retry, managed worker startup, drain release, and stale
provider sessions.

## What Changed

- Retain at most 4096 retired route/session bindings with their original
close promises.
- Remove live channel authority before cleanup starts. Concurrent and
later closes join the same promise and return the stored binding.
- Reject reuse of a retained route ID. Keep unknown and evicted routes
unable to provide a session acknowledgement.
- Add a real child-process host/SDK-worker/Daytona regression with only
the external Daytona SDK and PTY simulated.
- Cover release, cancellation, stop, destroy, resume, a stalled close,
concurrent closes, failed cleanup, forged IDs and bounded eviction.
- Document close receipt scope and its separate provider-termination
checks.

## Verification

- RED before the repair: the actual host and SDK worker reproduced the
retirement log, SIGKILL, stopped state, and later `environmentExecute`
rejection. The direct provider cases also failed because the session ID
was missing.
- Combined-branch provider suite: 364 passed.
- Combined-branch host manager, duplex, managed readiness and composed
suites: 142 passed. The final test-only manifest-loading change passed
all 6 composed cases again.
- Full repository typecheck (`node scripts/typecheck-low-memory.mjs`),
focused server types and the standalone Daytona build passed. The
provider build uses pinned SDK `0.203.0` and TypeScript `5.9.3`.
- The composed cases check the same worker/PID, one later command,
preserved saved content, revoked channel writes/stops, and continued
rejection of commands on closed leases. A stalled close still retires
the worker and dispatches no later command.
- Reproduce with `pnpm --dir server exec vitest run --config
vitest.config.ts
src/__tests__/plugin-worker-manager-daytona-close.test.ts` from a
checkout with the SDK build dependencies prepared. Run the provider
suite with its own Vitest config.
- An initial local typecheck lacked the excluded providers' SDK links
after an install with scripts disabled. The standard root postinstall
linker restored those local links, and full typechecking then passed.
The fixture loads its actual built manifest. Failed setup output is
retained; no type configuration was relaxed.
- The first two composed attempts stopped during fixture setup. They are
not the behavioral RED above. Their failed outputs are retained.
- The first broad local typecheck stopped at an IPC permission error
during preflight. It passed after rerunning with allowed local IPC and
the standard SDK links.
- [Full repository
CI](https://github.com/paperclipai/paperclip/actions/runs/41945022881)
passed all 48 jobs on this head. Selection fell back to the complete
task plan, so the run was unfiltered. The complete check inventory has
54 successes and two expected Storybook skips; the separate Snyk status
passed. Fresh Greptile review on
`daebbeb5005b9d9bd59d32ea30094279048b9213` is [5/5 with no actionable
comments](https://github.com/paperclipai/paperclip/pull/19150#issuecomment-6646621914).
Independent production and fixture review found no actionable findings.
Hosted acceptance of this repair has not run.

## Risks

- A receipt removed at the 4096-entry bound cannot confirm an old bound
close. The existing host guard remains effective in that case.
- Close acknowledgements identify channels. They do not certify remote
process termination or clear unknown external work. Provider termination
receipts and existing recovery gates remain in force.
- Hosted evidence confirms the unacknowledged-close retirement class.
The historical log does not distinguish a missing session ID from a
timeout or another close error. The regression proves and repairs the
missing-binding path; it also preserves timeout retirement.
- The provider is outside the main pnpm workspace. A new Cloud image
must compile its changed source. The unchanged plugin version alone does
not prove activation.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, source editing and tool
execution. The exact model ID and context window are not exposed in this
session.

## 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
#` / `Refs #` 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>

packages/plugins/sandbox-providers/daytona/src/plugin.test.ts

@@ -40,6 +40,7 @@ import type { PluginContext } from "@paperclipai/plugin-sdk";
 import { environmentSyncErrorData, readEnvironmentSyncErrorDiagnostic, withEnvironmentSyncErrorCapture, environmentCreationCleanupErrorData, readEnvironmentCreationCleanupError, readEnvironmentAcquisitionDiagnostic } from "@paperclipai/plugin-sdk";
 import manifest from "./manifest.js";
 import { parseTarVerboseListingLine, splitLinkEntryOnce } from "./file-sync.js";
+import * as duplexCommands from "./duplex-command-stream.js";
 
 describe("Daytona read-only resource inspection", () => {
   const params = { driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment",
@@ -729,6 +730,8 @@ describe("Daytona sandbox provider plugin", () => {
       });
       expect(killed).toBeGreaterThanOrEqual(1);
       expect(disconnected).toBe(1);
+      await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: "route-1" }))
+        .resolves.toEqual({ hostRouteId: "route-1", workerSessionId });
 
       // The channel entry is gone, so a later write is a no-op and reaches no
       // process.
@@ -746,6 +749,94 @@ describe("Daytona sandbox provider plugin", () => {
     }
   });
 
+  describe("retired duplex ownership", () => {
+    async function fixture() {
+      const params = { driverKey: "daytona", companyId: "company-1", environmentId: "env-1",
+        config: { apiKey: "fixture-key", image: "fixture-image", timeoutMs: 3_000, livenessTimeoutMs: 1_000, reuseLease: true } };
+      const pty = { waitForConnection: vi.fn(async () => {}), sendInput: vi.fn(async () => {}),
+        wait: vi.fn(() => new Promise<{ exitCode: number }>(() => {})),
+        kill: vi.fn(async () => {}), disconnect: vi.fn(async () => {}) };
+      const sandbox = createMockSandbox();
+      const createPty = vi.fn(async () => pty);
+      Object.assign(sandbox.process, { createPty });
+      mockCreate.mockResolvedValue(sandbox);
+      await plugin.definition.onEnvironmentAcquireLease!({ ...params, runId: "run-1" });
+      const opening = { driverKey: params.driverKey, companyId: params.companyId, environmentId: params.environmentId,
+        providerLeaseId: sandbox.id, command: ["fixture-bridge"] };
+      return { params: { ...params, providerLeaseId: sandbox.id }, opening, pty, createPty };
+    }
+
+    it("joins pending teardown for concurrent closes while all channel authority stays revoked", async () => {
+      const f = await fixture();
+      const opened = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "pending-route" });
+      let finishKill!: () => void, enteredKill!: () => void;
+      const entered = new Promise<void>(resolve => { enteredKill = resolve; });
+      const killing = new Promise<void>(resolve => { finishKill = resolve; });
+      f.pty.kill.mockImplementation(async () => { enteredKill(); await killing; });
+      const requests: Promise<unknown>[] = [];
+      try {
+        requests.push(plugin.definition.onEnvironmentReleaseLease!(f.params));
+        await entered;
+        let replies = 0;
+        for (let index = 0; index < 2; index++) requests.push(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId })
+          .then(value => { replies++; return value; }));
+        await new Promise<void>(resolve => setImmediate(resolve));
+        expect(replies).toBe(0);
+        expect(f.pty.kill).toHaveBeenCalledTimes(1);
+        expect(f.pty.disconnect).not.toHaveBeenCalled();
+        await plugin.definition.onDuplexChannelWrite!({ ...opened, data: Buffer.from("closed").toString("base64") });
+        await plugin.definition.onDuplexChannelStop!(opened);
+        expect(f.pty.sendInput).toHaveBeenCalledTimes(1);
+        expect(f.pty.kill).toHaveBeenCalledTimes(1);
+        await expect(plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: opened.hostRouteId }))
+          .rejects.toThrow("already been used");
+        expect(f.createPty).toHaveBeenCalledTimes(1);
+        finishKill();
+        const result = await Promise.all(requests);
+        expect(result.slice(1)).toEqual([opened, opened]);
+        await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId })).resolves.toEqual(opened);
+        expect(f.pty.kill).toHaveBeenCalledTimes(1);
+        expect(f.pty.disconnect).toHaveBeenCalledTimes(1);
+        await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: "unknown-route", workerSessionId: opened.workerSessionId } as never))
+          .resolves.toEqual({ hostRouteId: "unknown-route" });
+      } finally { finishKill(); await Promise.allSettled(requests); }
+    });
+
+    it("does not turn a rejected retired close into an acknowledgement", async () => {
+      const f = await fixture();
+      const close = vi.fn().mockRejectedValue(new Error("fixture closure remains unknown"));
+      const restore = vi.spyOn(duplexCommands, "openDaytonaDuplexChannelSession").mockResolvedValueOnce({
+        onData() {}, write: vi.fn(), kill: vi.fn(), wait: () => new Promise(() => {}), close,
+      });
+      try {
+        const opened = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "failed-route" });
+        await plugin.definition.onEnvironmentReleaseLease!(f.params);
+        await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId }))
+          .rejects.toThrow("fixture closure remains unknown");
+        expect(close).toHaveBeenCalledTimes(1);
+        await expect(plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: opened.hostRouteId }))
+          .rejects.toThrow("already been used");
+      } finally { restore.mockRestore(); }
+    });
+
+    it("bounds retained acknowledgements and never echoes a supplied session id for evicted routes", async () => {
+      const f = await fixture();
+      const first = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "oldest-route" });
+      await plugin.definition.onDuplexChannelClose!({ hostRouteId: first.hostRouteId });
+      let latest = first;
+      for (let index = 0; index < 4_096; index++) {
+        latest = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: `bounded-route-${index}` });
+        await plugin.definition.onDuplexChannelClose!({ hostRouteId: latest.hostRouteId });
+      }
+      await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: first.hostRouteId, workerSessionId: first.workerSessionId } as never))
+        .resolves.toEqual({ hostRouteId: first.hostRouteId });
+      await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: latest.hostRouteId, workerSessionId: "forged-session" } as never))
+        .resolves.toEqual(latest);
+      expect(f.pty.kill).toHaveBeenCalledTimes(4_097);
+      expect(f.pty.disconnect).toHaveBeenCalledTimes(4_097);
+    });
+  });
+
   it("normalizes config and validates the API key fallback", async () => {
     process.env.DAYTONA_API_KEY = "host-key";
 

packages/plugins/sandbox-providers/daytona/src/plugin.ts

@@ -1728,6 +1728,7 @@ export function __resetDaytonaSandboxHandleCacheForTest(): void {
   sandboxHandleWritableDirs.reset();
   sandboxHandleSessionStore.reset();
   runnerIngressGenerationStore.reset();
+  retiredDaytonaDuplexChannelByRoute.clear();
 }
 
 /**
@@ -2489,12 +2490,24 @@ interface DaytonaDuplexChannelEntry {
 }
 const daytonaDuplexChannelByRoute = new Map<string, DaytonaDuplexChannelEntry>();
 const daytonaDuplexChannelBySession = new Map<string, DaytonaDuplexChannelEntry>();
+const retiredDaytonaDuplexChannelByRoute = new Map<string, { workerSessionId: string; closed: Promise<void> }>();
+const MAX_RETIRED_DAYTONA_DUPLEX_CHANNELS = 4096;
 
 function forgetDaytonaDuplexChannel(entry: DaytonaDuplexChannelEntry): void {
   daytonaDuplexChannelByRoute.delete(entry.hostRouteId);
   daytonaDuplexChannelBySession.delete(entry.workerSessionId);
 }
 
+function closeDaytonaDuplexChannel(entry: DaytonaDuplexChannelEntry): Promise<void> {
+  forgetDaytonaDuplexChannel(entry);
+  const closed = Promise.resolve().then(() => entry.session.close());
+  if (retiredDaytonaDuplexChannelByRoute.size >= MAX_RETIRED_DAYTONA_DUPLEX_CHANNELS) {
+    retiredDaytonaDuplexChannelByRoute.delete(retiredDaytonaDuplexChannelByRoute.keys().next().value!);
+  }
+  retiredDaytonaDuplexChannelByRoute.set(entry.hostRouteId, { workerSessionId: entry.workerSessionId, closed });
+  return closed;
+}
+
 // Close every open duplex channel that belongs to one provider lease and drop its
 // entry. The lease teardown hooks (release, destroy, resume) call this, so a
 // channel never outlives the sandbox that carries it. The close kills the child
@@ -2504,8 +2517,7 @@ async function closeDaytonaDuplexChannelsForLease(providerLeaseId: string): Prom
   const matches = [...daytonaDuplexChannelByRoute.values()].filter(
     (entry) => entry.providerLeaseId === providerLeaseId,
   );
-  for (const entry of matches) forgetDaytonaDuplexChannel(entry);
-  await Promise.allSettled(matches.map(entry => entry.session.close()));
+  await Promise.allSettled(matches.map(closeDaytonaDuplexChannel));
 }
 
 async function stopLease(params: PluginEnvironmentReleaseLeaseParams): Promise<PluginEnvironmentTerminationReceipt> {
@@ -3716,6 +3728,9 @@ const plugin = definePlugin(trackDaytonaHooks({
   // `ctx.duplexChannel`, bound to the returned worker session id. Fail closed when
   // no cached sandbox matches the lease.
   async onDuplexChannelOpen(params) {
+    if (daytonaDuplexChannelByRoute.has(params.hostRouteId) || retiredDaytonaDuplexChannelByRoute.has(params.hostRouteId)) {
+      throw new Error("Daytona duplex channel: host route has already been used.");
+    }
     const sandbox = await sandboxHandleCache.findByProviderLeaseId(params.providerLeaseId);
     if (!sandbox) {
       throw new Error(
@@ -3794,10 +3809,14 @@ const plugin = definePlugin(trackDaytonaHooks({
     const entry = daytonaDuplexChannelByRoute.get(params.hostRouteId);
     if (entry) {
       const boundWorkerSessionId = entry.workerSessionId;
-      forgetDaytonaDuplexChannel(entry);
-      await entry.session.close().catch(() => undefined);
+      await closeDaytonaDuplexChannel(entry).catch(() => undefined);
       return { hostRouteId: params.hostRouteId, workerSessionId: boundWorkerSessionId };
     }
+    const retired = retiredDaytonaDuplexChannelByRoute.get(params.hostRouteId);
+    if (retired) {
+      await retired.closed;
+      return { hostRouteId: params.hostRouteId, workerSessionId: retired.workerSessionId };
+    }
     return { hostRouteId: params.hostRouteId };
   },
 
@@ -3814,6 +3833,7 @@ const plugin = definePlugin(trackDaytonaHooks({
     const openChannels = [...daytonaDuplexChannelByRoute.values()];
     daytonaDuplexChannelByRoute.clear();
     daytonaDuplexChannelBySession.clear();
+    retiredDaytonaDuplexChannelByRoute.clear();
     for (const entry of openChannels) {
       await entry.session.close().catch(() => undefined);
     }