Keep integration streams alive across token refresh (#120)

* adopt explicit relayfile mount contract

* share Slack writeback command root grammar

* chore: apply pr-reviewer fixes for #117

* Keep integration event streams on refreshable tokens

* chore: apply pr-reviewer fixes for #120

---------

Co-authored-by: kjgbot <kjgbot@agentrelay.dev>
Co-authored-by: agent-relay-code[bot] <agent-relay-code[bot]@users.noreply.github.com>

Author: 3 people

package-lock.json

@@ -29,7 +29,7 @@
     "@agent-relay/sdk": "^8.1.2",
     "@agentworkforce/deploy": "^3.0.50",
     "@relayburn/sdk": "^3.2.0",
-    "@relayfile/sdk": "^0.8.10",
+    "@relayfile/sdk": "^0.8.11",
     "@xterm/addon-fit": "^0.10.0",
     "@xterm/addon-web-links": "^0.11.0",
     "@xterm/addon-webgl": "^0.18.0",

package.json

@@ -8,6 +8,7 @@ import {
   IntegrationEventBridge,
   createWorkspaceScopedEventClient,
   integrationSubscriptionSummaries,
+  integrationRelayFileSyncOptions,
   localWatchEventPathsForFilename,
   localWatchRootsFor,
   relayfileSdkPathFiltersFor,
@@ -271,6 +272,20 @@ test('relayfile sdk path filters broaden partial-segment Slack DM globs', () =>
   ])
 })
 
+test('integration event remote stream keeps a refreshable relayfile token provider', () => {
+  const tokenProvider = async () => 'workspace-token'
+  const options = integrationRelayFileSyncOptions({
+    client: {} as never,
+    workspaceId: 'workspace-id',
+    baseUrl: 'https://relayfile.example',
+    tokenProvider,
+    from: 'legacy',
+    paths: ['/slack/channels/*/**']
+  })
+
+  assert.equal(options.token, tokenProvider)
+})
+
 test('integration event remote stream falls back to event feed polling after repeated stream errors', async () => {
   const syncs: FakeRelayFileSync[] = []
   const received: ChangeEvent[] = []

src/main/__tests__/integration-event-bridge.test.ts

@@ -184,6 +184,10 @@ type RelayfileWorkspaceHandle = {
 type TokenProvider = () => Promise<string | undefined>
 type RelayFileSyncFactory = (options: RelayFileSyncOptions) => RelayFileSync
 
+type IntegrationRelayFileSyncOptionsInput = Omit<RelayFileSyncOptions, 'token'> & {
+  tokenProvider: TokenProvider
+}
+
 type IntegrationEventBridgeDeps = {
   broker?: BrokerEventBridge
   getWorkspaceHandle?: () => Promise<RelayfileWorkspaceHandle>
@@ -264,6 +268,16 @@ export function resetIntegrationEventTelemetryForTests(): void {
   aggregatedWarnings.clear()
 }
 
+export function integrationRelayFileSyncOptions(
+  input: IntegrationRelayFileSyncOptionsInput
+): RelayFileSyncOptions {
+  const { tokenProvider, ...options } = input
+  return {
+    ...options,
+    token: tokenProvider
+  }
+}
+
 function toErrorMessage(error: unknown): string {
   if (error instanceof Error) {
     const message = error.message.trim()
@@ -274,21 +288,21 @@ function toErrorMessage(error: unknown): string {
     return message || 'empty string error'
   }
   if (isRecord(error)) {
-    const message = error.message
+    const record = error as Record<string, unknown>
+    const message = record.message
     if (typeof message === 'string' && message.trim()) return message.trim()
-    const reason = error.reason
+    const reason = record.reason
     if (typeof reason === 'string' && reason.trim()) return reason.trim()
     const parts = [
-      typeof error.name === 'string' && error.name.trim() ? error.name.trim() : null,
-      typeof error.type === 'string' && error.type.trim() ? `type=${error.type.trim()}` : null,
-      typeof error.code === 'string' && error.code.trim() ? `code=${error.code.trim()}` : typeof error.code === 'number' ? `code=${error.code}` : null,
-      typeof error.status === 'string' && error.status.trim() ? `status=${error.status.trim()}` : typeof error.status === 'number' ? `status=${error.status}` : null,
-      typeof error.statusCode === 'string' && error.statusCode.trim() ? `statusCode=${error.statusCode.trim()}` : typeof error.statusCode === 'number' ? `statusCode=${error.statusCode}` : null,
-      typeof error.httpStatus === 'string' && error.httpStatus.trim() ? `httpStatus=${error.httpStatus.trim()}` : typeof error.httpStatus === 'number' ? `httpStatus=${error.httpStatus}` : null
+      typeof record.name === 'string' && record.name.trim() ? record.name.trim() : null,
+      typeof record.type === 'string' && record.type.trim() ? `type=${record.type.trim()}` : null,
+      typeof record.code === 'string' && record.code.trim() ? `code=${record.code.trim()}` : typeof record.code === 'number' ? `code=${record.code}` : null,
+      typeof record.status === 'string' && record.status.trim() ? `status=${record.status.trim()}` : typeof record.status === 'number' ? `status=${record.status}` : null,
+      typeof record.statusCode === 'string' && record.statusCode.trim() ? `statusCode=${record.statusCode.trim()}` : typeof record.statusCode === 'number' ? `statusCode=${record.statusCode}` : null,
+      typeof record.httpStatus === 'string' && record.httpStatus.trim() ? `httpStatus=${record.httpStatus.trim()}` : typeof record.httpStatus === 'number' ? `httpStatus=${record.httpStatus}` : null
     ].filter((entry): entry is string => entry !== null)
     if (parts.length > 0) return parts.join(' ')
-
-    const constructorName = (error as { constructor?: { name?: string } }).constructor?.name
+    const constructorName = record.constructor?.name
     if (constructorName && constructorName !== 'Object') return constructorName
 
     try {
@@ -770,11 +784,11 @@ export function createWorkspaceScopedEventClient(
             from: options?.from ?? 'now',
             transport: baseUrl ? 'websocket' : 'polling'
           })
-          sync = syncFactory({
+          sync = syncFactory(integrationRelayFileSyncOptions({
             client,
             workspaceId,
             baseUrl,
-            token,
+            tokenProvider,
             from: options?.from ?? 'now',
             paths: relayfilePathFilters,
             onPollingFallback: (info) => {
@@ -787,7 +801,7 @@ export function createWorkspaceScopedEventClient(
                 }
               )
             }
-          })
+          }))
           sync.on('event', handleEvent)
           sync.on('state', (state) => {
             if (state === 'open') consecutiveStreamErrors = 0

src/main/integration-event-bridge.ts

@@ -271,6 +271,19 @@ describe('IntegrationMountManager', () => {
     })
   })
 
+  it('rejects Slack command roots with traversal segments', async () => {
+    const manager = new IntegrationMountManager()
+
+    await manager.ensureMounted([
+      {
+        provider: 'slack',
+        mountPaths: ['/slack/channels/C123/../messages']
+      }
+    ])
+
+    expect(mock.mountInputs).toHaveLength(0)
+  })
+
   it('scopes bare discovery mounts to the integration provider', async () => {
     const manager = new IntegrationMountManager()
 

src/main/integration-mounts.test.ts

@@ -59,6 +59,13 @@ function remotePathSegments(remotePath: string): string[] {
     .map(sanitizePathSegment)
 }
 
+function hasTraversalPathSegment(remotePath: string): boolean {
+  return remotePath
+    .split(/[\\/]+/u)
+    .map((segment) => segment.trim())
+    .some((segment) => segment === '.' || segment === '..')
+}
+
 export function integrationMountRootForWorkspace(workspaceId: string): string {
   return integrationMountWorkspaceRoot(workspaceId)
 }
@@ -88,6 +95,7 @@ export function integrationProviderRoot(mountPath: string): string | null {
 }
 
 function canonicalIntegrationMountPath(mountPath: string, provider: string): string | null {
+  if (hasTraversalPathSegment(mountPath)) return null
   const providerSegment = sanitizePathSegment(provider.trim().toLowerCase())
   const segments = remotePathSegments(mountPath)
   if (segments[0] === 'discovery') {
@@ -361,7 +369,8 @@ export class IntegrationMountManager {
   }
 
   private createContractLauncher(spec: IntegrationMountSpec, launcher: MountLauncher): MountLauncher {
-    return Object.assign({
+    return {
+      ...launcher,
       start: (input: MountLauncherStart) => launcher.start({
         ...input,
         env: {
@@ -370,9 +379,7 @@ export class IntegrationMountManager {
           RELAYFILE_MOUNT_SYNC_MODE: spec.syncMode
         }
       })
-    }, {
-      __options: (launcher as { __options?: unknown }).__options
-    }) as MountLauncher
+    }
   }
 
   private async removeLegacyIntegrationMountRoot(mountRoot: string): Promise<void> {

src/main/integration-mounts.ts

@@ -1,6 +1,7 @@
 const SLACK_WRITEBACK_COLLECTIONS = new Set(['channels', 'dms', 'users'])
 
 function normalizeRemotePath(path: string): string | null {
+  if (typeof path !== 'string') return null
   const segments = path
     .trim()
     .split(/[\\/]+/)
@@ -10,6 +11,7 @@ function normalizeRemotePath(path: string): string | null {
 }
 
 function isSlackProvider(provider: string): boolean {
+  if (typeof provider !== 'string') return false
   const normalized = provider.trim().toLowerCase()
   return normalized === 'slack' || normalized.startsWith('slack-')
 }