Keep parsed storage sync non-blocking

src/app/api/sources/[sourceId]/chunks/route.test.ts

@@ -498,6 +498,7 @@ describe("GET /api/sources/[sourceId]/chunks", () => {
       revisionKey: "job_1",
       page: 1,
       pageSize: 1,
+      assetUrlPolicy: "durable",
     })
   })
 
@@ -659,6 +660,7 @@ describe("GET /api/sources/[sourceId]/chunks", () => {
       revisionKey: "job_result_1",
       page: 1,
       pageSize: 1,
+      assetUrlPolicy: "durable",
     })
   })
 })

src/domains/chunks/index.ts

@@ -181,8 +181,8 @@ export function toParsedChunkView(
 
 /**
  * Map an SDK `KnowledgeReadChunk` (from `knowledge.readChunks`) to the view
- * model. Display reads do not request durable asset URL hardening; chat
- * hardens the specific assets it returns separately.
+ * model. Display reads request durable asset URLs from the SDK; chat still
+ * hardens only the specific assets it returns separately.
  */
 export function toParsedChunkViewFromReadChunk(
   chunk: KnowledgeReadChunk,

src/domains/chunks/read.test.ts

@@ -20,7 +20,7 @@ function makeReadChunk(overrides: Record<string, unknown> = {}) {
 }
 
 describe("readSourceChunkPage", () => {
-  it("reads a page without durable asset hardening and maps chunks to the view model", async () => {
+  it("reads a page with durable asset URLs and maps chunks to the view model", async () => {
     const readChunks = vi.fn(async () => ({
       document: { localDocumentId: "doc_1" },
       chunks: [
@@ -48,6 +48,7 @@ describe("readSourceChunkPage", () => {
       revisionKey: "rev_1",
       page: 2,
       pageSize: 50,
+      assetUrlPolicy: "durable",
     })
     expect(result.pagination).toEqual({
       page: 2,
@@ -84,6 +85,7 @@ describe("readSourceChunkPage", () => {
       documentId: "doc_1",
       page: 1,
       pageSize: 50,
+      assetUrlPolicy: "durable",
     })
   })
 })
@@ -116,6 +118,13 @@ describe("readAllSourceChunks", () => {
     })
 
     expect(readChunks).toHaveBeenCalledTimes(2)
+    expect(readChunks).toHaveBeenNthCalledWith(1, {
+      documentId: "doc_1",
+      revisionKey: "rev_1",
+      page: 1,
+      pageSize: 200,
+      assetUrlPolicy: "durable",
+    })
     expect(chunks.map((chunk) => chunk.parserChunkId)).toEqual(["c1", "c2"])
   })
 })

src/domains/chunks/read.ts

@@ -16,8 +16,8 @@ type ReadableSource = {
 /**
  * Read a single display page of parsed chunks through the SDK. The SDK serves
  * from configured Blob storage when fresh and falls back to Knowhere remote
- * otherwise. Display reads intentionally do not request durable asset URLs:
- * chat hardens only the specific assets it sends back to the user.
+ * otherwise. Display reads request durable asset URLs so media chunks remain
+ * usable whether the page came from Blob or the remote Knowhere fallback.
  */
 export async function readSourceChunkPage(input: {
   readonly knowledge: Knowledge
@@ -29,6 +29,7 @@ export async function readSourceChunkPage(input: {
     ...(input.source.revisionKey ? { revisionKey: input.source.revisionKey } : {}),
     page: input.params.page,
     pageSize: input.params.pageSize,
+    assetUrlPolicy: "durable",
   })
 
   const chunks = response.chunks.map((chunk) =>
@@ -68,6 +69,7 @@ export async function readAllSourceChunks(input: {
         : {}),
       page,
       pageSize: loadAllPageSize,
+      assetUrlPolicy: "durable",
     })
     for (const chunk of response.chunks) {
       chunks.push(

src/domains/sources/parsed-sync-route-workflow.test.ts

@@ -5,10 +5,6 @@ const mocks = vi.hoisted(() => ({
   makeKnowhereClientWithParsedStorage: vi.fn(),
   releaseSyncCapacity: vi.fn(),
   updateSyncStatus: vi.fn(),
-  findInWorkspace: vi.fn(),
-  markFailed: vi.fn(),
-  markReady: vi.fn(),
-  markSourceReadyAfterReconciliation: vi.fn(),
   loggerInfo: vi.fn(),
   loggerError: vi.fn(),
 }))
@@ -20,16 +16,9 @@ vi.mock("@/integrations/knowhere", () => ({
 vi.mock("./workflow-runtime", () => ({
   sourceWorkflowRuntime: {
     updateSyncStatus: mocks.updateSyncStatus,
-    findInWorkspace: mocks.findInWorkspace,
-    markFailed: mocks.markFailed,
-    markReady: mocks.markReady,
   },
 }))
 
-vi.mock("./source-reconcile-workflow", () => ({
-  markSourceReadyAfterReconciliation: mocks.markSourceReadyAfterReconciliation,
-}))
-
 vi.mock("./parsed-document-sync-capacity", () => ({
   parsedDocumentSyncCapacityGuard: {
     acquire: mocks.acquireSyncCapacity,
@@ -73,16 +62,13 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => {
       },
     })
     mocks.releaseSyncCapacity.mockResolvedValue(undefined)
-    mocks.markSourceReadyAfterReconciliation.mockResolvedValue({
-      status: "ready",
-    })
   })
 
   afterEach(() => {
     vi.clearAllMocks()
   })
 
-  it("marks the source ready when sync completes in one segment", async () => {
+  it("records completion when sync completes in one segment", async () => {
     const syncParsedDocument = vi.fn(async () => ({
       documentId: "doc_1",
       revisionKey: "rev_1",
@@ -105,18 +91,13 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => {
       "source_1",
       { revisionKey: "rev_1", syncStatus: "completed" },
     )
-    expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({
-      workspaceId: "workspace_1",
-      sourceId: "source_1",
-      documentId: "doc_1",
-    })
     expect(mocks.releaseSyncCapacity).toHaveBeenCalledWith({
       leaseToken: "lease_1",
       releaseReason: "completed",
     })
   })
 
-  it("marks ready and triggers a continuation when sync is incomplete", async () => {
+  it("triggers a continuation when sync is incomplete", async () => {
     const syncParsedDocument = vi.fn(async () => ({
       documentId: "doc_1",
       revisionKey: "rev_1",
@@ -145,11 +126,6 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => {
       restore()
     }
 
-    expect(mocks.markSourceReadyAfterReconciliation).toHaveBeenCalledWith({
-      workspaceId: "workspace_1",
-      sourceId: "source_1",
-      documentId: "doc_1",
-    })
     expect(triggered).toHaveLength(1)
     expect(triggered[0]?.segmentIndex).toBe(1)
     expect(triggered[0]?.workflowRunId).toBe("doc_1-sync-rev_1-1")
@@ -249,16 +225,45 @@ describe("parsedSyncRouteWorkflow.runParsedSyncWorkflow", () => {
     ])
     expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled()
   })
+
+  it("does not release capacity when Upstash aborts during a planned step", async () => {
+    const workflowAbort = new Error("planned workflow step")
+    workflowAbort.name = "WorkflowAbort"
+    const syncParsedDocument = vi.fn(async () => ({
+      documentId: "doc_1",
+      revisionKey: "rev_1",
+      completed: true,
+    }))
+    mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({
+      client: {},
+      knowledge: { syncParsedDocument },
+    })
+    const run: RunStep = async (id, task) => {
+      if (id === "sync-0-0") throw workflowAbort
+      return task()
+    }
+
+    await expect(
+      parsedSyncRouteWorkflow.runParsedSyncWorkflow({
+        context: {
+          run,
+          url: "https://notebook.example/api/sources/parsed-sync",
+        },
+        payload: basePayload,
+      }),
+    ).rejects.toThrow("planned workflow step")
+
+    expect(mocks.releaseSyncCapacity).not.toHaveBeenCalled()
+    expect(mocks.updateSyncStatus).not.toHaveBeenCalled()
+  })
 })
 
 describe("parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure", () => {
   afterEach(() => {
     vi.clearAllMocks()
   })
 
-  it("fails a parsing source with failure_stage storage_sync", async () => {
-    mocks.findInWorkspace.mockResolvedValue({ id: "source_1", status: "parsing" })
-
+  it("records failed sync metadata without failing the source", async () => {
     await parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure(
       { ...basePayload, revisionKey: "rev_1" },
       "boom",
@@ -269,28 +274,5 @@ describe("parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure", () => {
       "source_1",
       { revisionKey: "rev_1", syncStatus: "failed", syncError: "boom" },
     )
-    expect(mocks.markFailed).toHaveBeenCalledWith(
-      "workspace_1",
-      "source_1",
-      expect.stringContaining("storage sync failed"),
-      "parsing",
-      "storage_sync",
-    )
-  })
-
-  it("does not fail an already-ready source, only records sync_status", async () => {
-    mocks.findInWorkspace.mockResolvedValue({ id: "source_1", status: "ready" })
-
-    await parsedSyncRouteWorkflow.markSyncFailedAfterWorkflowFailure(
-      basePayload,
-      "boom",
-    )
-
-    expect(mocks.updateSyncStatus).toHaveBeenCalledWith(
-      "workspace_1",
-      "source_1",
-      { revisionKey: undefined, syncStatus: "failed", syncError: "boom" },
-    )
-    expect(mocks.markFailed).not.toHaveBeenCalled()
   })
 })

src/domains/sources/parsed-sync-route-workflow.ts

@@ -1,6 +1,6 @@
 import "server-only"
 
-import { Client, type WorkflowContext } from "@upstash/workflow"
+import { Client, WorkflowAbort, type WorkflowContext } from "@upstash/workflow"
 import type { KnowledgeSyncParsedDocumentResponse } from "@ontos-ai/knowhere-sdk"
 
 import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere"
@@ -10,7 +10,6 @@ import {
   type ParsedSyncPayload,
 } from "./parsed-document-sync-scheduler"
 import { parsedDocumentSyncCapacityGuard } from "./parsed-document-sync-capacity"
-import { markSourceReadyAfterReconciliation } from "./source-reconcile-workflow"
 import { sourceWorkflowRuntime } from "./workflow-runtime"
 
 type ParsedSyncWorkflowContext = Pick<
@@ -72,20 +71,10 @@ async function runParsedSyncWorkflow(input: {
     workspaceId,
   })
 
-  const preSyncReady = await context.run(
-    `source-ready-before-sync-${payload.segmentIndex}`,
-    async () =>
-      markSourceReadyAfterReconciliation({
-        workspaceId,
-        sourceId,
-        documentId,
-      }),
-  )
-  if (preSyncReady.status === "gone") return
-
   let revisionKey = payload.revisionKey
   let completed = false
   let releaseReason: SyncLeaseReleaseReason = "incomplete"
+  let shouldReleaseLease = true
 
   const capacity = await context.run(
     `acquire-sync-capacity-${payload.segmentIndex}`,
@@ -196,24 +185,29 @@ async function runParsedSyncWorkflow(input: {
       return
     }
   } catch (error) {
+    if (isWorkflowControlAbort(error)) {
+      shouldReleaseLease = false
+      throw error
+    }
     releaseReason = "failed"
     throw error
   } finally {
-    await context.run(`release-sync-capacity-${payload.segmentIndex}`, async () =>
-      releaseCapacityLease({
-        leaseToken: capacity.leaseToken,
-        releaseReason,
-        sourceId,
-        documentId,
-      }),
-    )
+    if (shouldReleaseLease) {
+      await context.run(`release-sync-capacity-${payload.segmentIndex}`, async () =>
+        releaseCapacityLease({
+          leaseToken: capacity.leaseToken,
+          releaseReason,
+          sourceId,
+          documentId,
+        }),
+      )
+    }
   }
 
   logger.info("parsed-sync: parsed document sync finished", {
     sourceId,
     documentId,
     revisionKey,
-    status: preSyncReady.status,
   })
 }
 
@@ -261,10 +255,9 @@ async function markSyncFailedAfterWorkflowFailure(
   const normalized = normalizeParsedSyncPayload(payload)
   const reason = getSafeFailureReason(failResponse)
 
-  // Record the storage-sync failure. A source still `parsing` is failed with
-  // failure_stage=storage_sync so a retry resumes sync without reparsing; an
-  // already-ready source is left ready (it still serves via remote fallback),
-  // only its sync_status is marked failed for observability.
+  // Parsed storage is a cache/read model. Exhausted sync failure is recorded for
+  // observability, but the source remains ready and SDK reads can fall back to
+  // Knowhere remote.
   await sourceWorkflowRuntime.updateSyncStatus(
     normalized.workspaceId,
     normalized.sourceId,
@@ -275,26 +268,11 @@ async function markSyncFailedAfterWorkflowFailure(
     },
   )
 
-  const source = await sourceWorkflowRuntime.findInWorkspace(
-    normalized.workspaceId,
-    normalized.sourceId,
-  )
-  if (source?.status === "parsing") {
-    await sourceWorkflowRuntime.markFailed(
-      normalized.workspaceId,
-      normalized.sourceId,
-      `Parsed document storage sync failed: ${reason}`,
-      "parsing",
-      "storage_sync",
-    )
-  }
-
   logger.error("parsed-sync: marked sync failed after workflow failure", {
     workspaceId: normalized.workspaceId,
     sourceId: normalized.sourceId,
     documentId: normalized.documentId,
     segmentIndex: normalized.segmentIndex,
-    sourceStatus: source?.status,
   })
 }
 
@@ -314,6 +292,13 @@ function setContinuationTriggerForTesting(
   }
 }
 
+function isWorkflowControlAbort(error: unknown): boolean {
+  return (
+    (error instanceof WorkflowAbort && error.constructor === WorkflowAbort) ||
+    (error instanceof Error && error.name === "WorkflowAbort")
+  )
+}
+
 export const parsedSyncRouteWorkflow = {
   markSyncFailedAfterWorkflowFailure,
   normalizeParsedSyncPayload,