fix(orchestration): negotiate honest worker stop receipts

src/main/runtime/rpc/methods/orchestration-worker-stop-capability.test.ts

@@ -0,0 +1,57 @@
+import { describe, expect, it, vi } from 'vitest'
+import { ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
+import type { OrcaRuntimeService } from '../../orca-runtime'
+import type { OrchestrationDb } from '../../orchestration/db'
+import { ORCHESTRATION_WORKER_STOP_METHODS } from './orchestration-worker-stop'
+
+describe('federated worker stop capability', () => {
+  it('does not trust a legacy server stop receipt', async () => {
+    const markWorkerStopUnknown = vi.fn((_dispatchId: string, reason: string) => ({
+      state: 'stop_unknown',
+      last_error: reason
+    }))
+    const db = {
+      getFederatedDispatch: vi.fn(() => ({
+        dispatch_id: 'ctx_remote',
+        environment_id: 'environment_linux',
+        environment_name: 'linux',
+        peer_fingerprint: 'peer-linux'
+      })),
+      beginWorkerStop: vi.fn(() => ({ disposition: 'started' })),
+      markWorkerStopUnknown
+    } as unknown as OrchestrationDb
+    const callOrchestrationWorkerServer = vi.fn(async (_environmentId, method) => {
+      if (method === 'status.get') {
+        return { capabilities: [ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY] }
+      }
+      return { state: 'stopped', processAction: 'closed_agent_terminal' }
+    })
+    const runtime = {
+      getOrchestrationDb: () => db,
+      resolveOrchestrationWorkerServer: () => ({
+        environmentId: 'environment_linux',
+        name: 'linux',
+        peerFingerprint: 'peer-linux'
+      }),
+      callOrchestrationWorkerServer
+    } as unknown as OrcaRuntimeService
+    const method = ORCHESTRATION_WORKER_STOP_METHODS[0]!
+
+    await expect(
+      method.handler(method.params!.parse({ dispatch: 'ctx_remote' }), {
+        runtime,
+        orchestrationMutation: {
+          callerFingerprint: 'coordinator',
+          requestId: 'request_stop',
+          method: 'orchestration.workerStop',
+          payloadHash: 'hash'
+        }
+      })
+    ).resolves.toMatchObject({ state: 'stop_unknown', processAction: 'none' })
+    expect(markWorkerStopUnknown).toHaveBeenCalledWith(
+      'ctx_remote',
+      'Connected server linux cannot prove the worker stop outcome.'
+    )
+    expect(callOrchestrationWorkerServer).toHaveBeenCalledTimes(1)
+  })
+})

src/main/runtime/rpc/methods/orchestration-worker-stop.ts

@@ -3,6 +3,8 @@ import { OrchestrationError } from '../../orchestration/orchestration-error'
 import { defineMethod, type RpcMethod } from '../core'
 import { requiredString } from '../schemas'
 import { describeUnconfirmedAgentStop } from '../../../../shared/pty-liveness-verdict'
+import { ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
+import type { RuntimeStatus } from '../../../../shared/runtime-types'
 import {
   inspectWorkerTerminal,
   resolvePinnedFederatedServer
@@ -30,6 +32,24 @@ export const ORCHESTRATION_WORKER_STOP_METHODS: RpcMethod[] = [
           return settledReceipt(params.dispatch, begun.worker.state)
         }
         try {
+          const status = (await runtime.callOrchestrationWorkerServer(
+            server.environmentId,
+            'status.get',
+            undefined,
+            30_000
+          )) as RuntimeStatus
+          if (
+            !status.capabilities?.includes(ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY)
+          ) {
+            return unknownReceipt(
+              params.dispatch,
+              db.markWorkerStopUnknown(
+                params.dispatch,
+                `Connected server ${server.name} cannot prove the worker stop outcome.`
+              ),
+              'none'
+            )
+          }
           const remote = (await runtime.callOrchestrationWorkerServer(
             server.environmentId,
             'orchestration.federationStop',

src/shared/protocol-version.ts

@@ -45,6 +45,8 @@ export const ORCHESTRATION_FEDERATION_CONTROL_MAIL_RUNTIME_CAPABILITY =
   'orchestration.federation-control-mail.v1' as const
 export const ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY =
   'orchestration.federation-lifecycle-settlement.v1' as const
+export const ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY =
+  'orchestration.worker-stop-verdict.v1' as const
 export const ORCHESTRATION_WORKER_LAUNCH_PREFERENCES_RUNTIME_CAPABILITY =
   'orchestration.worker-launch-preferences.v1' as const
 export const ORCHESTRATION_FEDERATION_CONTROL_MAIL_PROTOCOL_VERSION = 2 as const
@@ -115,6 +117,7 @@ export const RUNTIME_CAPABILITIES = [
   ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY,
   ORCHESTRATION_FEDERATION_CONTROL_MAIL_RUNTIME_CAPABILITY,
   ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY,
+  ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY,
   ORCHESTRATION_WORKER_LAUNCH_PREFERENCES_RUNTIME_CAPABILITY,
   ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY,
   BROWSER_SCREENCAST_RUNTIME_CAPABILITY,