Anson-Se · Oct 24, 2026 · Oct 24, 2026 · Oct 24, 2026 · Oct 24, 2026 · Oct 24, 2026
@@ -20,6 +20,7 @@ export const MEMORY_WORKER_JOB_TYPES = [
   "l2_induction",
   "l3_abstraction",
   "l3_world_model_update",
+  "l3_unit_update",
   "project_environment_profile",
   "skill_crystallization",
   "skill_cluster_assign",

@@ -83,6 +83,51 @@ export const L3WorldModelBoundaryResponseSchema = z.object({
 }).strict();
 export type L3WorldModelBoundaryResponse = z.infer<typeof L3WorldModelBoundaryResponseSchema>;
 
+const AbsolutePathSchema = z.string().min(1).max(4096)
+  .refine((value) => value.startsWith("/") || /^[A-Za-z]:[\\/]/.test(value), "path must be absolute");
+
+/** Full list of folders the user picked in the desktop app; each push replaces the previous one. */
+export const L3UserProjectsReplaceRequestSchema = z.object({
+  ...L3WorldModelRequestEnvelopeShape,
+  projects: z.array(z.object({
+    rootPath: AbsolutePathSchema,
+    name: z.string().min(1).max(512)
+  }).strict()).max(500),
+  standaloneRoots: z.array(AbsolutePathSchema).max(20).default([])
+}).strict().superRefine(assertEnvelopeSourceConsistency);
+export type L3UserProjectsReplaceRequest = z.input<typeof L3UserProjectsReplaceRequestSchema>;
+
+export const L3UserProjectsReplaceResponseSchema = z.object({
+  count: z.number().int().nonnegative(),
+  serverTime: z.string().datetime()
+}).strict();
+export type L3UserProjectsReplaceResponse = z.infer<typeof L3UserProjectsReplaceResponseSchema>;
+
+/** Puts the user's new-L3 batches that ended in dead letter back in the queue. */
+export const L3FailedBatchesRetryRequestSchema = z.object({
+  ...L3WorldModelRequestEnvelopeShape
+}).strict().superRefine(assertEnvelopeSourceConsistency);
+export type L3FailedBatchesRetryRequest = z.input<typeof L3FailedBatchesRetryRequestSchema>;
+
+export const L3FailedBatchesRetryResponseSchema = z.object({
+  requeued: z.number().int().nonnegative(),
+  serverTime: z.string().datetime()
+}).strict();
+export type L3FailedBatchesRetryResponse = z.infer<typeof L3FailedBatchesRetryResponseSchema>;
+
+/** End of a hosted import: freeze what it added to these scanned Sessions into L3 batches. */
+export const L3ScanSessionsFreezeRequestSchema = z.object({
+  ...L3WorldModelRequestEnvelopeShape,
+  sessionIds: z.array(NonEmptyStringSchema).max(10_000)
+}).strict().superRefine(assertEnvelopeSourceConsistency);
+export type L3ScanSessionsFreezeRequest = z.input<typeof L3ScanSessionsFreezeRequestSchema>;
+
+export const L3ScanSessionsFreezeResponseSchema = z.object({
+  batches: z.number().int().nonnegative(),
+  serverTime: z.string().datetime()
+}).strict();
+export type L3ScanSessionsFreezeResponse = z.infer<typeof L3ScanSessionsFreezeResponseSchema>;
+
 export const SessionL3WorldModelContextResponseSchema = z.object({
   schemaVersion: z.literal(2),
   projectId: NonEmptyStringSchema.nullable(),

@@ -54,6 +54,7 @@ export const JobTypeSchema = z.enum([
   "l2_induction",
   "l3_abstraction",
   "l3_world_model_update",
+  "l3_unit_update",
   "project_environment_profile",
   "skill_crystallization",
   "skill_trial_resolve",

@@ -5,6 +5,7 @@ import {
   CloseSessionOutputSchema,
   CompleteTurnOutputSchema,
   SourceTurnCompleteOutputSchema,
+  L3ScanSessionsFreezeResponseSchema,
   DeleteMemoryOutputSchema,
   EmbeddingInferenceOutputSchema,
   DeletePanelTaskOutputSchema,
@@ -26,6 +27,7 @@ import {
   RetryMemoryProcessingOutputSchema,
   WorkerRunOutputSchema
 } from "@memmy/local-api-contracts";
+import { randomUUID } from "node:crypto";
 import { z, type ZodType } from "zod";
 import { MemoryLayerError, MemoryLayerNetworkError } from "./errors.js";
 import { buildMemoryLayerUrl, MEMORY_LAYER_PATHS } from "./memory-layer-endpoints.js";
@@ -186,6 +188,12 @@ export function createHttpMemoryClient(
       return request("POST", "completeSourceTurn", SourceTurnCompleteOutputSchema, { body: input, context });
     },
 
+    async freezeScanSessions(source, sessionIds, context) {
+      // The L3 envelope resolves the owner the same way the source turns that opened these Sessions did.
+      const body = { requestId: randomUUID(), adapterId: `agent-source:${source}`, source, sessionIds };
+      return request("POST", "freezeScanSessions", L3ScanSessionsFreezeResponseSchema, { body, context });
+    },
+
     async search(input, context) {
       return request("POST", "search", SearchOutputSchema, { body: input, context });
     },

@@ -11,6 +11,7 @@ export const MEMORY_LAYER_PATHS = Object.freeze({
   startTurn: "/api/v1/turns/start",
   completeTurn: "/api/v1/turns/:turnId/complete",
   completeSourceTurn: "/api/v1/source-turns/complete",
+  freezeScanSessions: "/api/v1/l3/scan-sessions/freeze",
   search: "/api/v1/memory/search",
   embeddingInference: "/api/v1/models/embedding/infer",
   addMemory: "/api/v1/memory/add",

@@ -30,6 +30,20 @@ describe("HttpMemoryClient", () => {
     expect(JSON.parse(String(calls[0]?.init?.body))).toEqual(input);
     expect(new Headers(calls[0]?.init?.headers).get("x-memmy-user-id")).toBe("fixture-user");
   });
+  it("hands scanned Sessions to L3 inside a full request envelope for their source", async () => {
+    const calls: Array<{ url: string; init?: RequestInit }> = [];
+    const client = createHttpMemoryClient({ baseUrl: "http://memory.test", token: "fixture-token", timeoutMs: 500, maxRetries: 0 }, {
+      fetchImpl: (async (url, init) => {
+        calls.push({ url: String(url), init });
+        return new Response(JSON.stringify({ batches: 2, serverTime: "2099-01-01T00:00:00.000Z" }), { status: 200, headers: { "content-type": "application/json" } });
+      }) as typeof fetch
+    });
+    expect(await client.freezeScanSessions("codex", ["s-1", "s-2"])).toEqual({ batches: 2, serverTime: "2099-01-01T00:00:00.000Z" });
+    expect(calls[0]?.url).toBe("http://memory.test/api/v1/l3/scan-sessions/freeze");
+    expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({
+      requestId: expect.stringMatching(/^[0-9a-f-]{36}$/), adapterId: "agent-source:codex", source: "codex", sessionIds: ["s-1", "s-2"]
+    });
+  });
   it("only defines path templates for the final memory HTTP APIs", () => {
     expect(Object.values(MEMORY_LAYER_PATHS)).toEqual([
       "/api/v1/health",
@@ -42,6 +56,7 @@ describe("HttpMemoryClient", () => {
       "/api/v1/turns/start",
       "/api/v1/turns/:turnId/complete",
       "/api/v1/source-turns/complete",
+      "/api/v1/l3/scan-sessions/freeze",
       "/api/v1/memory/search",
       "/api/v1/models/embedding/infer",
       "/api/v1/memory/add",

@@ -86,6 +86,7 @@ describe("createMockMemoryClient", () => {
       "completeSourceTurn",
       "deleteMemory",
       "enqueueImportSummaries",
+      "freezeScanSessions",
       "getMemory",
       "getMemoryProcessingStatus",
       "getMemoryTokenBudget",

@@ -74,6 +74,8 @@ export interface MemoryClient {
   completeTurn(input: CompleteTurnInput & { turnId: string }, context?: MemoryRequestContext): Promise<CompleteTurnOutput>;
 
   completeSourceTurn(input: SourceTurnCompleteInput, context?: MemoryRequestContext): Promise<SourceTurnCompleteOutput>;
+  /** Freeze what an import added to these scanned Sessions into L3 batches. */
+  freezeScanSessions(source: string, sessionIds: string[], context?: MemoryRequestContext): Promise<{ batches: number }>;
 
   search(input: SearchInput, context?: MemoryRequestContext): Promise<SearchOutput>;
   embeddingInference?(input: EmbeddingInferenceInput, options?: { signal?: AbortSignal }): Promise<EmbeddingInferenceOutput>;

@@ -132,7 +132,7 @@ export interface CreateAgentSourceServiceOptions {
   sourceRegistry: SourceRegistry;
   agentSourceRepository: AgentSourceRepository;
   ingestionService: IngestionService;
-  memoryClient: Pick<MemoryClient, "addMemory" | "completeSourceTurn" | "enqueueImportSummaries" | "getMemoryProcessingStatus" | "runWorker">;
+  memoryClient: Pick<MemoryClient, "addMemory" | "completeSourceTurn" | "enqueueImportSummaries" | "getMemoryProcessingStatus" | "runWorker"> & Partial<Pick<MemoryClient, "freezeScanSessions">>;
   skillDistributionService: SkillDistributionService;
   agentSourceAnalytics?: AgentSourceLifecycleAnalytics;
   memoryAddAnalytics?: Pick<
@@ -741,7 +741,20 @@ async function ingestPersistentStagedSource(
   store.saveMeta({ jobId: store.getMeta()?.jobId ?? scanOptions.scanJobId ?? "", sourceId: store.getMeta()?.sourceId ?? sourceId, mode, phase: "ingest", createdAt: store.getMeta()?.createdAt ?? now(), updatedAt: now() });
   const preparedState = store.getSourceState(sourceId);
   store.saveSourceState({ ...(preparedState ?? { sourceId, mode, messageCount: store.count(sourceId), resultCount: store.resultCount(sourceId), errorCount: stage.scanErrorCount, updatedAt: now() }), phase: "ingest", updatedAt: now() });
-  const ingestion = await ingestPersistentSource(options, store, sourceId, scanOptions, []);
+  const sessionIds = new Set<string>();
+  let ingestion: Awaited<ReturnType<typeof ingestPersistentSource>>;
+  try {
+    ingestion = await ingestPersistentSource(options, store, sourceId, scanOptions, [], sessionIds);
+  } finally {
+    // Scanned history never ends its Sessions, so hand what this pass added to L3 now,
+    // including a pass that was cancelled midway. Freezing only writes rows in Memory;
+    // a failure here must not fail the scan.
+    if (sessionIds.size > 0 && options.memoryClient.freezeScanSessions) {
+      await options.memoryClient.freezeScanSessions(sourceId, [...sessionIds]).catch((error: unknown) => {
+        console.warn(`[agent-source] l3.freeze_failed ${error instanceof Error ? error.message : String(error)}`);
+      });
+    }
+  }
   const scannedAt = now();
   options.agentSourceRepository.setLastScannedAt(sourceId, scannedAt);
   const skillResult = await ingestSourceSkills(options, sourceId, scanOptions, store, mode);
@@ -851,7 +864,8 @@ async function ingestPersistentSource(
   store: AppAgentSourceScanStore,
   sourceId: string,
   scanOptions: AgentSourceScanOptions,
-  initialErrors: readonly { conversationId: string; reason: string }[]
+  initialErrors: readonly { conversationId: string; reason: string }[],
+  sessionIds: Set<string>
 ): Promise<{ memoryIds: string[]; memoryIdCount: number; deduped: number; errorCount: number; errors: Array<{ conversationId: string; reason: string }>; latestSeenAt: string | null; retryFrom: string | null; hasUncommittedSkips: boolean }> {
   const memoryIds: string[] = [];
   let memoryIdCount = 0;
@@ -970,6 +984,7 @@ async function ingestPersistentSource(
           emitAddProgress("Capturing conversation turns");
           continue;
         }
+        if (result.result?.sessionId) sessionIds.add(result.result.sessionId);
         const ids = result.result?.l1MemoryIds ?? [];
         if (result.status === "stored") {
           memoryIdCount += ids.length;

@@ -56,6 +56,8 @@ export interface IngestionStats {
   dedupedMemories: number;
   failedMemories: number;
   memoryIds: string[];
+  /** Memory Sessions this import added turns to; frozen into L3 when the import ends. */
+  sessionIds?: string[];
   /**
    * Subset of `memoryIds` written through the legacy add-memory path, which still needs a
    * summary job. A native turn is summarized inside the Memory capture transaction, so its
@@ -77,7 +79,7 @@ export interface IngestionItemSkip {
 
 /** Contract for create ingestion service options. */
 export interface CreateIngestionServiceOptions {
-  memoryClient: Pick<MemoryClient, "addMemory" | "completeSourceTurn">;
+  memoryClient: Pick<MemoryClient, "addMemory" | "completeSourceTurn"> & Partial<Pick<MemoryClient, "freezeScanSessions">>;
   agentSourceRepository: Pick<AgentSourceRepository, "hasSeen" | "markSeen">;
   memoryAddAnalytics?: Pick<
     MemoryDesktopAddAnalytics,
@@ -117,20 +119,30 @@ export function createIngestionService(options: CreateIngestionServiceOptions):
       };
 
       const conversations = groupByConversation(messages, stats)[Symbol.asyncIterator]();
-      while (true) {
-        if (ctx.signal?.aborted) {
-          break;
+      try {
+        while (true) {
+          if (ctx.signal?.aborted) {
+            break;
+          }
+
+          const next = await conversations.next();
+          if (next.done) {
+            break;
+          }
+
+          stats.conversations += 1;
+          await processConversation(options, next.value, ctx, stats);
         }
-
-        const next = await conversations.next();
-        if (next.done) {
-          break;
+      } finally {
+        // Imported history never ends its Sessions, so hand what this import added to L3 now, also
+        // when it stopped early. Freezing only writes rows in Memory; a failure must not fail the import.
+        const sessionIds = [...new Set(stats.sessionIds ?? [])];
+        if (sessionIds.length > 0 && options.memoryClient.freezeScanSessions) {
+          await options.memoryClient.freezeScanSessions(ctx.sourceId, sessionIds).catch((error: unknown) => {
+            console.warn(`[agent-source] l3.freeze_failed ${error instanceof Error ? error.message : String(error)}`);
+          });
         }
-
-        stats.conversations += 1;
-        await processConversation(options, next.value, ctx, stats);
       }
-
       return stats;
     }
   };
@@ -333,6 +345,8 @@ async function processNativeConversation(
         ...(legacyImportTurnId ? { legacyImportTurnId } : {}),
         ...(ctx.scanMode === "initial_subset" || ctx.scanMode === "full" ? { captureLegacyHistory: true } : {})
       });
+      // Stored and already-stored turns both count: a re-import must freeze what a failed freeze left.
+      if (result.result?.sessionId) (stats.sessionIds ??= []).push(result.result.sessionId);
       if (result.status === "pending" || result.status === "conflict") {
         failed = true;
         stats.failed += turn.messages.length;

@@ -47,6 +47,41 @@ describe("native Codex ingestion", () => {
     expect(markSeen).toHaveBeenCalledTimes(2);
   });
 
+  it("freezes the sessions an import touched once it ends, and a failed freeze does not fail the import", async () => {
+    const completeSourceTurn = vi.fn().mockResolvedValue({ status: "stored", result: { sessionId: "s-1", l1MemoryIds: ["l1-native"] } });
+    const freezeScanSessions = vi.fn().mockResolvedValue({ batches: 1 });
+    await createService({ completeSourceTurn, freezeScanSessions }).ingest(toAsyncIterable(nativeMessages()), { sourceId: "codex" });
+    expect(freezeScanSessions).toHaveBeenCalledWith("codex", ["s-1"]);
+
+    const failing = vi.fn().mockRejectedValue(new Error("memory unavailable"));
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const stats = await createService({ completeSourceTurn, freezeScanSessions: failing })
+      .ingest(toAsyncIterable(nativeMessages()), { sourceId: "codex" });
+    expect(stats.memoryIds).toEqual(["l1-native"]);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("l3.freeze_failed"));
+    warn.mockRestore();
+
+    // Importing the same history again returns "existing"; the freeze is asked again so it can catch up.
+    const again = vi.fn().mockResolvedValue({ batches: 1 });
+    const existing = vi.fn().mockResolvedValue({ status: "existing", result: { sessionId: "s-1", l1MemoryIds: ["l1-native"] } });
+    await createService({ completeSourceTurn: existing, freezeScanSessions: again }).ingest(toAsyncIterable(nativeMessages()), { sourceId: "codex" });
+    expect(again).toHaveBeenCalledWith("codex", ["s-1"]);
+  });
+
+  it("still freezes what an import added when the import stops with an error", async () => {
+    const completeSourceTurn = vi.fn().mockResolvedValue({ status: "stored", result: { sessionId: "s-1", l1MemoryIds: ["l1-native"] } });
+    const freezeScanSessions = vi.fn().mockResolvedValue({ batches: 1 });
+    async function* brokenSource() {
+      yield* nativeMessages();
+      // A later conversation fails to read after the first one was stored.
+      yield { ...nativeMessages()[0]!, conversationId: "next-conversation", messageId: "next-0" };
+      throw new Error("source read failed");
+    }
+    await expect(createService({ completeSourceTurn, freezeScanSessions }).ingest(brokenSource(), { sourceId: "codex" }))
+      .rejects.toThrow("source read failed");
+    expect(freezeScanSessions).toHaveBeenCalledWith("codex", ["s-1"]);
+  });
+
   it("emits add analytics only when a native scan turn is newly stored", async () => {
     const events: Array<{ name: string; payload: Record<string, unknown> }> = [];
     const analytics = {

@@ -93,6 +93,44 @@ describe("persistent Codex scan", () => {
     expect(repository.getScanWatermark("codex")).toMatchObject({ latestSeenCreatedAt: at });
   });
 
+  it("hands the sessions a scan pass touched to L3 once, and a failed hand-off does not fail the scan", async () => {
+    const root = mkdtempSync(join(tmpdir(), "native-backend-scan-")); roots.push(root);
+    const store = createAppStateStore({ databasePath: join(root, "app.sqlite") }); stores.push(store);
+    const repository = store.repositories.agentSources;
+    const at = "2099-09-09T10:00:00.000Z";
+    const event = (type: string, payload: Record<string, unknown>) => ({ type, timestamp: at, payload });
+    const turn = (turnId: string, text: string) => [
+      event("event_msg", { type: "task_started", turn_id: turnId }),
+      event("response_item", { type: "message", role: "user", content: [{ text }] }),
+      event("response_item", { type: "message", role: "assistant", content: [{ text: "Done." }] }),
+      event("event_msg", { type: "task_complete", turn_id: turnId })
+    ];
+    writeFileSync(join(root, "rollout-a.jsonl"), [event("session_meta", { id: "session-a" }), ...turn("turn-1", "First."), ...turn("turn-2", "Second.")]
+      .map((value) => JSON.stringify(value)).join("\n") + "\n");
+    const client = createMockMemoryClient();
+    vi.spyOn(client, "completeSourceTurn").mockImplementation(async (input) => ({ status: "stored", result: {
+      turnId: input.sourceTurn.turnId, sessionId: "memory-session", episodeId: "episode", rawTurnId: "raw", l1MemoryId: input.sourceTurn.turnId,
+      l1MemoryIds: [input.sourceTurn.turnId], closedEpisodeIds: [], scheduledEvolution: false, jobs: [], serverTime: at
+    } }));
+    const freeze = vi.spyOn(client, "freezeScanSessions").mockRejectedValue(new Error("memory offline"));
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const service = createAgentSourceService({
+      sourceRegistry: createSourceRegistry([createCodexSourceAdapter({ sessionsRoot: root })]),
+      memoryClient: client,
+      agentSourceRepository: repository,
+      ingestionService: createIngestionService({ memoryClient: client, agentSourceRepository: repository }),
+      skillDistributionService: { install: async () => undefined, uninstall: async () => undefined, installPlugin: async () => undefined, uninstallPlugin: async () => undefined },
+      scanStoreDirectory: join(root, "scans")
+    });
+    const result = await service.scanOne("codex", { scanJobId: "freeze-job", mode: "full" });
+    expect(result.errors).toEqual([]);
+    expect(freeze).toHaveBeenCalledOnce();
+    expect(freeze).toHaveBeenCalledWith("codex", ["memory-session"]);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("l3.freeze_failed memory offline"));
+    expect(repository.getConversationCheckpoint("codex", "session-a")).not.toBeNull();
+    warn.mockRestore();
+  });
+
   it("starts the first watermark at a recent unfinished turn and captures it after it completes in place", async () => {
     const root = mkdtempSync(join(tmpdir(), "native-backend-scan-")); roots.push(root);
     const store = createAppStateStore({ databasePath: join(root, "app.sqlite") }); stores.push(store);

@@ -141,6 +141,11 @@ export function createMockMemoryClient(options: CreateMockMemoryClientOptions =
       };
     },
 
+    async freezeScanSessions() {
+      failIfNeeded();
+      return { batches: 0 };
+    },
+
     async completeSourceTurn(input) {
       failIfNeeded();
       const l1MemoryId = randomUUID();