Add live factory subscription pickup

packages/factory-sdk/src/cli/fleet.ts

@@ -193,7 +193,7 @@ async function runFactoryCommand(
       return 0
     }
 
-    await factory.start()
+    await factory.start({ mode: 'live' })
     writeJson(out, factory.status())
     await waitForShutdown(factory)
     return 0

packages/factory-sdk/src/config/schema.ts

@@ -10,6 +10,12 @@ export const FactoryConfigSchema = z.object({
     labels: z.array(z.string()).default([]),
     assignees: z.array(z.string()).default([]),
   }).default({}),
+  liveSubscription: z.object({
+    transport: z.enum(['subscribe-and-poll', 'subscribe', 'poll']).default('subscribe-and-poll'),
+    pollIntervalMs: z.number().int().min(50).default(5_000),
+    eventLimit: z.number().int().min(1).max(1_000).default(1_000),
+    replaySkewMarginMs: z.number().int().min(0).default(60_000),
+  }).default({}),
   repos: z.object({
     byLabel: z.record(z.string(), z.string()),
     byProject: z.record(z.string(), z.string()).default({}),

packages/factory-sdk/src/index.ts

@@ -56,7 +56,7 @@ export type {
   GithubMergeGatePort,
   GithubMergeGateVerdict,
 } from './github'
-export { BatchTracker, createFactory, FactoryLoop, issueKey, parseLinearIssue } from './orchestrator'
+export { BatchTracker, createFactory, FactoryLoop, issueKey, isRealLinearIssue, parseLinearIssue } from './orchestrator'
 export type { InFlightIssue, QueuedIssue, TrackedAgent } from './orchestrator'
 export {
   HeuristicTriage,
@@ -154,7 +154,9 @@ export type {
   DispatchResult,
   Factory,
   FactoryEventPayload,
+  FactoryLiveSubscriptionOptions,
   FactoryPorts,
+  FactoryStartOptions,
   FactoryStatus,
   IssueRef,
   IterationReport,

packages/factory-sdk/src/mount/relayfile-cloud-mount-client.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it, vi } from 'vitest'
-import type { OperationStatusResponse } from '@relayfile/sdk'
+import type { ChangeEvent, OperationStatusResponse } from '@relayfile/sdk'
 
 import { RelayfileCloudMountClient, type RelayFileClientLike } from './relayfile-cloud-mount-client'
 
@@ -91,6 +91,25 @@ class FakeRelayFileClient implements RelayFileClientLike {
     return { events: this.events, nextCursor: null }
   }
 
+  async listLastNChanges(_limit: number, _context?: { workspaceId: string }): Promise<{ events: ChangeEvent[] }> {
+    return {
+      events: this.events.map((event) => ({
+        id: event.eventId,
+        workspace: 'rw_test',
+        type: 'relayfile.changed' as const,
+        occurredAt: event.timestamp,
+        resource: {
+          path: event.path,
+          kind: 'file',
+          id: event.path,
+          provider: 'linear',
+        },
+        summary: {},
+        expand: async () => ({ level: 'summary' as const, path: event.path, summary: {} }),
+      }) as unknown as ChangeEvent),
+    }
+  }
+
   async getResourceAtEvent(eventId: string, context?: { workspaceId: string }) {
     return { path: `/events/${eventId}.json`, data: context, digest: eventId }
   }
@@ -145,6 +164,21 @@ describe('RelayfileCloudMountClient', () => {
     expect(fake.getEventsCalls[0]).toEqual({ workspaceId: 'rw_test', opts: { cursor: 'evt-0', limit: 10 } })
   })
 
+  it('selects the numeric event high-watermark instead of lexicographic max', async () => {
+    const fake = new FakeRelayFileClient()
+    fake.events = ['7', '8', '9', '10', '11'].map((eventId) => ({
+      eventId,
+      type: 'file.updated' as const,
+      path: `/linear/issues/AR-${eventId}.json`,
+      revision: eventId,
+      timestamp: '2026-01-01T00:00:00.000Z',
+    }))
+    const mount = new RelayfileCloudMountClient({ workspaceId: 'rw_test', client: fake })
+
+    await expect(mount.getEventHighWatermark()).resolves.toBe('11')
+    expect(fake.getEventsCalls).toEqual([])
+  })
+
   it('writes through the RelayFileClient with workspace id and live baseRevision', async () => {
     const fake = new FakeRelayFileClient()
     fake.files.set('/linear/issues/AR-1.json', {

packages/factory-sdk/src/mount/relayfile-cloud-mount-client.ts

@@ -13,14 +13,13 @@ import {
   type RelayfileCloudTokenSet,
   type ResourceAtEventResult,
   type OperationStatusResponse,
-  type SubscribeOptions,
   type Subscription,
   type TreeResponse,
   type WriteFileInput,
   type WriteQueuedResponse,
 } from '@relayfile/sdk'
 
-import type { EventPage, MountClient } from '../ports'
+import type { EventPage, MountClient, SubscribeOptions } from '../ports'
 import {
   createWorkspaceScopedEventClient,
   type RelayfileEventClient,
@@ -53,6 +52,7 @@ export type RelayFileClientLike =
     deleteFile(input: DeleteFileInput): Promise<WriteQueuedResponse>
     listTree(workspaceId: string, options?: ListTreeOptions): Promise<TreeResponse>
     getEvents(workspaceId: string, options?: GetEventsOptions): Promise<EventFeedResponse>
+    listLastNChanges?(limit: number, context?: { workspaceId: string; token?: string }): Promise<{ events: ChangeEvent[] }>
     getResourceAtEvent(eventId: string, context?: { workspaceId: string; token?: string }): Promise<ResourceAtEventResult>
     getOp?(workspaceId: string, opId: string): Promise<OperationStatusResponse>
     getToken?(): Promise<string> | string
@@ -230,6 +230,15 @@ export class RelayfileCloudMountClient implements MountClient {
     }
   }
 
+  async getEventHighWatermark(opts: { provider?: string } = {}): Promise<string | undefined> {
+    if (!this.#client.listLastNChanges) return undefined
+    const response = await this.#client.listLastNChanges(10, { workspaceId: this.workspaceId })
+    const events = opts.provider
+      ? response.events.filter((event) => event.resource.provider === opts.provider)
+      : response.events
+    return maxEventId(events.map((event) => event.id))
+  }
+
   async confirmWrite(
     path: string,
     opts: { timeoutMs?: number } = {},
@@ -347,3 +356,31 @@ const errorMessage = (error: unknown): string =>
   error instanceof Error ? error.message : String(error)
 
 const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms))
+
+const maxEventId = (ids: string[]): string | undefined => {
+  let max: string | undefined
+  for (const id of ids) {
+    if (!max || compareEventIds(id, max) > 0) {
+      max = id
+    }
+  }
+  return max
+}
+
+const compareEventIds = (left: string, right: string): number => {
+  const leftSequence = eventSequenceNumber(left)
+  const rightSequence = eventSequenceNumber(right)
+  if (leftSequence !== undefined && rightSequence !== undefined) {
+    return leftSequence - rightSequence
+  }
+  return left.localeCompare(right)
+}
+
+const eventSequenceNumber = (eventId: string): number | undefined => {
+  const whole = Number(eventId)
+  if (Number.isFinite(whole)) return whole
+  const trailing = eventId.match(/(\d+)$/u)?.[1]
+  if (!trailing) return undefined
+  const parsed = Number(trailing)
+  return Number.isFinite(parsed) ? parsed : undefined
+}