From f158cd70f8a0c32093e5f2526dc003d7adb68f96 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:11:43 +0200
Subject: [PATCH 1/6] feat(session-ingest-contracts): add cloneUpToMessageId
 and clone_cut_not_found

---
 packages/session-ingest-contracts/src/rpc-contract.ts | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/packages/session-ingest-contracts/src/rpc-contract.ts b/packages/session-ingest-contracts/src/rpc-contract.ts
index 0963c8d751b..d09de0114b6 100644
--- a/packages/session-ingest-contracts/src/rpc-contract.ts
+++ b/packages/session-ingest-contracts/src/rpc-contract.ts
@@ -163,6 +163,9 @@ export const createSessionForCloudAgentSchema = z.object({
   // Compatibility: old callers omit this field and create empty destination
   // storage. Remove that path only when all deployed callers require cloning.
   cloneFromKiloSessionId: sessionIdSchema.optional(),
+  // Inclusive cut for a clone: keep messages with ID <= this ID (and their
+  // parts) and drop the session diff. Only meaningful with cloneFromKiloSessionId.
+  cloneUpToMessageId: messageIdSchema.optional(),
 });
 export type CreateSessionForCloudAgentParams = z.input<typeof createSessionForCloudAgentSchema>;
 
@@ -178,7 +181,8 @@ export type SessionCloneRejectionCode =
   | 'missing_source_body'
   | 'source_digest_changed'
   | 'destination_conflict'
-  | 'clone_setup_failed';
+  | 'clone_setup_failed'
+  | 'clone_cut_not_found';
 
 /**
  * Clone acknowledgement carried by the `ready` result. `sessionId` is the

From 2c0d6743be7682a7486fe0268877cafb0d62a395 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:11:43 +0200
Subject: [PATCH 2/6] feat(session-ingest): add shared inclusive message cut
 filter

---
 .../src/util/message-cut.test.ts              | 39 +++++++++++++++++++
 .../session-ingest/src/util/message-cut.ts    | 27 +++++++++++++
 2 files changed, 66 insertions(+)
 create mode 100644 services/session-ingest/src/util/message-cut.test.ts
 create mode 100644 services/session-ingest/src/util/message-cut.ts

diff --git a/services/session-ingest/src/util/message-cut.test.ts b/services/session-ingest/src/util/message-cut.test.ts
new file mode 100644
index 00000000000..a05c08e084b
--- /dev/null
+++ b/services/session-ingest/src/util/message-cut.test.ts
@@ -0,0 +1,39 @@
+import { describe, expect, it } from 'vitest';
+
+import { isItemWithinMessageCut } from './message-cut';
+
+describe('isItemWithinMessageCut', () => {
+  const cut = 'msg_0002';
+
+  it('always keeps the session row', () => {
+    expect(isItemWithinMessageCut({ itemType: 'session', itemId: 'session' }, cut)).toBe(true);
+  });
+
+  it('keeps messages up to and including the cut and drops later ones', () => {
+    expect(isItemWithinMessageCut({ itemType: 'message', itemId: 'message/msg_0001' }, cut)).toBe(
+      true
+    );
+    expect(isItemWithinMessageCut({ itemType: 'message', itemId: 'message/msg_0002' }, cut)).toBe(
+      true
+    );
+    expect(isItemWithinMessageCut({ itemType: 'message', itemId: 'message/msg_0003' }, cut)).toBe(
+      false
+    );
+  });
+
+  it('keeps parts by the message ID in their item identity', () => {
+    expect(isItemWithinMessageCut({ itemType: 'part', itemId: 'msg_0002/prt_0009' }, cut)).toBe(
+      true
+    );
+    expect(isItemWithinMessageCut({ itemType: 'part', itemId: 'msg_0003/prt_0001' }, cut)).toBe(
+      false
+    );
+  });
+
+  it('always drops the session diff and other item types', () => {
+    expect(isItemWithinMessageCut({ itemType: 'session_diff', itemId: 'session_diff' }, cut)).toBe(
+      false
+    );
+    expect(isItemWithinMessageCut({ itemType: 'kilo_meta', itemId: 'kilo_meta' }, cut)).toBe(false);
+  });
+});
diff --git a/services/session-ingest/src/util/message-cut.ts b/services/session-ingest/src/util/message-cut.ts
new file mode 100644
index 00000000000..ef27e6df224
--- /dev/null
+++ b/services/session-ingest/src/util/message-cut.ts
@@ -0,0 +1,27 @@
+/**
+ * Inclusive message cut shared by session clone and session export.
+ *
+ * With a cut, a transcript keeps the session row, every message whose ID is
+ * <= `upToMessageId` (message IDs are ascending, compared as strings), and the
+ * parts of those messages. The session diff is always dropped: it describes the
+ * whole session, so files are rebuilt from the kept messages' summary diffs.
+ *
+ * The test reads only the ingest item identity (`message/{messageId}` for
+ * messages, `{messageId}/{partId}` for parts), never the item body, so it works
+ * the same for inline and R2-backed rows.
+ */
+export function isItemWithinMessageCut(
+  item: { itemType: string; itemId: string },
+  upToMessageId: string
+): boolean {
+  switch (item.itemType) {
+    case 'session':
+      return true;
+    case 'message':
+      return item.itemId.slice('message/'.length) <= upToMessageId;
+    case 'part':
+      return item.itemId.split('/', 1)[0] <= upToMessageId;
+    default:
+      return false;
+  }
+}

From 45616ded66b1bc20f907336eae115d0756f3be8a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:11:44 +0200
Subject: [PATCH 3/6] feat(session-ingest): clone a session up to a message cut

---
 .../session-ingest/src/clone/session-clone.ts |  87 ++++--
 .../src/dos/SessionIngestDO.test.ts           |  93 +++++++
 .../session-ingest/src/dos/SessionIngestDO.ts | 137 ++++++---
 .../src/session-ingest-rpc.test.ts            | 262 +++++++++++++++++-
 .../session-ingest/src/session-ingest-rpc.ts  |   6 +-
 5 files changed, 509 insertions(+), 76 deletions(-)

diff --git a/services/session-ingest/src/clone/session-clone.ts b/services/session-ingest/src/clone/session-clone.ts
index 19c76558391..190aeae8f88 100644
--- a/services/session-ingest/src/clone/session-clone.ts
+++ b/services/session-ingest/src/clone/session-clone.ts
@@ -12,12 +12,14 @@ import {
   CLONE_ITEM_TYPES,
   type CloneBatchRow,
   type CloneItemType,
+  type CloneStageIdentity,
   type ExportCloneBatchResult,
   type FinalizeCloneStageResult,
   type IngestOrderCursor,
   type InspectCloneStageResult,
   type StageCloneBatchResult,
 } from '../dos/SessionIngestDO';
+import { isItemWithinMessageCut } from '../util/message-cut';
 
 const CLONE_BATCH_MAX_ROWS = 100;
 const CLONE_BATCH_MAX_MATERIALIZED_BYTES = 4 * 1024 * 1024;
@@ -36,6 +38,8 @@ export type CloneSessionParams = {
   sourceOrganizationId: string | null;
   destinationOrganizationId: string | null;
   destinationTitle?: string;
+  /** Inclusive message cut; absent clones the full transcript. */
+  upToMessageId?: string;
 };
 
 type RewrittenCloneRow = {
@@ -70,13 +74,17 @@ function sourceDigestLine(row: CloneBatchRow): string {
 }
 
 /**
- * Keep only clone-eligible item types. The source DO already filters its
- * export to `CLONE_ITEM_TYPES`; this guard keeps the helper consistent if a
- * caller ever hands it rows outside that set, so the staged count, the resume
- * digest, and the final verification digest all agree.
+ * Keep only clone-eligible item types inside the optional message cut. The
+ * source DO already filters its export to `CLONE_ITEM_TYPES`; this guard keeps
+ * the helper consistent if a caller ever hands it rows outside that set. Every
+ * pass (copy, resume digest, final verification digest) uses this predicate,
+ * so the staged count and digests all agree.
  */
-function cloneEligibleRows(rows: CloneBatchRow[]): CloneBatchRow[] {
-  return rows.filter(row => CLONE_ITEM_TYPES.includes(row.itemType));
+function isCloneEligibleRow(row: CloneBatchRow, upToMessageId: string | undefined): boolean {
+  return (
+    CLONE_ITEM_TYPES.includes(row.itemType) &&
+    (upToMessageId === undefined || isItemWithinMessageCut(row, upToMessageId))
+  );
 }
 
 function parseItemJson(value: string): unknown {
@@ -238,12 +246,16 @@ async function materializeBatchWithinBudget(
   r2: R2Bucket,
   sourceSessionId: string,
   destinationSessionId: string,
+  upToMessageId: string | undefined,
   writtenR2Keys: string[]
 ): Promise<{ rows: RewrittenCloneRow[]; nextCursor: IngestOrderCursor | null; done: boolean }> {
-  const eligible = cloneEligibleRows(exported.rows);
   const rows: RewrittenCloneRow[] = [];
   let bytes = 0;
-  for (const row of eligible) {
+  // Source rows consumed by this batch, counting rows the cut skips, so the
+  // resume cursor lands just before the first row that did not fit.
+  let consumedSourceRows = exported.rows.length;
+  for (const [index, row] of exported.rows.entries()) {
+    if (!isCloneEligibleRow(row, upToMessageId)) continue;
     const rewritten = await materializeAndRewriteRow(
       row,
       r2,
@@ -254,6 +266,7 @@ async function materializeBatchWithinBudget(
       rows.length > 0 &&
       bytes + rewritten.materializedByteLength > CLONE_BATCH_MAX_MATERIALIZED_BYTES
     ) {
+      consumedSourceRows = index;
       break;
     }
     if (rewritten.itemDataR2Key !== null) {
@@ -264,7 +277,7 @@ async function materializeBatchWithinBudget(
     bytes += rewritten.materializedByteLength;
   }
 
-  if (rows.length === eligible.length) {
+  if (consumedSourceRows === exported.rows.length) {
     return { rows, nextCursor: exported.nextCursor, done: exported.done };
   }
 
@@ -272,14 +285,15 @@ async function materializeBatchWithinBudget(
   // the correct resume cursor (the row id is not part of CloneBatchRow).
   const prefix = await withDORetry<SessionIngestDOStub, ExportCloneBatchResult>(
     sourceStub,
-    stub => stub.exportCloneBatch(cursor, rows.length),
+    stub => stub.exportCloneBatch(cursor, consumedSourceRows),
     'SessionIngestDO.exportCloneBatch'
   );
   return { rows, nextCursor: prefix.nextCursor, done: false };
 }
 
 async function exportAllSourceRows(
-  sourceStub: () => SessionIngestDOStub
+  sourceStub: () => SessionIngestDOStub,
+  upToMessageId: string | undefined
 ): Promise<{ digestLines: string[]; itemCount: number }> {
   const digestLines: string[] = [];
   let cursor: IngestOrderCursor | null = null;
@@ -290,7 +304,9 @@ async function exportAllSourceRows(
       'SessionIngestDO.exportCloneBatch'
     );
     if (exported.rows.length === 0) break;
-    for (const row of cloneEligibleRows(exported.rows)) digestLines.push(sourceDigestLine(row));
+    for (const row of exported.rows) {
+      if (isCloneEligibleRow(row, upToMessageId)) digestLines.push(sourceDigestLine(row));
+    }
     cursor = exported.nextCursor;
     if (exported.done) break;
   }
@@ -299,7 +315,8 @@ async function exportAllSourceRows(
 
 async function rebuildPrefixDigestLines(
   sourceStub: () => SessionIngestDOStub,
-  copiedItemCount: number
+  copiedItemCount: number,
+  upToMessageId: string | undefined
 ): Promise<string[]> {
   const digestLines: string[] = [];
   let cursor: IngestOrderCursor | null = null;
@@ -310,7 +327,8 @@ async function rebuildPrefixDigestLines(
       'SessionIngestDO.exportCloneBatch'
     );
     if (exported.rows.length === 0) break;
-    for (const row of cloneEligibleRows(exported.rows)) {
+    for (const row of exported.rows) {
+      if (!isCloneEligibleRow(row, upToMessageId)) continue;
       if (digestLines.length >= copiedItemCount) break;
       digestLines.push(sourceDigestLine(row));
     }
@@ -366,6 +384,7 @@ export async function cloneSessionIntoDestination(
     sourceOrganizationId,
     destinationOrganizationId,
     destinationTitle,
+    upToMessageId,
   } = params;
 
   if (sourceOrganizationId !== destinationOrganizationId) {
@@ -379,15 +398,33 @@ export async function cloneSessionIntoDestination(
   // Destination R2 keys written by this request that were never staged. Deleted
   // on reject so a failed clone leaves no orphaned destination bodies.
   const writtenDestinationR2Keys: string[] = [];
+  // The stage identity includes the cut, so a stage built for another cut is a
+  // mismatch exactly like a stage built for another source.
+  const stageIdentity: CloneStageIdentity = {
+    sourceSessionId,
+    destinationSessionId,
+    ...(upToMessageId !== undefined ? { upToMessageId } : {}),
+  };
+
+  if (upToMessageId !== undefined) {
+    const hasCutMessage = await withDORetry<SessionIngestDOStub, boolean>(
+      sourceStub,
+      stub => stub.hasMessageItem(upToMessageId),
+      'SessionIngestDO.hasMessageItem'
+    );
+    if (!hasCutMessage) {
+      return { status: 'rejected', code: 'clone_cut_not_found' };
+    }
+  }
 
   const inspected = await withDORetry<SessionIngestDOStub, InspectCloneStageResult>(
     destinationStub,
-    stub => stub.inspectCloneStage({ sourceSessionId, destinationSessionId }),
+    stub => stub.inspectCloneStage(stageIdentity),
     'SessionIngestDO.inspectCloneStage'
   );
 
   if (inspected.status === 'complete') {
-    const { itemCount } = await exportAllSourceRows(sourceStub);
+    const { itemCount } = await exportAllSourceRows(sourceStub, upToMessageId);
     return { status: 'ready', copiedItemCount: itemCount };
   }
 
@@ -399,7 +436,7 @@ export async function cloneSessionIntoDestination(
   if (inspected.status === 'in_progress') {
     cursor = inspected.nextCursor;
     copiedItemCount = inspected.copiedItemCount;
-    digestLines = await rebuildPrefixDigestLines(sourceStub, copiedItemCount);
+    digestLines = await rebuildPrefixDigestLines(sourceStub, copiedItemCount, upToMessageId);
     const rebuiltDigest = await sha256Hex(digestLines.join('\n'));
     if (rebuiltDigest !== inspected.rollingDigest) {
       await resetDestinationStage(destinationStub);
@@ -407,9 +444,9 @@ export async function cloneSessionIntoDestination(
     }
     needsCloneLoop = cursor !== null;
   } else if (inspected.status === 'mismatch') {
-    // A complete destination that belongs to another source is already
+    // A complete destination that belongs to another source or cut is already
     // published: reject terminally and never reset it. An incomplete stage from
-    // another source is unpublished and may be reset before restarting.
+    // another source or cut is unpublished and may be reset before restarting.
     if (inspected.storedStage === 'complete') {
       return { status: 'rejected', code: 'destination_conflict' };
     }
@@ -435,6 +472,7 @@ export async function cloneSessionIntoDestination(
           r2,
           sourceSessionId,
           destinationSessionId,
+          upToMessageId,
           writtenDestinationR2Keys
         );
       } catch (error) {
@@ -456,8 +494,7 @@ export async function cloneSessionIntoDestination(
         destinationStub,
         stub =>
           stub.stageCloneBatch({
-            sourceSessionId,
-            destinationSessionId,
+            ...stageIdentity,
             rows: batch.rows.map(row => ({
               itemId: row.itemId,
               itemType: row.itemType,
@@ -490,7 +527,7 @@ export async function cloneSessionIntoDestination(
 
   // Final verification: re-export the source from scratch and compare the
   // source digest and item count against the staged rolling digest and count.
-  const final = await exportAllSourceRows(sourceStub);
+  const final = await exportAllSourceRows(sourceStub, upToMessageId);
   const finalDigest = await sha256Hex(final.digestLines.join('\n'));
   const rollingDigest = await sha256Hex(digestLines.join('\n'));
 
@@ -505,8 +542,7 @@ export async function cloneSessionIntoDestination(
       destinationStub,
       stub =>
         stub.stageCloneBatch({
-          sourceSessionId,
-          destinationSessionId,
+          ...stageIdentity,
           rows: [
             {
               itemId: 'session',
@@ -532,8 +568,7 @@ export async function cloneSessionIntoDestination(
     destinationStub,
     stub =>
       stub.finalizeCloneStage({
-        sourceSessionId,
-        destinationSessionId,
+        ...stageIdentity,
         finalDigest,
         finalItemCount: copiedItemCount,
       }),
diff --git a/services/session-ingest/src/dos/SessionIngestDO.test.ts b/services/session-ingest/src/dos/SessionIngestDO.test.ts
index df038a5a107..0228e5a26ed 100644
--- a/services/session-ingest/src/dos/SessionIngestDO.test.ts
+++ b/services/session-ingest/src/dos/SessionIngestDO.test.ts
@@ -2057,6 +2057,99 @@ describe('SessionIngestDO clone primitives', () => {
     ).toEqual({ status: 'empty' });
   });
 
+  it('treats a stage built for another message cut as a different clone', () => {
+    const harness = makeCloneHarness();
+    const cutIdentity = {
+      sourceSessionId: 'src',
+      destinationSessionId: 'dst',
+      upToMessageId: 'msg_02',
+    };
+
+    expect(
+      harness.durableObject.stageCloneBatch({
+        ...cutIdentity,
+        rows: [
+          {
+            itemId: 'message/msg_01',
+            itemType: 'message' as const,
+            itemData: '{}',
+            itemDataR2Key: null,
+            ingestedAt: 1,
+          },
+        ],
+        nextCursor: { ingestedAt: 1, id: 1 },
+        rollingDigest: 'digest-cut',
+        copiedItemCount: 1,
+      }).status
+    ).toBe('staged');
+    expect(harness.meta.get('cloneUpToMessageId')).toBe('msg_02');
+
+    expect(harness.durableObject.inspectCloneStage(cutIdentity)).toMatchObject({
+      status: 'in_progress',
+      copiedItemCount: 1,
+    });
+    for (const other of [
+      { sourceSessionId: 'src', destinationSessionId: 'dst' },
+      { ...cutIdentity, upToMessageId: 'msg_03' },
+    ]) {
+      expect(harness.durableObject.inspectCloneStage(other)).toMatchObject({
+        status: 'mismatch',
+        storedStage: 'in_progress',
+      });
+      expect(
+        harness.durableObject.stageCloneBatch({
+          ...other,
+          rows: [],
+          nextCursor: null,
+          rollingDigest: 'digest-other',
+          copiedItemCount: 0,
+        }).status
+      ).toBe('mismatch');
+      expect(
+        harness.durableObject.finalizeCloneStage({
+          ...other,
+          finalDigest: 'digest-cut',
+          finalItemCount: 1,
+        }).status
+      ).toBe('mismatch');
+    }
+
+    expect(
+      harness.durableObject.finalizeCloneStage({
+        ...cutIdentity,
+        finalDigest: 'digest-cut',
+        finalItemCount: 1,
+      }).status
+    ).toBe('complete');
+    expect(harness.durableObject.inspectCloneStage(cutIdentity)).toMatchObject({
+      status: 'complete',
+    });
+    expect(
+      harness.durableObject.inspectCloneStage({ ...cutIdentity, upToMessageId: 'msg_01' })
+    ).toMatchObject({ status: 'mismatch', storedStage: 'complete' });
+  });
+
+  it('records no message cut for a full clone stage', () => {
+    const harness = makeCloneHarness();
+    harness.durableObject.stageCloneBatch({
+      sourceSessionId: 'src',
+      destinationSessionId: 'dst',
+      rows: [],
+      nextCursor: null,
+      rollingDigest: 'digest',
+      copiedItemCount: 0,
+    });
+
+    expect(harness.meta.has('cloneUpToMessageId')).toBe(false);
+    expect(
+      harness.durableObject.inspectCloneStage({
+        sourceSessionId: 'src',
+        destinationSessionId: 'dst',
+        upToMessageId: 'msg_01',
+      })
+    ).toMatchObject({ status: 'mismatch' });
+  });
+
   it('swallows an R2 delete failure so resetCloneStage still returns a typed reject', async () => {
     const harness = makeCloneHarness([itemRow(1, 'message/src', 'message', 1, 'items/blob')]);
     harness.deleteObject.mockRejectedValueOnce(new Error('r2 unavailable'));
diff --git a/services/session-ingest/src/dos/SessionIngestDO.ts b/services/session-ingest/src/dos/SessionIngestDO.ts
index c2538a346a4..3329039c712 100644
--- a/services/session-ingest/src/dos/SessionIngestDO.ts
+++ b/services/session-ingest/src/dos/SessionIngestDO.ts
@@ -12,6 +12,7 @@ import type { Env } from '../env';
 import type { IngestBatch } from '../types/session-sync';
 import type { SessionDataItem } from '../types/session-sync';
 import { getItemIdentity, getPartItemIdentityRange } from '../util/compaction';
+import { isItemWithinMessageCut } from '../util/message-cut';
 import {
   extractNormalizedGitBranchFromItem,
   extractNormalizedGitUrlFromItem,
@@ -141,7 +142,8 @@ type CloneMetaKey =
   | 'cloneNextCursor'
   | 'cloneRollingDigest'
   | 'cloneCopiedItemCount'
-  | 'cloneCompleteSourceId';
+  | 'cloneCompleteSourceId'
+  | 'cloneUpToMessageId';
 
 function readCloneMeta(db: DrizzleSqliteDODatabase, key: CloneMetaKey): string | null {
   const row = db
@@ -275,9 +277,17 @@ export type ExportCloneBatchResult = {
   digest: string;
 };
 
-export type StageCloneBatchParams = {
+/**
+ * A clone stage is identified by source, destination, and the optional
+ * inclusive message cut. Absent `upToMessageId` means the full transcript.
+ */
+export type CloneStageIdentity = {
   sourceSessionId: string;
   destinationSessionId: string;
+  upToMessageId?: string;
+};
+
+export type StageCloneBatchParams = CloneStageIdentity & {
   rows: CloneBatchRow[];
   nextCursor: IngestOrderCursor | null;
   rollingDigest: string;
@@ -300,10 +310,7 @@ export type StageCloneBatchResult =
       storedDestinationSessionId: string | null;
     };
 
-export type InspectCloneStageParams = {
-  sourceSessionId: string;
-  destinationSessionId: string;
-};
+export type InspectCloneStageParams = CloneStageIdentity;
 
 export type InspectCloneStageResult =
   | { status: 'empty' }
@@ -323,9 +330,7 @@ export type InspectCloneStageResult =
       storedStage: 'complete' | 'in_progress';
     };
 
-export type FinalizeCloneStageParams = {
-  sourceSessionId: string;
-  destinationSessionId: string;
+export type FinalizeCloneStageParams = CloneStageIdentity & {
   finalDigest: string;
   finalItemCount: number;
 };
@@ -348,16 +353,34 @@ export type FinalizeCloneStageResult =
 
 type CloneStageState =
   | { kind: 'empty' }
-  | { kind: 'complete'; sourceSessionId: string; destinationSessionId: string }
+  | {
+      kind: 'complete';
+      sourceSessionId: string;
+      destinationSessionId: string;
+      upToMessageId: string | null;
+    }
   | {
       kind: 'in_progress';
       sourceSessionId: string;
       destinationSessionId: string;
+      upToMessageId: string | null;
       nextCursor: IngestOrderCursor | null;
       rollingDigest: string;
       copiedItemCount: number;
     };
 
+/** A stage built for another source, destination, or cut is a different clone. */
+function isSameCloneStage(
+  stage: Exclude<CloneStageState, { kind: 'empty' }>,
+  params: CloneStageIdentity
+): boolean {
+  return (
+    stage.sourceSessionId === params.sourceSessionId &&
+    stage.destinationSessionId === params.destinationSessionId &&
+    stage.upToMessageId === (params.upToMessageId ?? null)
+  );
+}
+
 async function computeCloneBatchDigest(rows: CloneBatchRow[]): Promise<string> {
   const joined = rows
     .map(row => `${row.itemId}:${row.itemType}:${row.itemDataR2Key ?? row.itemData}`)
@@ -870,7 +893,12 @@ export class SessionIngestDO extends DurableObject<Env> {
     return readKiloSdkMessages(this.db, this.env.SESSION_INGEST_R2, params);
   }
 
-  async getAllStream(): Promise<ReadableStream<Uint8Array>> {
+  /**
+   * Stream the session export. With `upToMessageId`, keep only messages inside
+   * the inclusive cut (see `isItemWithinMessageCut`) and emit an empty
+   * `sessionDiff`; without it the output is the full export.
+   */
+  async getAllStream(upToMessageId?: string): Promise<ReadableStream<Uint8Array>> {
     const db = this.db;
     const r2 = this.env.SESSION_INGEST_R2;
     const encoder = new TextEncoder();
@@ -923,6 +951,15 @@ export class SessionIngestDO extends DurableObject<Env> {
             msgCursor = ingestOrderCursor(msgBatch[msgBatch.length - 1]);
 
             for (const msgRow of msgBatch) {
+              if (
+                upToMessageId !== undefined &&
+                !isItemWithinMessageCut(
+                  { itemType: 'message', itemId: msgRow.item_id },
+                  upToMessageId
+                )
+              ) {
+                continue;
+              }
               if (!firstMsg) controller.enqueue(encoder.encode(','));
               firstMsg = false;
 
@@ -974,15 +1011,18 @@ export class SessionIngestDO extends DurableObject<Env> {
 
           controller.enqueue(encoder.encode(']'));
           controller.enqueue(encoder.encode(',"sessionDiff":'));
-          const diffRow = db
-            .select({
-              item_data: ingestItems.item_data,
-              item_data_r2_key: ingestItems.item_data_r2_key,
-            })
-            .from(ingestItems)
-            .where(eq(ingestItems.item_type, 'session_diff'))
-            .limit(1)
-            .get();
+          const diffRow =
+            upToMessageId === undefined
+              ? db
+                  .select({
+                    item_data: ingestItems.item_data,
+                    item_data_r2_key: ingestItems.item_data_r2_key,
+                  })
+                  .from(ingestItems)
+                  .where(eq(ingestItems.item_type, 'session_diff'))
+                  .limit(1)
+                  .get()
+              : undefined;
           if (diffRow) {
             await enqueueItemData(controller, diffRow, r2, encoder, '[]');
           } else {
@@ -1381,13 +1421,27 @@ export class SessionIngestDO extends DurableObject<Env> {
     return { rows: batchRows, nextCursor, done, digest };
   }
 
+  /** True when this session holds a `message` row with exactly this message ID. */
+  hasMessageItem(messageId: string): boolean {
+    const row = this.db
+      .select({ id: ingestItems.id })
+      .from(ingestItems)
+      .where(
+        and(eq(ingestItems.item_type, 'message'), eq(ingestItems.item_id, `message/${messageId}`))
+      )
+      .get();
+    return row !== undefined;
+  }
+
   private readCloneStage(): CloneStageState {
     const completeSource = readCloneMeta(this.db, 'cloneCompleteSourceId');
+    const upToMessageId = readCloneMeta(this.db, 'cloneUpToMessageId');
     if (completeSource !== null) {
       return {
         kind: 'complete',
         sourceSessionId: completeSource,
         destinationSessionId: readCloneMeta(this.db, 'cloneDestinationSessionId') ?? '',
+        upToMessageId,
       };
     }
 
@@ -1407,6 +1461,7 @@ export class SessionIngestDO extends DurableObject<Env> {
         kind: 'in_progress',
         sourceSessionId: source,
         destinationSessionId: destination,
+        upToMessageId,
         nextCursor,
         rollingDigest: readCloneMeta(this.db, 'cloneRollingDigest') ?? '',
         copiedItemCount: Number(readCloneMeta(this.db, 'cloneCopiedItemCount') ?? '0') || 0,
@@ -1419,18 +1474,15 @@ export class SessionIngestDO extends DurableObject<Env> {
   /**
    * Persist one batch of already-rewritten clone rows and advance the clone stage
    * state. Idempotent for the same source: re-staging the same batch upserts the
-   * same rows and rewrites the same meta. A different stored source or destination
-   * is rejected without mutation.
+   * same rows and rewrites the same meta. A different stored source, destination,
+   * or message cut is rejected without mutation.
    */
   stageCloneBatch(params: StageCloneBatchParams): StageCloneBatchResult {
     if (this.isDeleted()) throw new Error('Session deleted');
     const stage = this.readCloneStage();
 
     if (stage.kind === 'complete') {
-      if (
-        stage.sourceSessionId === params.sourceSessionId &&
-        stage.destinationSessionId === params.destinationSessionId
-      ) {
+      if (isSameCloneStage(stage, params)) {
         return {
           status: 'complete',
           sourceSessionId: stage.sourceSessionId,
@@ -1444,17 +1496,12 @@ export class SessionIngestDO extends DurableObject<Env> {
       };
     }
 
-    if (stage.kind === 'in_progress') {
-      if (
-        stage.sourceSessionId !== params.sourceSessionId ||
-        stage.destinationSessionId !== params.destinationSessionId
-      ) {
-        return {
-          status: 'mismatch',
-          storedSourceSessionId: stage.sourceSessionId,
-          storedDestinationSessionId: stage.destinationSessionId,
-        };
-      }
+    if (stage.kind === 'in_progress' && !isSameCloneStage(stage, params)) {
+      return {
+        status: 'mismatch',
+        storedSourceSessionId: stage.sourceSessionId,
+        storedDestinationSessionId: stage.destinationSessionId,
+      };
     }
 
     for (const row of params.rows) {
@@ -1481,6 +1528,11 @@ export class SessionIngestDO extends DurableObject<Env> {
 
     writeCloneMeta(this.db, 'cloneSourceSessionId', params.sourceSessionId);
     writeCloneMeta(this.db, 'cloneDestinationSessionId', params.destinationSessionId);
+    // Only cut clones record a cut: an empty or same-identity stage never holds
+    // this key for a full clone, so full-clone state is unchanged.
+    if (params.upToMessageId !== undefined) {
+      writeCloneMeta(this.db, 'cloneUpToMessageId', params.upToMessageId);
+    }
     writeCloneMeta(this.db, 'cloneNextCursor', JSON.stringify(params.nextCursor));
     writeCloneMeta(this.db, 'cloneRollingDigest', params.rollingDigest);
     writeCloneMeta(this.db, 'cloneCopiedItemCount', String(params.copiedItemCount));
@@ -1502,11 +1554,7 @@ export class SessionIngestDO extends DurableObject<Env> {
       return { status: 'empty' };
     }
 
-    const sameIdentity =
-      stage.sourceSessionId === params.sourceSessionId &&
-      stage.destinationSessionId === params.destinationSessionId;
-
-    if (!sameIdentity) {
+    if (!isSameCloneStage(stage, params)) {
       return {
         status: 'mismatch',
         storedSourceSessionId: stage.sourceSessionId,
@@ -1540,10 +1588,7 @@ export class SessionIngestDO extends DurableObject<Env> {
       return { status: 'empty' };
     }
 
-    if (
-      stage.sourceSessionId !== params.sourceSessionId ||
-      stage.destinationSessionId !== params.destinationSessionId
-    ) {
+    if (!isSameCloneStage(stage, params)) {
       return {
         status: 'mismatch',
         storedSourceSessionId: stage.sourceSessionId,
diff --git a/services/session-ingest/src/session-ingest-rpc.test.ts b/services/session-ingest/src/session-ingest-rpc.test.ts
index 878046f2884..8fd7c0ecd90 100644
--- a/services/session-ingest/src/session-ingest-rpc.test.ts
+++ b/services/session-ingest/src/session-ingest-rpc.test.ts
@@ -674,17 +674,19 @@ describe('createSessionForCloudAgent clone path', () => {
       kind: 'empty' as 'empty' | 'in_progress' | 'complete',
       sourceSessionId: null as string | null,
       destinationSessionId: null as string | null,
+      upToMessageId: null as string | null,
       nextCursor: null as { ingestedAt: number | null; id: number } | null,
       rollingDigest: '',
       copiedItemCount: 0,
       stagedRows: [] as CloneItem[],
     };
 
-    const inspectCloneStage = vi.fn(() => {
+    const inspectCloneStage = vi.fn((params?: { upToMessageId?: string }) => {
       if (state.kind === 'empty') return { status: 'empty' };
       const sameIdentity =
         state.sourceSessionId === sourceSessionId &&
-        state.destinationSessionId === destinationSessionId;
+        state.destinationSessionId === destinationSessionId &&
+        state.upToMessageId === (params?.upToMessageId ?? null);
       if (!sameIdentity) {
         return {
           status: 'mismatch',
@@ -714,6 +716,7 @@ describe('createSessionForCloudAgent clone path', () => {
       (params: {
         sourceSessionId: string;
         destinationSessionId: string;
+        upToMessageId?: string;
         rows: CloneItem[];
         nextCursor: { ingestedAt: number | null; id: number } | null;
         rollingDigest: string;
@@ -739,7 +742,8 @@ describe('createSessionForCloudAgent clone path', () => {
         if (
           state.kind === 'in_progress' &&
           (state.sourceSessionId !== params.sourceSessionId ||
-            state.destinationSessionId !== params.destinationSessionId)
+            state.destinationSessionId !== params.destinationSessionId ||
+            state.upToMessageId !== (params.upToMessageId ?? null))
         ) {
           return {
             status: 'mismatch',
@@ -750,6 +754,7 @@ describe('createSessionForCloudAgent clone path', () => {
         state.kind = 'in_progress';
         state.sourceSessionId = params.sourceSessionId;
         state.destinationSessionId = params.destinationSessionId;
+        state.upToMessageId = params.upToMessageId ?? null;
         state.nextCursor = params.nextCursor;
         state.rollingDigest = params.rollingDigest;
         state.copiedItemCount = params.copiedItemCount;
@@ -815,6 +820,7 @@ describe('createSessionForCloudAgent clone path', () => {
       state.kind = 'empty';
       state.sourceSessionId = null;
       state.destinationSessionId = null;
+      state.upToMessageId = null;
       state.nextCursor = null;
       state.rollingDigest = '';
       state.copiedItemCount = 0;
@@ -1531,6 +1537,256 @@ describe('createSessionForCloudAgent clone path', () => {
       expect.anything()
     );
   });
+
+  describe('with cloneUpToMessageId', () => {
+    const cutMessage = (id: string, ingestedAt: number, role = 'user'): CloneItem => ({
+      itemId: `message/${id}`,
+      itemType: 'message',
+      itemData: JSON.stringify({
+        id,
+        sessionID: sourceSessionId,
+        role,
+        time: { created: ingestedAt },
+        agent: 'build',
+        model: { providerID: 'openrouter', modelID: 'anthropic/claude-sonnet-4' },
+      }),
+      itemDataR2Key: null,
+      ingestedAt,
+    });
+
+    const cutPart = (messageId: string, partId: string, ingestedAt: number): CloneItem => ({
+      itemId: `${messageId}/${partId}`,
+      itemType: 'part',
+      itemData: JSON.stringify({
+        id: partId,
+        sessionID: sourceSessionId,
+        messageID: messageId,
+        type: 'text',
+        text: `text of ${partId}`,
+      }),
+      itemDataR2Key: null,
+      ingestedAt,
+    });
+
+    const sessionDiffItem = (ingestedAt: number): CloneItem => ({
+      itemId: 'session_diff',
+      itemType: 'session_diff',
+      itemData: JSON.stringify([{ file: 'src/a.ts', additions: 1, deletions: 0 }]),
+      itemDataR2Key: null,
+      ingestedAt,
+    });
+
+    // Ingest order is not message order: a late part of msg_02 lands after msg_03.
+    const interleavedSource = (): CloneItem[] => [
+      sourceSessionItem(),
+      cutMessage('msg_01', 2),
+      cutPart('msg_01', 'prt_01', 3),
+      cutMessage('msg_02', 4, 'assistant'),
+      cutMessage('msg_03', 5),
+      cutPart('msg_02', 'prt_02', 6),
+      cutPart('msg_03', 'prt_03', 7),
+      sessionDiffItem(8),
+    ];
+
+    function makeCutRpc(items: CloneItem[], pageSize: number) {
+      const sourceStub = {
+        ...makePagedSourceStub(items, pageSize),
+        hasMessageItem: vi.fn(async (messageId: string) =>
+          items.some(item => item.itemType === 'message' && item.itemId === `message/${messageId}`)
+        ),
+      };
+      const dest = makeDestinationStub();
+      const r2 = { get: vi.fn(async () => null), put: vi.fn(async () => {}) };
+      const { db, values } = makeCloneDb({
+        sourceRows: [{ sessionId: sourceSessionId, organizationId }],
+        created: createdRow,
+      });
+      const rpc = makeCloneRpc({ db, sourceStub, destStub: dest.stub, r2 });
+      return { rpc, sourceStub, dest, r2, values };
+    }
+
+    it('copies messages up to the inclusive cut with their parts and drops the session diff', async () => {
+      const { rpc, sourceStub, dest, r2 } = makeCutRpc(interleavedSource(), 100);
+
+      await expect(
+        rpc.createSessionForCloudAgent(cloneParams({ cloneUpToMessageId: 'msg_02' }))
+      ).resolves.toEqual({
+        status: 'ready',
+        clone: { sessionId: destinationSessionId, copiedItemCount: 5 },
+      });
+
+      expect(dest.state.stagedRows.map(row => row.itemId)).toEqual([
+        'session',
+        'message/msg_01',
+        'msg_01/prt_01',
+        'message/msg_02',
+        'msg_02/prt_02',
+      ]);
+      expect(dest.state.kind).toBe('complete');
+      expect(dest.state.upToMessageId).toBe('msg_02');
+      expect(dest.stub.inspectCloneStage).toHaveBeenCalledWith({
+        sourceSessionId,
+        destinationSessionId,
+        upToMessageId: 'msg_02',
+      });
+      expect(dest.stub.finalizeCloneStage).toHaveBeenCalledWith(
+        expect.objectContaining({ upToMessageId: 'msg_02', finalItemCount: 5 })
+      );
+      expect(sourceStub.hasMessageItem).toHaveBeenCalledWith('msg_02');
+      expect(r2.get).not.toHaveBeenCalled();
+    });
+
+    it('keeps the full clone, including the session diff, when no cut is given', async () => {
+      const { rpc, sourceStub, dest } = makeCutRpc(interleavedSource(), 100);
+
+      await expect(rpc.createSessionForCloudAgent(cloneParams())).resolves.toEqual({
+        status: 'ready',
+        clone: { sessionId: destinationSessionId, copiedItemCount: 8 },
+      });
+
+      expect(dest.state.stagedRows.map(row => row.itemId)).toEqual(
+        interleavedSource().map(item => item.itemId)
+      );
+      expect(dest.state.upToMessageId).toBeNull();
+      expect(dest.stub.inspectCloneStage.mock.calls[0]?.[0]).toStrictEqual({
+        sourceSessionId,
+        destinationSessionId,
+      });
+      expect(sourceStub.hasMessageItem).not.toHaveBeenCalled();
+    });
+
+    it('rejects clone_cut_not_found before touching the destination when the cut message is absent', async () => {
+      // A part may reference the ID, but only a message row satisfies the cut.
+      const { rpc, sourceStub, dest, values } = makeCutRpc(
+        [...interleavedSource(), cutPart('msg_99', 'prt_99', 9)],
+        100
+      );
+
+      await expect(
+        rpc.createSessionForCloudAgent(cloneParams({ cloneUpToMessageId: 'msg_99' }))
+      ).resolves.toEqual({ status: 'rejected', code: 'clone_cut_not_found' });
+
+      expect(dest.stub.inspectCloneStage).not.toHaveBeenCalled();
+      expect(dest.stub.stageCloneBatch).not.toHaveBeenCalled();
+      expect(dest.stub.resetCloneStage).not.toHaveBeenCalled();
+      expect(sourceStub.exportCloneBatch).not.toHaveBeenCalled();
+      expect(values).not.toHaveBeenCalled();
+    });
+
+    it('resets an unpublished stage built for another cut and restarts from scratch', async () => {
+      const { rpc, dest } = makeCutRpc(interleavedSource(), 100);
+      Object.assign(dest.state, {
+        kind: 'in_progress',
+        sourceSessionId,
+        destinationSessionId,
+        upToMessageId: 'msg_01',
+        nextCursor: { ingestedAt: 3, id: 3 },
+        rollingDigest: 'digest-for-msg_01',
+        copiedItemCount: 3,
+        stagedRows: interleavedSource().slice(0, 3),
+      });
+
+      await expect(
+        rpc.createSessionForCloudAgent(cloneParams({ cloneUpToMessageId: 'msg_02' }))
+      ).resolves.toEqual({
+        status: 'ready',
+        clone: { sessionId: destinationSessionId, copiedItemCount: 5 },
+      });
+
+      expect(dest.stub.resetCloneStage).toHaveBeenCalledTimes(1);
+      expect(dest.state.upToMessageId).toBe('msg_02');
+      expect(dest.state.stagedRows.map(row => row.itemId)).toEqual([
+        'session',
+        'message/msg_01',
+        'msg_01/prt_01',
+        'message/msg_02',
+        'msg_02/prt_02',
+      ]);
+    });
+
+    it('rejects destination_conflict for a published stage built for another cut', async () => {
+      const { rpc, dest } = makeCutRpc(interleavedSource(), 100);
+      Object.assign(dest.state, {
+        kind: 'complete',
+        sourceSessionId,
+        destinationSessionId,
+        upToMessageId: 'msg_01',
+      });
+
+      await expect(
+        rpc.createSessionForCloudAgent(cloneParams({ cloneUpToMessageId: 'msg_02' }))
+      ).resolves.toEqual({ status: 'rejected', code: 'destination_conflict' });
+
+      expect(dest.stub.resetCloneStage).not.toHaveBeenCalled();
+      expect(dest.stub.stageCloneBatch).not.toHaveBeenCalled();
+    });
+
+    it('resumes a cut clone across requests with a digest over the kept rows only', async () => {
+      // A dropped row first, then 20 messages; one row per page so the first
+      // request stops after 16 batches with 15 kept rows staged.
+      const items = [
+        cutPart('msg_19', 'prt_19', 0),
+        ...Array.from({ length: 20 }, (_, index) =>
+          cutMessage(`msg_${String(index).padStart(2, '0')}`, index + 1)
+        ),
+      ];
+      const { rpc, dest } = makeCutRpc(items, 1);
+      const params = cloneParams({ cloneUpToMessageId: 'msg_14' });
+
+      await expect(rpc.createSessionForCloudAgent(params)).resolves.toEqual({
+        status: 'in_progress',
+      });
+      expect(dest.state.copiedItemCount).toBe(15);
+
+      await expect(rpc.createSessionForCloudAgent(params)).resolves.toEqual({
+        status: 'ready',
+        clone: { sessionId: destinationSessionId, copiedItemCount: 15 },
+      });
+      expect(dest.state.kind).toBe('complete');
+      expect(dest.stub.resetCloneStage).not.toHaveBeenCalled();
+    });
+
+    it('resumes after a byte-budget overflow at the right source row when the cut skips rows', async () => {
+      const hugeBody = JSON.stringify({
+        id: 'msg_02',
+        sessionID: sourceSessionId,
+        role: 'assistant',
+        time: { created: 4 },
+        agent: 'build',
+        model: { providerID: 'openrouter', modelID: 'anthropic/claude-sonnet-4' },
+        text: 'x'.repeat(5 * 1024 * 1024),
+      });
+      const items: CloneItem[] = [
+        sourceSessionItem(),
+        cutPart('msg_09', 'prt_09', 1),
+        sessionDiffItem(2),
+        cutMessage('msg_01', 3),
+        { ...cutMessage('msg_02', 4, 'assistant'), itemData: '{}', itemDataR2Key: 'items/huge' },
+        cutMessage('msg_09', 5),
+        cutPart('msg_02', 'prt_02', 6),
+      ];
+      const { rpc, sourceStub, dest, r2 } = makeCutRpc(items, 100);
+      r2.get.mockImplementation((async (key: string) =>
+        key === 'items/huge' ? { text: async () => hugeBody } : null) as never);
+
+      await expect(
+        rpc.createSessionForCloudAgent(cloneParams({ cloneUpToMessageId: 'msg_02' }))
+      ).resolves.toEqual({
+        status: 'ready',
+        clone: { sessionId: destinationSessionId, copiedItemCount: 4 },
+      });
+
+      // The overflow at the huge row re-exports the 4 source rows before it,
+      // including the two skipped rows, so no kept row is staged twice.
+      expect(sourceStub.exportCloneBatch).toHaveBeenCalledWith(null, 4);
+      expect(dest.state.stagedRows.map(row => row.itemId)).toEqual([
+        'session',
+        'message/msg_01',
+        'message/msg_02',
+        'msg_02/prt_02',
+      ]);
+    });
+  });
 });
 
 describe('deleteSessionForCloudAgent clone rollback', () => {
diff --git a/services/session-ingest/src/session-ingest-rpc.ts b/services/session-ingest/src/session-ingest-rpc.ts
index a1692372ee2..a9834375bc9 100644
--- a/services/session-ingest/src/session-ingest-rpc.ts
+++ b/services/session-ingest/src/session-ingest-rpc.ts
@@ -157,7 +157,10 @@ export class SessionIngestRPC extends WorkerEntrypoint<Env> implements SessionIn
    * When `cloneFromKiloSessionId` is absent the legacy empty-storage path runs
    * unchanged and returns a zero-item clone acknowledgement. When present, the
    * source transcript is cloned into the destination Durable Object before the
-   * PostgreSQL insert; the clone is finalized before the row is written.
+   * PostgreSQL insert; the clone is finalized before the row is written. With
+   * `cloneUpToMessageId`, only the transcript up to and including that message
+   * is cloned and the session diff is dropped; an absent cut message rejects
+   * with `clone_cut_not_found`.
    */
   async createSessionForCloudAgent(
     params: CreateSessionForCloudAgentParams
@@ -195,6 +198,7 @@ export class SessionIngestRPC extends WorkerEntrypoint<Env> implements SessionIn
         sourceOrganizationId: source.organizationId,
         destinationOrganizationId: parsed.organizationId ?? null,
         destinationTitle: parsed.title,
+        upToMessageId: parsed.cloneUpToMessageId,
       });
 
       if (outcome.status === 'rejected') return { status: 'rejected', code: outcome.code };

From c76fc53016eb0882e595fb934fafd33a37bf9ac0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:11:44 +0200
Subject: [PATCH 4/6] feat(session-ingest): support upToMessageId on session
 export

---
 services/session-ingest/src/app.ts            |  6 +-
 .../session-ingest/src/routes/api.test.ts     | 71 +++++++++++++++++++
 services/session-ingest/src/routes/api.ts     | 23 ++++--
 .../src/services/session-export.ts            | 38 +++++++---
 .../integration/session-ingest-do.test.ts     | 42 +++++++++++
 5 files changed, 162 insertions(+), 18 deletions(-)

diff --git a/services/session-ingest/src/app.ts b/services/session-ingest/src/app.ts
index 62348451067..7990726d5b3 100644
--- a/services/session-ingest/src/app.ts
+++ b/services/session-ingest/src/app.ts
@@ -143,8 +143,8 @@ app.get('/internal/session/:sessionId/export', requireValidInternalSecret, async
   const parsed = sessionIdSchema.safeParse(c.req.param('sessionId'));
   if (!parsed.success) return c.json({ success: false, error: 'Invalid sessionId' }, 400);
 
-  const stream = await getSessionExport(c.env, parsed.data, kiloUserId);
-  if (stream === null) return c.json({ success: false, error: 'Session not found' }, 404);
+  const result = await getSessionExport(c.env, parsed.data, kiloUserId);
+  if (result.status !== 'ok') return c.json({ success: false, error: 'Session not found' }, 404);
 
-  return c.body(stream, 200, { 'content-type': 'application/json; charset=utf-8' });
+  return c.body(result.stream, 200, { 'content-type': 'application/json; charset=utf-8' });
 });
diff --git a/services/session-ingest/src/routes/api.test.ts b/services/session-ingest/src/routes/api.test.ts
index e43750fa126..d00b4ea8650 100644
--- a/services/session-ingest/src/routes/api.test.ts
+++ b/services/session-ingest/src/routes/api.test.ts
@@ -1259,6 +1259,77 @@ describe('api routes', () => {
     expect(ingestStub.getAllStream).toHaveBeenCalled();
   });
 
+  it('GET /session/:sessionId/export passes an upToMessageId cut to the DO export', async () => {
+    const { db, fns } = makeDbFakes();
+    vi.mocked(getWorkerDb).mockReturnValue(db);
+    fns.selectResult.mockResolvedValueOnce([{ session_id: 'ses_12345678901234567890123456' }]);
+
+    const payload = JSON.stringify({ info: {}, messages: [], sessionDiff: [] });
+    const ingestStub = {
+      hasMessageItem: vi.fn(async () => true),
+      getAllStream: vi.fn(async () => new Response(payload).body!),
+    };
+    vi.mocked(getSessionIngestDO).mockReturnValue(
+      ingestStub as unknown as ReturnType<typeof getSessionIngestDO>
+    );
+
+    const app = makeApiApp();
+    const res = await app.fetch(
+      new Request(
+        'http://local/session/ses_12345678901234567890123456/export?upToMessageId=msg_02',
+        { method: 'GET' }
+      ),
+      makeTestEnv()
+    );
+
+    expect(res.status).toBe(200);
+    expect(await res.text()).toBe(payload);
+    expect(ingestStub.hasMessageItem).toHaveBeenCalledWith('msg_02');
+    expect(ingestStub.getAllStream).toHaveBeenCalledWith('msg_02');
+  });
+
+  it('GET /session/:sessionId/export returns 404 clone_cut_not_found for an absent cut message', async () => {
+    const { db, fns } = makeDbFakes();
+    vi.mocked(getWorkerDb).mockReturnValue(db);
+    fns.selectResult.mockResolvedValueOnce([{ session_id: 'ses_12345678901234567890123456' }]);
+
+    const ingestStub = {
+      hasMessageItem: vi.fn(async () => false),
+      getAllStream: vi.fn(),
+    };
+    vi.mocked(getSessionIngestDO).mockReturnValue(
+      ingestStub as unknown as ReturnType<typeof getSessionIngestDO>
+    );
+
+    const app = makeApiApp();
+    const res = await app.fetch(
+      new Request(
+        'http://local/session/ses_12345678901234567890123456/export?upToMessageId=msg_99',
+        { method: 'GET' }
+      ),
+      makeTestEnv()
+    );
+
+    expect(res.status).toBe(404);
+    expect(await res.json()).toEqual({ success: false, error: 'clone_cut_not_found' });
+    expect(ingestStub.getAllStream).not.toHaveBeenCalled();
+  });
+
+  it('GET /session/:sessionId/export returns 400 for an invalid upToMessageId', async () => {
+    const app = makeApiApp();
+    const res = await app.fetch(
+      new Request(
+        'http://local/session/ses_12345678901234567890123456/export?upToMessageId=not-a-message',
+        { method: 'GET' }
+      ),
+      makeTestEnv()
+    );
+
+    expect(res.status).toBe(400);
+    expect(await res.json()).toMatchObject({ success: false, error: 'Invalid upToMessageId' });
+    expect(getSessionIngestDO).not.toHaveBeenCalled();
+  });
+
   it('GET /session/:sessionId/messages returns 400 for invalid sessionId', async () => {
     const app = makeApiApp();
     const res = await app.fetch(
diff --git a/services/session-ingest/src/routes/api.ts b/services/session-ingest/src/routes/api.ts
index 67bf69a98fc..b195b4671f8 100644
--- a/services/session-ingest/src/routes/api.ts
+++ b/services/session-ingest/src/routes/api.ts
@@ -7,6 +7,7 @@ import {
   DEFAULT_KILO_SDK_MESSAGE_PAGE_SIZE,
   getSessionMessagesSchema,
   MAX_KILO_SDK_MESSAGE_HISTORY_PAGE_SIZE,
+  messageIdSchema,
   persistedKiloSdkMessageHistorySchema,
   sessionIdSchema,
 } from '@kilocode/session-ingest-contracts';
@@ -476,14 +477,28 @@ api.get('/session/:sessionId/export', async c => {
     return c.json({ success: false, error: 'Invalid sessionId', issues: parsed.error.issues }, 400);
   }
 
+  // Optional inclusive cut: export only the transcript up to this message.
+  const rawUpToMessageId = c.req.query('upToMessageId');
+  let upToMessageId: string | undefined;
+  if (rawUpToMessageId !== undefined) {
+    const parsedCut = messageIdSchema.safeParse(rawUpToMessageId);
+    if (!parsedCut.success) {
+      return c.json(
+        { success: false, error: 'Invalid upToMessageId', issues: parsedCut.error.issues },
+        400
+      );
+    }
+    upToMessageId = parsedCut.data;
+  }
+
   const kiloUserId = c.get('user_id');
-  const stream = await getSessionExport(c.env, parsed.data, kiloUserId);
+  const result = await getSessionExport(c.env, parsed.data, kiloUserId, upToMessageId);
 
-  if (stream === null) {
-    return c.json({ success: false, error: 'session_not_found' }, 404);
+  if (result.status !== 'ok') {
+    return c.json({ success: false, error: result.status }, 404);
   }
 
-  return c.body(stream, 200, {
+  return c.body(result.stream, 200, {
     'content-type': 'application/json; charset=utf-8',
   });
 });
diff --git a/services/session-ingest/src/services/session-export.ts b/services/session-ingest/src/services/session-export.ts
index 8eb2b9e10c6..7d5fcee459d 100644
--- a/services/session-ingest/src/services/session-export.ts
+++ b/services/session-ingest/src/services/session-export.ts
@@ -3,31 +3,47 @@ import { getSessionIngestDO } from '../dos/SessionIngestDO';
 import { resolveAccessibleKiloSession } from './session-access';
 import { withDORetry } from '@kilocode/worker-utils';
 
+export type SessionExportResult =
+  | { status: 'ok'; stream: ReadableStream<Uint8Array> }
+  | { status: 'session_not_found' }
+  | { status: 'clone_cut_not_found' };
+
 /**
- * Fetch the full session export as a streaming ReadableStream.
+ * Fetch the session export as a streaming ReadableStream.
  *
  * Verifies that the session belongs to the user and that organization access
- * is current before reading the DO.
- *
- * @returns A ReadableStream of the JSON payload, or `null` if the session
- *          does not exist or does not belong to the user.
+ * is current before reading the DO. With `upToMessageId`, the export holds only
+ * the transcript up to and including that message and an empty `sessionDiff`;
+ * a cut message that is not in the session yields `clone_cut_not_found`.
  */
 export async function getSessionExport(
   env: Env,
   sessionId: string,
-  kiloUserId: string
-): Promise<ReadableStream<Uint8Array> | null> {
+  kiloUserId: string,
+  upToMessageId?: string
+): Promise<SessionExportResult> {
   const accessibleSession = await resolveAccessibleKiloSession(env, {
     kiloUserId,
     kiloSessionId: sessionId,
   });
   if (!accessibleSession) {
-    return null;
+    return { status: 'session_not_found' };
+  }
+
+  const stub = () => getSessionIngestDO(env, { kiloUserId, sessionId });
+  if (upToMessageId !== undefined) {
+    const hasCut = await withDORetry(
+      stub,
+      s => s.hasMessageItem(upToMessageId),
+      'SessionIngestDO.hasMessageItem'
+    );
+    if (!hasCut) return { status: 'clone_cut_not_found' };
   }
 
-  return withDORetry(
-    () => getSessionIngestDO(env, { kiloUserId, sessionId }),
-    stub => stub.getAllStream(),
+  const stream = await withDORetry(
+    stub,
+    s => s.getAllStream(upToMessageId),
     'SessionIngestDO.getAllStream'
   );
+  return { status: 'ok', stream };
 }
diff --git a/services/session-ingest/test/integration/session-ingest-do.test.ts b/services/session-ingest/test/integration/session-ingest-do.test.ts
index e89ea4367cb..4f58109f478 100644
--- a/services/session-ingest/test/integration/session-ingest-do.test.ts
+++ b/services/session-ingest/test/integration/session-ingest-do.test.ts
@@ -219,6 +219,48 @@ describe('SessionIngestDO integration', () => {
       expect(snapshot.messages[0].info.id).toBe('msg_1');
       expect(snapshot.messages[1].info.id).toBe('msg_2');
     });
+
+    it('exports only the transcript up to an inclusive message cut and drops the session diff', async () => {
+      const sessionId = 'ses_roundtrip_cut_0000000005';
+      const stub = getStub(kiloUserId, sessionId);
+
+      // Ingest order is not message order: a late part of msg_2 lands after msg_3.
+      await stub.ingest(
+        [
+          { type: 'session', data: { title: 'Cut Session' } },
+          { type: 'message', data: { id: 'msg_1', role: 'user', content: 'first' } },
+          { type: 'part', data: { id: 'part_1', messageID: 'msg_1', type: 'text' } },
+          { type: 'message', data: { id: 'msg_2', role: 'assistant', content: 'second' } },
+          { type: 'message', data: { id: 'msg_3', role: 'user', content: 'third' } },
+          { type: 'part', data: { id: 'part_2', messageID: 'msg_2', type: 'text' } },
+          { type: 'part', data: { id: 'part_3', messageID: 'msg_3', type: 'text' } },
+          { type: 'session_diff', data: [{ file: 'a.ts', additions: 1, deletions: 0 }] },
+        ],
+        kiloUserId,
+        sessionId,
+        1
+      );
+
+      expect(await stub.hasMessageItem('msg_2')).toBe(true);
+      expect(await stub.hasMessageItem('msg_9')).toBe(false);
+
+      const cut = JSON.parse(await stub.getAllStream('msg_2').then(s => new Response(s).text()));
+      expect(cut.info).toEqual({ title: 'Cut Session' });
+      expect(
+        cut.messages.map((m: { info: { id: string }; parts: Array<{ id: string }> }) => [
+          m.info.id,
+          m.parts.map(p => p.id),
+        ])
+      ).toEqual([
+        ['msg_1', ['part_1']],
+        ['msg_2', ['part_2']],
+      ]);
+      expect(cut.sessionDiff).toEqual([]);
+
+      const full = JSON.parse(await stub.getAllStream().then(s => new Response(s).text()));
+      expect(full.messages).toHaveLength(3);
+      expect(full.sessionDiff).toEqual([{ file: 'a.ts', additions: 1, deletions: 0 }]);
+    });
   });
 
   describe('SDK session snapshot reads', () => {

From c52bdaac8da09df5c1c572f95e459d692a973e95 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:57:12 +0200
Subject: [PATCH 5/6] fix(session-ingest): roll back a clone stage built with a
 message cut

---
 .../src/session-ingest-rpc.test.ts            | 26 +++++++++++++++++++
 .../session-ingest/src/session-ingest-rpc.ts  | 10 ++++++-
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/services/session-ingest/src/session-ingest-rpc.test.ts b/services/session-ingest/src/session-ingest-rpc.test.ts
index 8fd7c0ecd90..660b69b4134 100644
--- a/services/session-ingest/src/session-ingest-rpc.test.ts
+++ b/services/session-ingest/src/session-ingest-rpc.test.ts
@@ -1860,6 +1860,32 @@ describe('deleteSessionForCloudAgent clone rollback', () => {
     expect(deleteFn).toHaveBeenCalled();
   });
 
+  it('rolls back a clone stage built with a message cut for the same source', async () => {
+    // The cut is part of the stage identity, so a cut-less inspect reports
+    // mismatch with the same stored source and destination.
+    const inspectCloneStage = vi.fn(() => ({
+      status: 'mismatch',
+      storedSourceSessionId: sourceSessionId,
+      storedDestinationSessionId: destinationSessionId,
+      storedStage: 'complete',
+    }));
+    const resetCloneStage = vi.fn(async () => {});
+    const { db, deleteFn } = makeDeleteDb({
+      session_id: destinationSessionId,
+      kilo_user_id: kiloUserId,
+    });
+    const rpc = makeDeleteRpc({ db, destStub: { inspectCloneStage, resetCloneStage } });
+
+    await rpc.deleteSessionForCloudAgent({
+      sessionId: destinationSessionId,
+      kiloUserId,
+      cloneSourceSessionId: sourceSessionId,
+    });
+
+    expect(resetCloneStage).toHaveBeenCalled();
+    expect(deleteFn).toHaveBeenCalled();
+  });
+
   it('never clears a destination clone that belongs to another source', async () => {
     const inspectCloneStage = vi.fn(() => ({
       status: 'mismatch',
diff --git a/services/session-ingest/src/session-ingest-rpc.ts b/services/session-ingest/src/session-ingest-rpc.ts
index a9834375bc9..29683a6d26e 100644
--- a/services/session-ingest/src/session-ingest-rpc.ts
+++ b/services/session-ingest/src/session-ingest-rpc.ts
@@ -668,7 +668,15 @@ export class SessionIngestRPC extends WorkerEntrypoint<Env> implements SessionIn
         'SessionIngestDO.inspectCloneStage'
       );
 
-      const matches = inspected.status === 'complete' || inspected.status === 'in_progress';
+      // Rollback owns the destination by source and destination only. A stage
+      // built with a message cut reports `mismatch` for a cut-less inspect, yet
+      // it is still this source's clone and must be rolled back.
+      const matches =
+        inspected.status === 'complete' ||
+        inspected.status === 'in_progress' ||
+        (inspected.status === 'mismatch' &&
+          inspected.storedSourceSessionId === cloneSourceSessionId &&
+          inspected.storedDestinationSessionId === parsed.sessionId);
       if (!matches) return;
 
       await this.deleteRootSessionRow(parsed);

From 3ac74ff7130c1bc614db337d7a27f4371044e458 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <iscekic@protonmail.com>
Date: Mon, 12 Oct 2026 19:57:13 +0200
Subject: [PATCH 6/6] fix(session-ingest): clear the clone cut key when staging
 a full clone

---
 .../src/dos/SessionIngestDO.test.ts             | 17 +++++++++++++----
 .../session-ingest/src/dos/SessionIngestDO.ts   |  6 ++++--
 2 files changed, 17 insertions(+), 6 deletions(-)

diff --git a/services/session-ingest/src/dos/SessionIngestDO.test.ts b/services/session-ingest/src/dos/SessionIngestDO.test.ts
index 0228e5a26ed..cd48cfe18c1 100644
--- a/services/session-ingest/src/dos/SessionIngestDO.test.ts
+++ b/services/session-ingest/src/dos/SessionIngestDO.test.ts
@@ -1579,10 +1579,17 @@ describe('SessionIngestDO clone primitives', () => {
           if (table === ingestItems) items.length = 0;
           if (table === ingestMeta) meta.clear();
         },
-        where: () => ({
+        where: (condition: unknown) => ({
           run: () => {
-            if (table === ingestMeta) {
-              for (const key of meta.keys()) if (key !== 'deleted') meta.delete(key);
+            if (table !== ingestMeta) return;
+            // `ne(key, 'deleted')` wipes all but the tombstone; `eq(key, k)` deletes one key.
+            const key = collectBoundParams(condition).find(
+              (p): p is string => typeof p === 'string'
+            );
+            if (key === 'deleted') {
+              for (const k of meta.keys()) if (k !== 'deleted') meta.delete(k);
+            } else if (key !== undefined) {
+              meta.delete(key);
             }
           },
         }),
@@ -2129,8 +2136,10 @@ describe('SessionIngestDO clone primitives', () => {
     ).toMatchObject({ status: 'mismatch', storedStage: 'complete' });
   });
 
-  it('records no message cut for a full clone stage', () => {
+  it('records no message cut for a full clone stage and clears a stale one', () => {
     const harness = makeCloneHarness();
+    // A leftover cut with no stage identity reads as an empty stage.
+    harness.meta.set('cloneUpToMessageId', 'msg_stale');
     harness.durableObject.stageCloneBatch({
       sourceSessionId: 'src',
       destinationSessionId: 'dst',
diff --git a/services/session-ingest/src/dos/SessionIngestDO.ts b/services/session-ingest/src/dos/SessionIngestDO.ts
index 3329039c712..b06be96a994 100644
--- a/services/session-ingest/src/dos/SessionIngestDO.ts
+++ b/services/session-ingest/src/dos/SessionIngestDO.ts
@@ -1528,10 +1528,12 @@ export class SessionIngestDO extends DurableObject<Env> {
 
     writeCloneMeta(this.db, 'cloneSourceSessionId', params.sourceSessionId);
     writeCloneMeta(this.db, 'cloneDestinationSessionId', params.destinationSessionId);
-    // Only cut clones record a cut: an empty or same-identity stage never holds
-    // this key for a full clone, so full-clone state is unchanged.
+    // Keep "no cut key <=> full clone" true on every write, without relying on
+    // resets wiping the whole meta table.
     if (params.upToMessageId !== undefined) {
       writeCloneMeta(this.db, 'cloneUpToMessageId', params.upToMessageId);
+    } else {
+      this.db.delete(ingestMeta).where(eq(ingestMeta.key, 'cloneUpToMessageId')).run();
     }
     writeCloneMeta(this.db, 'cloneNextCursor', JSON.stringify(params.nextCursor));
     writeCloneMeta(this.db, 'cloneRollingDigest', params.rollingDigest);