限制托管执行回执并分离超时输出

Author: 1625567290

packages/eval/harbor/relay_agent.py

@@ -393,7 +393,10 @@ async def _prepare_command(
         if secret_path is not None:
             secret_path.unlink(missing_ok=True)
     subject = shlex.join([request["command"], *request["args"]])
-    output_redirect = "" if capture_stdout else " >/dev/null"
+    # An exit-code subject has no structured stdout to preserve. Detach both
+    # streams so a task-owned background process cannot keep Harbor's
+    # `docker compose exec` output pipe open after the subject leader exits.
+    output_redirect = "" if capture_stdout else " >/dev/null 2>&1"
     scope_error = shlex.quote(f"{SCOPE_ERROR_PREFIX} {result_token}\\n")
     inner = (
         "umask 077; "

packages/eval/harbor/test_relay_lifecycle.py

@@ -433,6 +433,23 @@ async def trial(*_args):
 
 
 class RelayLifecycleTest(unittest.IsolatedAsyncioTestCase):
+    async def test_exit_code_subject_detaches_both_output_streams(self):
+        relay = load_relay()
+        command = await relay._prepare_command(
+            ExplodingSubjectEnvironment(),
+            {
+                "command": "/bin/sh",
+                "args": ["-c", "run-subject"],
+                "credentials": {},
+                "resultToken": "0" * 32,
+                "captureStdout": False,
+            },
+            "detach-output",
+            "/tmp/maka-eval-detach-output.pid",
+        )
+
+        self.assertIn(">/dev/null 2>&1", command)
+
     async def test_framework_timeout_uses_its_own_budget_and_completes_the_relay(self):
         relay = load_relay()
         environment = FrameworkBudgetEnvironment()

packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts

@@ -162,6 +162,63 @@ test('fast settlement remains cached and cannot execute twice', async () => {
   assert.equal(runs, 1);
 });
 
+test('one-shot Host rejects every different execution identity and retains one receipt', async () => {
+  let runs = 0;
+  let aborts = 0;
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution, signal) => {
+      runs += 1;
+      signal.addEventListener('abort', () => {
+        aborts += 1;
+      });
+      return settled(execution.executionId, 'completed');
+    },
+    () => {},
+  );
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+
+  const outcomes = await Promise.all(
+    Array.from({ length: 200 }, (_, index) =>
+      coordinator.handlers['hosted.execution.admit'](
+        input(`00000000-0000-4000-8001-${String(index).padStart(12, '0')}`),
+        context(),
+      ),
+    ),
+  );
+
+  assert.equal(
+    outcomes.every((outcome) => !outcome.ok),
+    true,
+  );
+  for (const outcome of outcomes) {
+    if (!outcome.ok) assert.equal(outcome.error.code, 'operation_conflict');
+  }
+  assert.equal(runs, 1);
+  assert.equal(aborts, 0);
+  coordinator.beginDrain();
+  assert.equal(aborts, 1);
+});
+
+test('cancelling first binds the one-shot Host to that execution identity', async () => {
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => settled(execution.executionId, 'completed'),
+    () => {},
+  );
+  const cancelled = await coordinator.handlers['hosted.execution.cancel'](
+    { executionId: ID },
+    context(),
+  );
+  assert.equal(cancelled.ok, true);
+
+  const different = await coordinator.handlers['hosted.execution.admit'](
+    input('00000000-0000-4000-8000-000000000002'),
+    context(),
+  );
+  assert.equal(different.ok, false);
+  if (!different.ok) assert.equal(different.error.code, 'operation_conflict');
+});
+
 test('admission authority cannot cross connection or Host epoch', async () => {
   const coordinator = new HostHostedExecutionCoordinator(
     async (execution) => settled(execution.executionId, 'completed'),
@@ -208,9 +265,9 @@ function settled(executionId: string, status: 'completed' | 'failed') {
   };
 }
 
-function input() {
+function input(executionId = ID) {
   return {
-    executionId: ID,
+    executionId,
     session: {
       workspace: { kind: 'host_path' as const, path: '/workspace' },
       modelTarget: { kind: 'explicit' as const, connectionSlug: 'env-openai', model: 'model' },

packages/runtime-host/src/server/hosted-execution-coordinator.ts

@@ -52,6 +52,7 @@ export class HostHostedExecutionCoordinator {
 
   readonly #executions = new Map<string, HostedExecutionRecord>();
   readonly #cancelled = new Set<string>();
+  #executionId: string | undefined;
   #accepting = true;
 
   constructor(
@@ -76,6 +77,9 @@ export class HostHostedExecutionCoordinator {
     input: HostedExecutionStartInput,
     context: ConnectionContext,
   ): Promise<OperationOutcome<'hosted.execution.admit'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.executionId) {
+      return conflict();
+    }
     if (this.#cancelled.has(input.executionId)) {
       return {
         ok: false,
@@ -105,6 +109,7 @@ export class HostHostedExecutionCoordinator {
       };
     }
 
+    this.#executionId = input.executionId;
     const execution = this.#createExecution(input, context);
     return {
       ok: true,
@@ -116,6 +121,9 @@ export class HostHostedExecutionCoordinator {
     input: HostedExecutionAdmittedStartInput,
     context: ConnectionContext,
   ): Promise<OperationOutcome<'hosted.execution.start'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.execution.executionId) {
+      return conflict();
+    }
     if (this.#cancelled.has(input.execution.executionId)) {
       this.requestDrain();
       return {
@@ -172,6 +180,10 @@ export class HostHostedExecutionCoordinator {
   async #cancel(
     input: HostedExecutionReferenceInput,
   ): Promise<OperationOutcome<'hosted.execution.cancel'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.executionId) {
+      return conflict();
+    }
+    this.#executionId = input.executionId;
     this.#cancelled.add(input.executionId);
     const execution = this.#executions.get(input.executionId);
     if (!execution) {
@@ -198,7 +210,7 @@ function sameAuthority(
 }
 
 function conflict<
-  K extends 'hosted.execution.admit' | 'hosted.execution.start',
+  K extends 'hosted.execution.admit' | 'hosted.execution.start' | 'hosted.execution.cancel',
 >(): OperationOutcome<K> {
   return {
     ok: false,