Create owned conversations for every benchmark execution

Author: icecrasher321

apps/sim/lib/benchmarks/README.md

@@ -21,7 +21,7 @@ Open **Run history**, choose **View** on a result, then **Compare** on another t
 
 Grading opens the saved report for human review. On any detail, choose **Mark as correct** or **Mark as incorrect**, add an optional note, and save. Human decisions update the report score and comparisons, while the AI grade, explanation, answers, and spec remain intact. **Edit review** changes the note; **Use AI grade** removes the override. The history list and report show the original AI score when human overrides exist. Reviews retain the same private ownership and current source-access checks as the benchmark; concurrent edits cannot silently overwrite each other.
 
-Reference specs can be prose or JSON text; **Import spec** accepts Markdown, text, and JSON files without reformatting their contents. **Generate blanks with AI** creates the redacted reference and expected answers together. **Edit JSON** also lets you paste or edit a mapping such as `{"queue":"Customer Escalations","handoff":"Engineering explicitly accepts the case"}` alongside the matching redacted reference. Keys are blank IDs and values are exact original passages as strings. Applying validates that the mapping restores the reference exactly and updates the draft; **Save changes** persists it.
+Generated reference specs are readable Markdown, with code blocks where exact schemas or mappings help. Imported reference specs can be prose or JSON text; **Import spec** accepts Markdown, text, and JSON files without reformatting their contents. **Generate blanks with AI** creates the redacted reference and expected answers together. **Edit JSON** also lets you paste or edit a mapping such as `{"queue":"Customer Escalations","handoff":"Engineering explicitly accepts the case"}` alongside the matching redacted reference. Keys are blank IDs and values are exact original passages as strings. Applying validates that the mapping restores the reference exactly and updates the draft; **Save changes** persists it.
 
 Benchmark stages have no benchmark-specific duration cutoff. The active process renews its persistence lease every 30 seconds; a two-minute lease detects an abandoned process rather than limiting a healthy run. Concurrent starts or stale completions cannot overwrite newer results. Interrupted steps are retryable; a hard server failure becomes retryable when its lease expires. This version runs a stage within its HTTP request, so the deployment must support long-lived requests. Page reload may interrupt a pending request; saved completed stages remain available.
 

apps/sim/lib/benchmarks/application/operations.ts

@@ -3,8 +3,8 @@ import { defineWorkspaceOperation } from '@/lib/core/application/workspace-opera
 
 export const benchmarkOperations = {
   /** permission-group-exempt: platform superusers administer benchmarks; source access is checked for the selected user. */
-  prepareReference: defineOperation({
-    id: 'benchmarks.reference.prepare',
+  prepareExecution: defineOperation({
+    id: 'benchmarks.execution.prepare',
     principalKinds: ['session'],
     capability: 'none',
   }),

…chmarks/application/prepare-reference.ts …chmarks/application/prepare-execution.tsapps/sim/lib/benchmarks/application/prepare-reference.ts renamed to apps/sim/lib/benchmarks/application/prepare-execution.ts apps/sim/lib/benchmarks/application/prepare-reference.ts renamed to apps/sim/lib/benchmarks/application/prepare-execution.ts

@@ -6,9 +6,9 @@ import { requireBenchmarkCaseAccess } from '@/lib/benchmarks/application/cases'
 import { benchmarkOperations } from '@/lib/benchmarks/application/operations'
 import { MOTHERSHIP_CHAT_DEFAULT_MODEL } from '@/lib/mothership/constants'
 
-/** Source reads use an owned workspace chat, separate from the planner's enterprise discovery context. */
-export const prepareBenchmarkReference = defineAuthorizedBenchmarkUseCase({
-  operation: benchmarkOperations.prepareReference,
+/** Each JSON execution uses a fresh owned workspace chat, separate from the planner's enterprise discovery context. */
+export const prepareBenchmarkExecution = defineAuthorizedBenchmarkUseCase({
+  operation: benchmarkOperations.prepareExecution,
   async execute({
     principal,
     input,
@@ -32,7 +32,7 @@ export const prepareBenchmarkReference = defineAuthorizedBenchmarkUseCase({
         lastSeenAt: new Date(),
       })
       .returning({ id: copilotChats.id })
-    if (!chat) throw new Error('Failed to create benchmark reference conversation')
+    if (!chat) throw new Error('Failed to create benchmark execution conversation')
     return { chatId: chat.id, userId }
   },
 })

apps/sim/lib/benchmarks/application/run-stage.ts

@@ -5,7 +5,6 @@ import { z } from 'zod'
 import { defineAuthorizedBenchmarkUseCase } from '@/lib/benchmarks/application/access'
 import { requireBenchmarkCaseAccess } from '@/lib/benchmarks/application/cases'
 import { benchmarkOperations } from '@/lib/benchmarks/application/operations'
-import { prepareBenchmarkReference } from '@/lib/benchmarks/application/prepare-reference'
 import {
   BENCHMARK_LEASE_MS,
   withBenchmarkStageLease,
@@ -44,7 +43,14 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
 const logger = createLogger('BenchmarkStage')
 
 const distillationSchema = z
-  .object({ taskBrief: benchmarkBriefSchema.min(1), referenceSpec: benchmarkSpecSchema.min(1) })
+  .object({
+    taskBrief: benchmarkBriefSchema.min(1),
+    referenceSpec: benchmarkSpecSchema
+      .min(1)
+      .describe(
+        'A complete, human-readable Markdown specification, with headings and prose. Use code blocks for exact mappings or code where needed.'
+      ),
+  })
   .strict()
 const redactionSchema = z
   .object({
@@ -91,23 +97,19 @@ async function performStage(
   const artifacts = benchmark.artifacts
   switch (stage) {
     case 'distill': {
-      const target = await prepareBenchmarkReference.execute({
-        principal,
-        input: { organizationId: benchmark.organizationId, benchmarkId: benchmark.id },
-      })
-      signal.throwIfAborted()
       const result = await executeBenchmarkJson({
+        principal,
         benchmark,
         signal,
         schema: distillationSchema,
         messages: distillationMessages(artifacts.taskBrief),
         profile: { stage: 'distill' },
-        chatId: target.chatId,
       })
       return { artifacts: applyBenchmarkPatch(artifacts, result), plannerChatId: null }
     }
     case 'redact': {
       const result = await executeBenchmarkJson({
+        principal,
         benchmark,
         signal,
         schema: redactionSchema,
@@ -134,6 +136,7 @@ async function performStage(
     }
     case 'reconstruct': {
       const result = await executeBenchmarkJson({
+        principal,
         benchmark,
         signal,
         schema: reconstructionSchema,
@@ -145,6 +148,7 @@ async function performStage(
     }
     case 'grade': {
       const result = await executeBenchmarkJson({
+        principal,
         benchmark,
         signal,
         schema: gradingSchema,

apps/sim/lib/benchmarks/prompts.ts

@@ -13,7 +13,7 @@ function messages(instruction: string, data: unknown): ExecuteMessage[] {
 
 export function distillationMessages(taskBrief: string): ExecuteMessage[] {
   return messages(
-    'Explore the selected workspace using the read-only workspace CLI and describe its implemented behavior as a detailed, self-contained reference specification. Begin by listing all active workflows, follow list pagination, inspect each workflow and its referenced workspace resources, and follow large-output continuations. Cover exact triggers and input shapes, conditions, ownership, field mappings, prompts and code behavior, actions, destinations, cross-workflow relationships, outputs and failure/recovery behavior. Preserve concrete names, values and business rules; do not compress them into a high-level overview. Distinguish implemented behavior from unresolved configuration or inferred intent. Do not invent missing values. Omit editor layout and credential values. Also draft a short taskBrief expressing the business goal a user would originally request, without disclosing the enterprise-specific answers. If a taskBrief was supplied, preserve it. The reference is for human review before evaluation. Use tools to inspect; return the final JSON only when inspection is complete.',
+    'Explore the selected workspace using the read-only workspace CLI and describe its implemented behavior as a detailed, self-contained reference specification. Begin by listing all active workflows, follow list pagination, inspect each workflow and its referenced workspace resources, and follow large-output continuations. Cover exact triggers and input shapes, conditions, ownership, field mappings, prompts and code behavior, actions, destinations, cross-workflow relationships, outputs and failure/recovery behavior. Preserve concrete names, values and business rules; do not compress them into a high-level overview. Distinguish implemented behavior from unresolved configuration or inferred intent. Do not invent missing values. Omit editor layout and credential values. Also draft a short taskBrief expressing the business goal a user would originally request, without disclosing the enterprise-specific answers. If a taskBrief was supplied, preserve it. Write referenceSpec as a human-readable Markdown document with descriptive headings, paragraphs and lists. Use fenced JSON or code blocks only for exact schemas, mappings or code that help explain the behavior. Do not serialize the entire specification as a JSON document inside referenceSpec. The outer JSON object is only the response envelope containing taskBrief and the Markdown referenceSpec string. The reference is for human review before evaluation. Use tools to inspect; return the final JSON only when inspection is complete.',
     taskBrief.trim() ? { taskBrief } : {}
   )
 }

apps/sim/lib/benchmarks/repository.integration.ts

@@ -8,21 +8,65 @@ import { drizzle, type PostgresJsDatabase } from 'drizzle-orm/postgres-js'
 import postgres from 'postgres'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
-const database = vi.hoisted(() => {
+const database = await vi.hoisted(async () => {
   process.env.MOTHERSHIP_BENCHMARK_ENABLED = 'true'
-  return { current: undefined as PostgresJsDatabase | undefined }
+  const { createServer } = await import('node:http')
+  const requests: Record<string, unknown>[] = []
+  const worker = createServer(async (request, response) => {
+    let body = ''
+    for await (const chunk of request) body += chunk
+    if (request.url !== '/api/mothership/execute') {
+      response.writeHead(200, { 'content-type': 'application/json' }).end('{}')
+      return
+    }
+    const payload = JSON.parse(body) as Record<string, unknown>
+    requests.push(payload)
+    const frames = [
+      { type: 'text', payload: { channel: 'assistant', text: '{"ok":true}' } },
+      { type: 'complete', payload: { status: 'complete' } },
+    ]
+    response.writeHead(200, { 'content-type': 'text/event-stream' })
+    response.end(
+      frames
+        .map(
+          (frame, index) =>
+            `data: ${JSON.stringify({
+              v: 1,
+              seq: index + 1,
+              ts: new Date().toISOString(),
+              stream: { streamId: payload.messageId },
+              ...frame,
+            })}\n\n`
+        )
+        .join('')
+    )
+  })
+  await new Promise<void>((resolve) => worker.listen(0, '127.0.0.1', resolve))
+  const address = worker.address()
+  if (!address || typeof address === 'string') throw new Error('Worker fixture did not bind')
+  process.env.MOTHERSHIP_BENCHMARK_URL = `http://127.0.0.1:${address.port}`
+  process.env.COPILOT_API_KEY = 'local-benchmark-fixture'
+  return { current: undefined as PostgresJsDatabase | undefined, worker, requests }
 })
 vi.mock('server-only', () => ({}))
-vi.mock('@sim/db', () => ({
-  get db() {
-    if (!database.current) throw new Error('Benchmark test database is not initialized')
-    return database.current
-  },
-}))
+vi.mock('@sim/db', () => {
+  const scopedDatabase = new Proxy(
+    {},
+    {
+      get(_target, property) {
+        if (!database.current) throw new Error('Benchmark test database is not initialized')
+        const value = Reflect.get(database.current, property)
+        return typeof value === 'function' ? value.bind(database.current) : value
+      },
+    }
+  )
+  return { db: scopedDatabase, dbFor: () => scopedDatabase }
+})
 
+import { z } from 'zod'
 import { createBenchmark, getBenchmark, listBenchmarks } from '@/lib/benchmarks/application/cases'
+import { prepareBenchmarkExecution } from '@/lib/benchmarks/application/prepare-execution'
 import { prepareBenchmarkPlan } from '@/lib/benchmarks/application/prepare-plan'
-import { prepareBenchmarkReference } from '@/lib/benchmarks/application/prepare-reference'
 import {
   getBenchmarkRun,
   listBenchmarkRuns,
@@ -46,6 +90,7 @@ import {
   updateBenchmarkRecord,
 } from '@/lib/benchmarks/repository'
 import { type BenchmarkArtifacts, emptyBenchmarkArtifacts } from '@/lib/benchmarks/types'
+import { executeBenchmarkJson } from '@/lib/benchmarks/worker'
 import { listOrganizationChats } from '@/lib/mothership/chat/organization-chats'
 
 describe('private benchmark persistence and attempt fencing', () => {
@@ -113,6 +158,9 @@ describe('private benchmark persistence and attempt fencing', () => {
       'user',
       'settings',
       'copilot_chats',
+      'copilot_runs',
+      'copilot_request_stops',
+      'copilot_organization_request_stops',
       'mothership_memory_selections',
       'mothership_memory_spaces',
       'organization',
@@ -128,6 +176,7 @@ describe('private benchmark persistence and attempt fencing', () => {
     ]) {
       await connection`CREATE TABLE ${connection(table)} (LIKE ${connection(`public.${table}`)} INCLUDING ALL)`
     }
+    await connection`ALTER TABLE copilot_runs ADD CONSTRAINT benchmark_chat_fk FOREIGN KEY (chat_id) REFERENCES copilot_chats(id)`
     database.current = drizzle(connection)
     await connection`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at) VALUES ('owner', 'Owner', 'owner@benchmark.test', true, now(), now()), ('peer', 'Peer', 'peer@benchmark.test', true, now(), now())`
     await connection`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at) VALUES ('target', 'Target', 'target@benchmark.test', true, now(), now())`
@@ -138,7 +187,7 @@ describe('private benchmark persistence and attempt fencing', () => {
   })
 
   beforeEach(async () => {
-    await connection`TRUNCATE mothership_benchmark_runs, mothership_benchmarks, permissions, copilot_chats`
+    await connection`TRUNCATE mothership_benchmark_runs, mothership_benchmarks, permissions, copilot_runs, copilot_chats`
     await connection`UPDATE "user" SET role = CASE WHEN id IN ('owner', 'peer') THEN 'admin' ELSE 'user' END, banned = false`
     await connection`UPDATE settings SET super_user_mode_enabled = true`
     await connection`INSERT INTO member (id, organization_id, user_id, role) VALUES ('owner-member', 'org', 'owner', 'member'), ('target-member', 'org', 'target', 'member') ON CONFLICT (id) DO UPDATE SET organization_id = 'org'`
@@ -157,7 +206,52 @@ describe('private benchmark persistence and attempt fencing', () => {
     } finally {
       database.current = undefined
       await connection.end()
+      await new Promise<void>((resolve, reject) => {
+        database.worker.close((error) => (error ? reject(error) : resolve()))
+        database.worker.closeAllConnections()
+      })
+    }
+  })
+
+  it('completes isolated JSON executions through the real lifecycle with fresh target-owned chats', async () => {
+    const { benchmark } = await createBenchmark.execute({
+      principal,
+      input: {
+        organizationId: 'org',
+        sourceWorkspaceId: 'workspace',
+        runAsUserId: 'target',
+        name: 'JSON execution',
+      },
+    })
+    database.requests.length = 0
+    for (let attempt = 0; attempt < 2; attempt++) {
+      const input = {
+        principal,
+        benchmark,
+        messages: [{ role: 'user' as const, content: 'Return the fixture result.' }],
+        schema: z.object({ ok: z.literal(true) }),
+        signal: new AbortController().signal,
+      }
+      expect(await executeBenchmarkJson(input)).toEqual({ ok: true })
     }
+    const runs =
+      await connection`SELECT r.status, r.user_id, c.user_id AS chat_user_id, c.config FROM copilot_runs r JOIN copilot_chats c ON c.id = r.chat_id`
+    expect(runs).toHaveLength(2)
+    for (const run of runs)
+      expect(run).toMatchObject({
+        status: 'complete',
+        user_id: 'target',
+        chat_user_id: 'target',
+        config: { benchmark: { id: benchmark.id, operatorUserId: 'owner' } },
+      })
+    expect(database.requests).toHaveLength(2)
+    expect(new Set(database.requests.map((request) => request.chatId)).size).toBe(2)
+    for (const request of database.requests)
+      expect(request).toMatchObject({
+        userId: 'target',
+        useConversationHistory: false,
+        messages: [{ role: 'user', content: 'Return the fixture result.' }],
+      })
   })
 
   it('requires a current superuser role and enabled toggle even when the deployment flag is on', async () => {
@@ -233,7 +327,7 @@ describe('private benchmark persistence and attempt fencing', () => {
       },
     })
     expect(benchmark).toMatchObject({ userId: 'owner', runAsUserId: 'target' })
-    const reference = await prepareBenchmarkReference.execute({
+    const reference = await prepareBenchmarkExecution.execute({
       principal,
       input: { organizationId: 'org', benchmarkId: benchmark.id },
     })
@@ -287,7 +381,7 @@ describe('private benchmark persistence and attempt fencing', () => {
       })
     ).rejects.toMatchObject({ code: 'forbidden' })
     await expect(
-      prepareBenchmarkReference.execute({
+      prepareBenchmarkExecution.execute({
         principal,
         input: { organizationId: 'org', benchmarkId: benchmark.id },
       })

apps/sim/lib/benchmarks/worker.ts

@@ -2,6 +2,7 @@ import type { Principal } from '@sim/auth/principal'
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
 import { z } from 'zod'
+import { prepareBenchmarkExecution } from '@/lib/benchmarks/application/prepare-execution'
 import { prepareBenchmarkPlan } from '@/lib/benchmarks/application/prepare-plan'
 import { getBenchmarkMothershipUrl } from '@/lib/benchmarks/config'
 import type { BenchmarkCase } from '@/lib/benchmarks/types'
@@ -58,20 +59,26 @@ async function stopIncompleteRun(
 
 /** Each invocation uses a fresh conversation; optional profiles expose only the stage's permitted reads. */
 export async function executeBenchmarkJson<S extends z.ZodType>(input: {
+  principal: Principal
   benchmark: BenchmarkCase
   messages: ExecuteMessage[]
   schema: S
   signal: AbortSignal
   profile?: BenchmarkExecution
-  chatId?: string
 }): Promise<z.output<S>> {
   getBenchmarkMothershipUrl()
-  const executionUserId = input.benchmark.runAsUserId ?? input.benchmark.userId
+  input.signal.throwIfAborted()
+  const target = await prepareBenchmarkExecution.execute({
+    principal: input.principal,
+    input: { organizationId: input.benchmark.organizationId, benchmarkId: input.benchmark.id },
+  })
+  input.signal.throwIfAborted()
+  const executionUserId = target.userId
   const messageId = generateId()
   const payload: ExecuteRequest = {
     protocolVersion: PROTOCOL_VERSION,
     messageId,
-    chatId: input.chatId ?? generateId(),
+    chatId: target.chatId,
     benchmark: input.profile,
     userId: executionUserId,
     workspaceId: input.benchmark.sourceWorkspaceId,