feat(mothership): receive private Sim controls over outbound transport

Author: Sg312

apps/sim/instrumentation-node.ts

@@ -403,4 +403,7 @@ export async function register() {
 
   const { startMemoryTelemetry } = await import('./lib/monitoring/memory-telemetry')
   startMemoryTelemetry()
+
+  const { startSimReceivers } = await import('./lib/mothership/transport/receiver')
+  await startSimReceivers()
 }

apps/sim/lib/core/config/env.ts

@@ -557,6 +557,7 @@ export const env = createEnv({
     E2B_FUNCTION_TEMPLATE_ID:               z.string().refine(isImmutableE2BTemplateRef, { message: `E2B_FUNCTION_TEMPLATE_ID ${IMMUTABLE_E2B_TEMPLATE_REF_ERROR}` }).optional(), // Immutable dedicated E2B build for Function JavaScript/Python/Shell and workspace sandbox layers; no Mothership fallback
     E2B_FUNCTION_TEMPLATE_GENERATION:       z.string().refine(isValidSandboxReleaseGeneration, { message: `E2B_FUNCTION_TEMPLATE_GENERATION ${SANDBOX_RELEASE_GENERATION_ERROR}` }).optional(), // Monotonic release epoch printed by the Function E2B builder
     MOTHERSHIP_E2B_TEMPLATE_ID:             z.string().optional(),                  // Mothership code-tool template; never a Function-base fallback
+    MOTHERSHIP_SIM_TRANSPORT:              z.enum(['direct', 'checkpoint']).optional(), // Server-side Sim delivery; hosted defaults to direct, self-hosted to outbound checkpoint delivery
     MOTHERSHIP_SANDBOX_CLI_ENDPOINT:        z.string().optional(),                  // Sim API base the sandboxed sim CLI calls back to; defaults to NEXT_PUBLIC_APP_URL (set when the public URL is not reachable from the sandbox network)
     MOTHERSHIP_E2B_DOC_TEMPLATE_ID:         z.string().optional(),                  // Dedicated E2B template with python-pptx/docx/openpyxl/reportlab for document generation; when set (and E2B enabled), docs compile via Python instead of the JS isolated-vm path
     E2B_PI_TEMPLATE_ID:                     z.string().optional(),                  // E2B template ID/alias with the Pi CLI + git baked in (Create PR, its Babysit continuation, and Review Code)

apps/sim/lib/mothership/agent-cli/sink.test.ts

@@ -40,6 +40,7 @@ describe('stdout sink publication', () => {
     expect(saved).toMatchObject({ exitCode: 0, stderr: result.stderr })
     expect(saved.stdout).toContain('Command succeeded')
     expect(saved.stdout).toContain('Do not repeat a mutation')
+    expect(saved.sinkError).toContain('Command succeeded')
     expect(saved.stdout).toContain(result.stdout)
     expect(write).not.toHaveBeenCalled()
   })
@@ -48,6 +49,7 @@ describe('stdout sink publication', () => {
     write.mockResolvedValue({ outcome: 'error', detail: 'storage unavailable' })
     const saved = await applySink(sink, 'chat', result)
     expect(saved.exitCode).toBe(0)
+    expect(saved.sinkError).toContain('could not be confirmed')
     expect(saved.stdout).toContain('Command succeeded')
     expect(saved.stdout).toContain(result.stdout)
   })

apps/sim/lib/mothership/agent-cli/sink.ts

@@ -22,10 +22,18 @@ export async function applySink(
   signal?: AbortSignal,
   observeOutput?: (value: string) => Promise<SessionFileObserver>
 ): Promise<AgentCliRawResult> {
-  if (result.exitCode !== 0 || signal?.aborted) return result
+  if (result.exitCode !== 0) return result
+  if (signal?.aborted)
+    return {
+      ...result,
+      sinkError:
+        'Command completed; output publication was cancelled. Inspect the file before using it. Do not repeat a mutation.',
+    }
   if (!sessionKey) {
     return {
       ...result,
+      sinkError:
+        'Command succeeded; outputFile was not written because no chat workbench exists. Do not repeat a mutation.',
       stdout: `${result.stdout}\n[outputFile not written: no chat-scoped machine — output returned inline instead]`,
     }
   }
@@ -49,6 +57,8 @@ export async function applySink(
   }
   return {
     ...result,
+    sinkError:
+      'Command succeeded; output publication could not be confirmed. Inspect the destination before using it. Do not repeat a mutation.',
     stdout: `[Command succeeded; writing its output to ${sink.path} could not be confirmed. The output follows inline. Do not repeat a mutation to recover its output.]\n${result.stdout}`,
   }
 }

apps/sim/lib/mothership/generated/agent-cli.ts

@@ -57,6 +57,7 @@ export const AgentCliRequest = z.object({
 export type AgentCliRequest = z.infer<typeof AgentCliRequest>;
 
 export const AgentCliRawResult = z.object({
+  sinkError: z.string().optional(),
   exitCode: z.number().int(),
   stdout: z.string(),
   stderr: z.string().default(""),

apps/sim/lib/mothership/generated/protocol.ts

@@ -16,6 +16,7 @@
  */
 
 import { z } from "zod";
+import { SimConnection } from "./sim-transport";
 
 export const PROTOCOL_VERSION = 1;
 
@@ -73,6 +74,7 @@ const WorkspaceInventorySchema = z.object({
 });
 
 export const ChatPayloadSchema = z.strictObject({
+  simConnection: SimConnection.optional(),
   message: z.string().min(1),
   ...ResponseReceiptSchema.shape,
   userId: z.string().min(1),
@@ -139,6 +141,7 @@ export interface StreamToolReplay {
 
 /** POST /api/mothership — the chat request sim sends. */
 export interface ChatRequest extends StreamResponseReceipt {
+  simConnection?: SimConnection | undefined;
   message: string;
   userId: string;
   /** Bump-gated (S43): senders include it; the worker 426s on mismatch. */

apps/sim/lib/mothership/generated/sim-transport.ts

@@ -0,0 +1,49 @@
+// GENERATED — do not edit. Source of truth: mothership worker packages/contracts/src/sim-transport.ts
+// Regenerate with `bun run contracts:sync` in the worker.
+
+import { z } from "zod";
+import { RunControlRequest } from "./run-control";
+import { TaskWakeRequest, WorkflowWatchRequest } from "./tasks";
+
+export const SimConnection = z.discriminatedUnion("mode", [
+  z.object({ mode: z.literal("direct") }),
+  z.object({ mode: z.literal("checkpoint"), channelId: z.string().regex(/^[a-f0-9]{64}$/) }),
+]);
+export type SimConnection = z.infer<typeof SimConnection>;
+
+export const SimScope = z.object({
+  userId: z.string().min(1),
+  workspaceId: z.uuid(),
+  chatId: z.uuid(),
+});
+export type SimScope = z.infer<typeof SimScope>;
+
+/** Only idempotent controls use the idle channel; tool effects use the run event log. */
+export const SimControlOperation = z.discriminatedUnion("kind", [
+  z.object({ kind: z.literal("run_control"), input: RunControlRequest }),
+  z.object({ kind: z.literal("workflow_status"), input: WorkflowWatchRequest }),
+  z.object({ kind: z.literal("wake"), input: TaskWakeRequest }),
+]);
+export type SimControlOperation = z.infer<typeof SimControlOperation>;
+
+export const SimControlRequest = z.object({
+  id: z.uuid(),
+  scope: SimScope,
+  operation: SimControlOperation,
+  expiresAt: z.number().int().positive(),
+});
+export type SimControlRequest = z.infer<typeof SimControlRequest>;
+
+/** HTTP and outbound delivery preserve the same status and JSON response bytes. */
+export const SimControlResult = z.object({
+  status: z.number().int().min(200).max(599),
+  body: z.string().max(16_000_000),
+});
+export type SimControlResult = z.infer<typeof SimControlResult>;
+
+export const SimChannelPoll = z.object({ channelId: z.string().regex(/^[a-f0-9]{64}$/) });
+export const SimChannelBatch = z.object({ requests: z.array(SimControlRequest).max(16) });
+export const SimChannelReply = SimChannelPoll.extend({
+  id: z.uuid(),
+  result: SimControlResult,
+});

apps/sim/lib/mothership/request/lifecycle/run.test.ts

@@ -39,6 +39,7 @@ const {
   mockUpdateRunStatus: vi.fn(),
   mockCheckAttributedUsageLimits: vi.fn(),
   mockEnv: {
+    INTERNAL_API_SECRET: 'transport-test-secret-000000000000000000',
     COPILOT_API_KEY: undefined as string | undefined,
     MSHIP_SYSPROMPT_OVERRIDE: undefined as string | undefined,
   },
@@ -547,7 +548,33 @@ describe('runCopilotLifecycle', () => {
     const sent = JSON.parse(capturedRequestBody)
     /** Receipt metadata leaves ordinary caller content unchanged. */
     expect(sent).not.toHaveProperty('byokApiKey')
-    expect(sent).toEqual({ ...payload, receivedTextChars: 0 })
+    expect(sent).toEqual({
+      ...payload,
+      receivedTextChars: 0,
+      simConnection: {
+        mode: 'checkpoint',
+        channelId: expect.stringMatching(/^[a-f0-9]{64}$/),
+      },
+    })
+  })
+
+  it('stamps server-owned transport over caller claims without exposing the instance secret', async () => {
+    let captured = ''
+    mockRunStreamLoop.mockImplementationOnce(async (_url: string, request: RequestInit) => {
+      captured = String(request.body)
+    })
+    await runCopilotLifecycle(
+      { message: 'hello', simConnection: { mode: 'direct' } },
+      {
+        userId: 'user-1',
+        workspaceId: 'ws-1',
+      }
+    )
+    expect(JSON.parse(captured).simConnection).toEqual({
+      mode: 'checkpoint',
+      channelId: expect.stringMatching(/^[a-f0-9]{64}$/),
+    })
+    expect(captured).not.toContain(mockEnv.INTERNAL_API_SECRET)
   })
 
   it('attaches the resolved enterprise BYOK key to the outbound payload', async () => {
@@ -949,7 +976,14 @@ describe('runCopilotLifecycle', () => {
 
       const sent = JSON.parse(capturedRequestBody)
       expect(sent).not.toHaveProperty('byokApiKey')
-      expect(sent).toEqual({ ...payload, receivedTextChars: 0 })
+      expect(sent).toEqual({
+        ...payload,
+        receivedTextChars: 0,
+        simConnection: {
+          mode: 'checkpoint',
+          channelId: expect.stringMatching(/^[a-f0-9]{64}$/),
+        },
+      })
     }
   )
 

apps/sim/lib/mothership/request/lifecycle/run.ts

@@ -68,6 +68,7 @@ import type {
 import type { SecretMountPolicy } from '@/lib/mothership/secret-mount-policy'
 import { getMothershipBaseURL } from '@/lib/mothership/server/agent-url'
 import { prepareExecutionContext } from '@/lib/mothership/tools/handlers/context'
+import { getSimConnection } from '@/lib/mothership/transport/connection'
 import { filterModelSafeWorkspaceFileAttachments } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
 import { refuseResolvedSecretProjection } from '@/executor/utils/resolved-secret-projection-refusal'
 import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
@@ -891,6 +892,10 @@ async function runCheckpointLoop(
   let retry: StreamRetryWindow | undefined
   const callerOnEvent = options.onEvent
   const mothershipBaseURL = await getMothershipBaseURL({ userId: options.userId })
+  if (initialRoute === '/api/mothership' || initialRoute === '/api/copilot') {
+    const simConnection = getSimConnection()
+    payload = { ...payload, simConnection }
+  }
   const lifecycleWorkspaceId = nonBlankString(options.workspaceId)
   const mothershipRequestId = nonBlankString(options.simRequestId) ?? generateId()
   if (!options.simRequestId) {

apps/sim/lib/mothership/tools/hosted-workbench.smoke.test.ts

@@ -102,6 +102,11 @@ describe.skipIf(!enabled)('configured hosted Mship workbench', () => {
     expect(written).toEqual({ outcome: 'written', path: '/home/user/input.bin' })
     expect(await listed(key)).toEqual([machine.sandboxId])
     expect(independent.sandboxId).not.toBe(machine.sandboxId)
+    const privateBoundary = await machine.runCode(
+      "import os\nassert not any(os.environ.get(k) for k in ['SIM_API_KEY', 'SIM_ENDPOINT', 'SIM_WORKSPACE'])\nprint('compute without Sim credentials')",
+      { timeoutMs: 20_000 }
+    )
+    expect(privateBoundary.stdout.trim()).toBe('compute without Sim credentials')
     const results = await Promise.all([
       machine.runCode(
         "from pathlib import Path\nb = Path('input.bin').read_bytes()\nPath('answer.bin').write_bytes(b[::-1])\nprint(sum(b))",