improvement(mothership): keep the durable-run predicates out of the chat protocol graph

Author: waleedlatif1

apps/sim/lib/mothership/async-runs/orphaned-runs.ts

@@ -30,7 +30,7 @@ import { getRedisClient } from '@/lib/core/config/redis'
 import type { DbTransaction } from '@/lib/db/types'
 import { SIM_TOOL_EXECUTION_VERSION } from '@/lib/mothership/async-runs/lifecycle'
 import { publishChatStatusChanged } from '@/lib/mothership/chat-status'
-import { isDurableStreamRun } from '@/lib/mothership/request/lifecycle/recovery-config'
+import { isDurableStreamRun } from '@/lib/mothership/request/lifecycle/durable-run'
 import {
   acquirePendingChatStream,
   getChatStreamLockOwners,

apps/sim/lib/mothership/async-runs/repository.ts

@@ -63,10 +63,10 @@ import {
   traceMothershipTransaction,
 } from '@/lib/mothership/observability/database'
 import {
-  type BillingAdmission,
   isDurableStreamRun,
   parseRecordedSecretAuthority,
-} from '@/lib/mothership/request/lifecycle/recovery-config'
+} from '@/lib/mothership/request/lifecycle/durable-run'
+import type { BillingAdmission } from '@/lib/mothership/request/lifecycle/recovery-config'
 import { markSpanForError } from '@/lib/mothership/request/otel'
 import { sealClientToolSettlement } from '@/lib/mothership/request/tools/client-completion-seal.server'
 import {

apps/sim/lib/mothership/chat/stream-liveness.ts

@@ -3,7 +3,7 @@ import { copilotChats, copilotRuns } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { and, eq, inArray, notInArray } from 'drizzle-orm'
-import { isDurableStreamRun } from '@/lib/mothership/request/lifecycle/recovery-config'
+import { isDurableStreamRun } from '@/lib/mothership/request/lifecycle/durable-run'
 import { getChatStreamLockOwners } from '@/lib/mothership/request/session'
 
 const logger = createLogger('ChatStreamLiveness')

apps/sim/lib/mothership/request/application/recover-stream.ts

@@ -30,8 +30,8 @@ import {
   claimRunController,
   planRecovery,
 } from '@/lib/mothership/request/lifecycle/controller-ownership'
+import { parseRecordedSecretAuthority } from '@/lib/mothership/request/lifecycle/durable-run'
 import {
-  parseRecordedSecretAuthority,
   RESOURCE_AUTHORING_VERSION,
   StreamRecoveryConfigSchema,
 } from '@/lib/mothership/request/lifecycle/recovery-config'

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

@@ -0,0 +1,38 @@
+/**
+ * What a run's saved request context says about its durability, kept free of the chat protocol
+ * schema so the run repository and the sweeps can read it without loading the lifecycle.
+ */
+import { type AnyColumn, type SQL, sql } from 'drizzle-orm'
+import { z } from 'zod'
+import { secretMountPolicySchema } from '@/lib/api/contracts/secret-mount-policy'
+import type { TurnSecretAuthority } from '@/lib/mothership/secret-mount-policy'
+
+/**
+ * What a durable turn's recovery restores about its audience. An interactive stream answers a
+ * person in the chat; an unattended stream (a scheduled job's occurrence) has nobody to answer a
+ * prompt, so a recovered leg runs it non-interactively too, exactly as it was admitted.
+ */
+export const DURABLE_STREAM_KINDS = ['interactive_stream', 'unattended_stream'] as const
+export type DurableStreamKind = (typeof DURABLE_STREAM_KINDS)[number]
+
+/** A run's saved request context names a durable turn that recovery can take over. */
+export function isDurableStreamRun(requestContext: AnyColumn): SQL {
+  return sql`${requestContext}->'recovery'->>'kind' IN (${sql.join(
+    DURABLE_STREAM_KINDS.map((kind) => sql`${kind}`),
+    sql`, `
+  )})`
+}
+
+const RecordedSecretAuthoritySchema = z.object({
+  secretMountPolicy: secretMountPolicySchema,
+  secretActorUserId: z.string().nullable(),
+})
+
+/**
+ * The secret authority a run recorded at admission (`requestContext.secretAuthority`), or
+ * undefined when it recorded none or garbage.
+ */
+export function parseRecordedSecretAuthority(value: unknown): TurnSecretAuthority | undefined {
+  const recorded = RecordedSecretAuthoritySchema.safeParse(value)
+  return recorded.success ? recorded.data : undefined
+}

apps/sim/lib/mothership/request/lifecycle/recovery-config.ts

@@ -1,11 +1,9 @@
-import { type AnyColumn, type SQL, sql } from 'drizzle-orm'
 import { z } from 'zod'
-import { secretMountPolicySchema } from '@/lib/api/contracts/secret-mount-policy'
 import { hasCopilotResourceAdmission } from '@/lib/mothership/auth/application-delegation'
 import { COPILOT_REQUEST_MODES } from '@/lib/mothership/constants'
 import { ChatPayloadSchema } from '@/lib/mothership/generated/protocol'
+import { DURABLE_STREAM_KINDS } from '@/lib/mothership/request/lifecycle/durable-run'
 import type { CopilotLifecycleOptions } from '@/lib/mothership/request/lifecycle/run'
-import type { TurnSecretAuthority } from '@/lib/mothership/secret-mount-policy'
 
 export const BillingAdmissionSchema = z
   .object({
@@ -40,22 +38,6 @@ export function durableResourceAuthoringVersion(
   return admitted ? RESOURCE_AUTHORING_VERSION : undefined
 }
 
-/**
- * What a durable turn's recovery restores about its audience. An interactive stream answers a
- * person in the chat; an unattended stream (a scheduled job's occurrence) has nobody to answer a
- * prompt, so a recovered leg runs it non-interactively too, exactly as it was admitted.
- */
-const DURABLE_STREAM_KINDS = ['interactive_stream', 'unattended_stream'] as const
-export type DurableStreamKind = (typeof DURABLE_STREAM_KINDS)[number]
-
-/** A run's saved request context names a durable turn that recovery can take over. */
-export function isDurableStreamRun(requestContext: AnyColumn): SQL {
-  return sql`${requestContext}->'recovery'->>'kind' IN (${sql.join(
-    DURABLE_STREAM_KINDS.map((kind) => sql`${kind}`),
-    sql`, `
-  )})`
-}
-
 export const StreamRecoveryConfigSchema = z
   .object({
     kind: z.enum(DURABLE_STREAM_KINDS),
@@ -109,17 +91,3 @@ export async function streamRecoveryConfig(
     ...(resourceAuthoringVersion ? { resourceAuthoringVersion } : {}),
   })
 }
-
-const RecordedSecretAuthoritySchema = z.object({
-  secretMountPolicy: secretMountPolicySchema,
-  secretActorUserId: z.string().nullable(),
-})
-
-/**
- * The secret authority a run recorded at admission (`requestContext.secretAuthority`), or
- * undefined when it recorded none or garbage.
- */
-export function parseRecordedSecretAuthority(value: unknown): TurnSecretAuthority | undefined {
-  const recorded = RecordedSecretAuthoritySchema.safeParse(value)
-  return recorded.success ? recorded.data : undefined
-}

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

@@ -60,7 +60,7 @@ import {
   authorizeLifecycleContinuation,
   restoreBillingAdmission,
 } from '@/lib/mothership/request/lifecycle/admission'
-import type { DurableStreamKind } from '@/lib/mothership/request/lifecycle/recovery-config'
+import type { DurableStreamKind } from '@/lib/mothership/request/lifecycle/durable-run'
 import { StreamRetryWindow } from '@/lib/mothership/request/lifecycle/stream-retry'
 import { recordDegraded } from '@/lib/mothership/request/metrics'
 import { AbortReason } from '@/lib/mothership/request/session/abort-reason'