fix(environments): honor stopped-task proof after warm preparation (#20289)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - A warm plugin Runner retains its task for later runs.
> - A follow-up can find that the retained task has stopped.
> - The resume path already asks settle to check status before it sends
stop.
> - A saved startup marker disabled this order, even after warm
preparation had finished.
> - This pull request applies the same stop-proof check to all
observed-stop releases.
> - The next run skips a repeated stop when status proves both execution
and required preparation have stopped.

## Linked Issues or Issue Description

Fixes: #20224
Refs #20203

## What Changed

- Remove the startup-marker exclusion from the existing status-first
path. Keep the existing shared confirmation check. It still requires
`preparationStopped: true` for startup work.
- Keep stop followed by status when the first status lacks the required
proof. Keep the startup drain, credential writeback, complete, and
cleanup failure handling.
- Add retained-task coverage with the startup metadata that warm
preparation saves. Cover a startup cleanup whose first status lacks
preparation proof.
- Update the plugin task lifecycle contract.

## Verification

- Local macOS, source base `3e8744fa91`: from `server`, `pnpm exec
vitest run src/__tests__/environment-task-runtime.test.ts
src/__tests__/plugin-warm-tasks.integration.test.ts
src/__tests__/plugin-task-credential-writeback.test.ts` passed: 3 files,
82 tests. The integration tests use real embedded PostgreSQL and
controlled plugin results.
- After restoring the fix, the two affected files passed again: 73
tests.
- Independent parent verification on commit `19f720ff68`: the two files
filtered to the changed startup-proof and retained-task cases passed (4
tests). Independent source review found no actionable issue.
- Baseline proof: with only the source condition restored to upstream,
the new retained-task test with saved startup metadata fails because it
sees an extra `stop`. The case without that metadata passes. The source
fix was then restored.
- The retained-task regression asserts exactly `environmentResumeLease`,
`status`, `complete`, and `environmentAcquireLease`, with both task
operations bound to the old task. The startup test also proves that
missing preparation proof sends `stop` and a second `status`.
- Test setup: `node scripts/agent-preflight.mjs --skip-rust-debug
--explain` installed dependencies, then stopped because Cargo is
unavailable on this Mac. Initial test collection failed because the
plugin SDK output was absent. `pnpm --filter @paperclipai/plugin-sdk
build` built that required dependency and the shared package; the
targeted test run then passed.
- Standard Greptile reviewed commit
`19f720ff686058b4dd6b34f27cdfb9c8ff60e50a` at 5/5 with no review
threads.
[Review](https://github.com/paperclipai/paperclip/pull/20289#issuecomment-6743829614).
- Full local checks were not run, as requested. [GitHub PR workflow
42644023868](https://github.com/paperclipai/paperclip/actions/runs/42644023868)
passed on final commit `19f720ff686058b4dd6b34f27cdfb9c8ff60e50a`. It
supplied broad typecheck, build, and test coverage. The separate Snyk
status passed. Optional Storybook checks were skipped.
- The change removes one provider stop call on the reported path. The
issue's measured eight calls would become seven if other calls stay the
same. The earlier five-call target is not claimed. No deployed latency
measurement is claimed.

## Risks

- A provider that cannot prove preparation stopped still gets stop and
status. That fallback can cost one extra status call, as the existing
status-first contract already allows.
- A status failure keeps cleanup pending and blocks replacement. No
termination proof or credential handoff is skipped.
- No schema, API, dependency, or timer change. No added work on
successful task reuse or ordinary task dispatch.
- Live provider acceptance remains unrun. The local evidence checks
orchestration and stop-proof handling.

## Model Used

- OpenAI Codex, GPT-6 family, with reasoning, repository inspection,
shell tools, and code execution. The exact serving model ID and context
window are not exposed by 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>

doc/plugins/PLUGIN_RUNNER_EXECUTION.md

@@ -350,7 +350,9 @@ call `onEnvironmentReleaseLease`. Core still calls it when the release did not
 complete a dispatched task, and the cleanup retry calls it after any failure.
 When `onEnvironmentResumeLease` returns no provider lease, Core sends `status`
 first. It sends `stop` only when that `status` does not report
-`executionStopped: true`. A driver must release task-scoped resources in its
+`executionStopped: true` and any required `preparationStopped: true` proof.
+An old warm-start preparation marker does not force a repeated stop when both
+proofs are present. A driver must release task-scoped resources in its
 `status` or `complete` handling, not only in `onEnvironmentReleaseLease`.
 
 Termination receipts cover all run leases that owned the retired physical task.

server/src/__tests__/environment-task-runtime.test.ts

@@ -88,7 +88,7 @@ describe("plugin-provided Runner execution", () => {
     if (preparationStopped) { expect(result).toEqual({ value: "completed", error: null }); expect(operations).toEqual(["stop", "status", "complete"]); }
     else { expect(result.error?.message).toContain("termination is unconfirmed"); expect(fixture.sets).toContainEqual(expect.objectContaining({ status: "pending_cleanup", cleanupStatus: "failed" })); }
   });
-  it("still stops a startup task first when resume already observed the stop", async () => {
+  it.each([true, false])("checks startup preparation proof after resume observed the stop (%s)", async preparationStopped => {
     const value = row();
     Object.assign(value.lease.metadata, { runnerTaskDispatched: false,
       runnerTaskStartupCleanup: { contextId: "startup", requiresPreparationStop: true } });
@@ -96,13 +96,15 @@ describe("plugin-provided Runner execution", () => {
     fixture.db.select = ((...args: unknown[]) => (database(value).db.select as (...values: unknown[]) => unknown)(...args)) as Db["select"];
     Object.assign(workers, { drainStartup: vi.fn(async () => undefined) });
     const operations: string[] = [];
+    let stopped = false;
     workers.call.mockImplementation(async (...args: any[]) => {
       const operation = args[2].operation.kind; operations.push(operation);
-      return operation === "status" ? { kind: "status", taskId: "attempt-1", phase: "cancelled", executionStopped: true, preparationStopped: true }
+      if (operation === "stop") stopped = true;
+      return operation === "status" ? { kind: "status", taskId: "attempt-1", phase: "cancelled", executionStopped: true, preparationStopped: preparationStopped || stopped }
         : { kind: "accepted", taskId: "attempt-1" };
     });
     await expect(settleDispatchedEnvironmentTask(fixture.db, workers as never, { companyId: "company", id: leaseId }, { executionStopObserved: true })).resolves.toBe("completed");
-    expect(operations).toEqual(["stop", "status", "complete"]);
+    expect(operations).toEqual(preparationStopped ? ["status", "complete"] : ["status", "stop", "status", "complete"]);
   });
   it("negotiates the transient context without reading transport for an older plugin", async () => {
     const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000);

server/src/__tests__/plugin-warm-tasks.integration.test.ts

@@ -48,7 +48,7 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
     if(method==="environmentResumeLease") {if(outage) throw Error("unreachable");
      if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped||unavailable?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};}
     if(method==="environmentReleaseLease") {if(releaseFails) throw Error("release unavailable");return {providerLeaseId:p.providerLeaseId,state:"stopped"};}
-    if(p.operation.kind==="status") return {kind:"status",taskId:p.taskId,phase:stopped?"completed":"running",executionStopped:stopped,sessionStateAvailable:stopped};
+    if(p.operation.kind==="status") return {kind:"status",taskId:p.taskId,phase:stopped?"completed":"running",executionStopped:stopped,sessionStateAvailable:stopped,preparationStopped:stopped};
     if(p.operation.kind==="read_state") return {kind:"state",taskId:p.taskId,state:{...p.operation.identity,lifecycle:"ready"}};
     if(p.operation.kind==="stop") stopped=true;
     if(p.operation.kind==="prepare") return {kind:"prepared",taskId:p.taskId,launch:{configurationId:"configuration",requirementsDigest:`sha256:${"a".repeat(64)}`,provider:{}}};
@@ -113,8 +113,9 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
   expect(f.calls).toContain("complete");
   await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId);
  });
- it("retires a stopped retained task with one status and complete before allocating its replacement",async()=>{
+ it.each([false,true])("retires a stopped retained task with one status and complete after warm preparation (%s)",async prepared=>{
   const f=await fixture();f.stopped();await f.nextRun();
+  if(prepared) await db.update(environmentLeases).set({metadata:sql`${environmentLeases.metadata} || ${JSON.stringify({runnerTaskStartup:{contextId:"settled-startup",deadlineAtMs:1,runId:f.identity.runId}})}::jsonb`}).where(eq(environmentLeases.id,f.first.lease.id));
   const calls=f.calls.length,dispatched=f.dispatched.length;
   const next=await f.acquire();
   expect(f.calls.slice(calls)).toEqual(["environmentResumeLease","status","complete","environmentAcquireLease"]);

server/src/services/environment-task-runtime.ts

@@ -513,7 +513,7 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor
     void draining?.catch(() => undefined);
     const confirmed = (state?: Awaited<ReturnType<typeof executeEnvironmentTask>>) => state?.kind === "status" && state.executionStopped === true &&
       (!(startup || startupCleanup?.requiresPreparationStop) || state.preparationStopped === true);
-    let state = options.executionStopObserved === true && !startup && !startupCleanup
+    let state = options.executionStopObserved === true
       ? await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } }) : undefined;
     if (!confirmed(state)) {
       await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "stop" } });