@@ -212,15 +212,15 @@ With sync off, learn makes no network requests for lessons and delivers no team
 - each lesson's text, tags, path triggers, pin, provenance, and coexistence declarations;
 - how often each team lesson was delivered, and helpful or harmful marks the curator gave it.
 
-Everything you send lands as a proposal. It reaches the team only after the workspace owner approves it, and that includes lessons promoted automatically on your machine. The first sync also uploads the lessons the project already has: the candidate if there is one, otherwise the approved set, at most 200 per run, starting with the most helpful. A lesson the candidate removed counts as an intentional removal and is not uploaded.
+Everything you send lands as a proposal. It reaches the team only after the workspace owner approves it, and that includes lessons promoted automatically on your machine. The first sync also uploads the lessons the project already has: the candidate if there is one, otherwise the approved set, at most 200 per run, starting with the most helpful. A lesson the candidate removed counts as an intentional removal and is not uploaded. The same upload runs again after sync was off or the pulled lessons were deleted. It skips a lesson the workspace already holds as it is now (your open proposal, an earlier submission, or the team version) and proposes one you changed since: a revision of your open proposal, or a new proposal. A lesson the team has a version of, or retired, is not proposed by this upload; edit it to propose a change. A lesson the workspace would refuse is not sent, and `learn status` lists it as not syncable: text over 140 characters, more than 32 tags or a tag over 64 characters, more than 32 path triggers, more than 100 coexistence declarations, or more than 50 replaced lessons. Provenance over 4,000 characters is shortened.
 
 **Editing team lessons.** Reflection sees team lessons but never writes them into `candidate.json` or `approved.json`. Editing, removing, or superseding a team lesson becomes a proposal that names the approved version it replaces. The current lesson stays live until the owner approves the change. Proposals are queued in `outbox.json` and sent after reflection and promotion, at session start, and by `learn sync` and `learn push`.
 
 **Conflicts.** If someone else changed a lesson on the server first, for example the owner edited your proposal or approved another edit, your proposal is held instead of being sent again over their change. If the server already holds what you proposed, the held proposal is dropped. Otherwise `learn status` lists it, and you resend it after review with `learn push --resubmit <lesson key>`. A proposal the owner rejected is never sent again.
 
 **Retirement.** When the owner retires a lesson, the next pull hides it in every project, including a copy restored by a Git checkout or `learn rollback`. A running session drops it from its "Team rules" section and from later requests, but text the agent has already received is not rewritten. Rolling back locally changes only your local lessons. If the workspace still serves a playbook skill published earlier with `learn promote --publish`, its lessons keep being delivered until the owner detaches that skill in the workspace's Skills page; `learn sync` warns about it. With sync on, `promote --publish` is refused.
 
-**Accounts and links.** The pulled lessons and the queue belong to one account, workspace, and repository. After you switch accounts, relink, or unlink, the old cache is no longer used. Queued proposals from the old scope are held and are never sent to the new one. `learn disable --sync` turns sync off and deletes the pulled lessons, and leaves capture and local lessons as they are. If the server does not support lesson sync yet, learn retries once a day.
+**Accounts and links.** The pulled lessons and the queue belong to one account, workspace, and repository. After you switch accounts, relink, or unlink, the old cache is no longer used. Queued proposals from the old scope are held and are never sent to the new one. `learn disable --sync` turns sync off and deletes the pulled lessons, and leaves capture and local lessons as they are. When `ALTIMATE_LEARN_SYNC` or a higher-precedence config keeps sync on, it reports that and keeps the pulled lessons. If the server does not support lesson sync yet, or not for your tenant, learn retries once a day.
 
 ## Learn vs memory vs a knowledge base
 
@@ -273,7 +273,7 @@ altimate-code learn status --json
 
 ### show
 
-Prints approved lessons, the candidate, a diff between them, and pending replacements. Each lesson shows its id, text, and counters (`helpful`, `harmful`, `applied`). Flagged lessons print a `WARNING` line. Lessons over 140 characters from older stores are marked `long`. With lesson sync on, lessons are marked `(team)` (approved in the workspace), `(pending review)` (your proposal is waiting for the owner) or `(local)`, and a "Team" section lists the team lessons.
+Prints approved lessons, the candidate, a diff between them, and pending replacements. Each lesson shows its id, text, and counters (`helpful`, `harmful`, `applied`). Flagged lessons print a `WARNING` line. Lessons over 140 characters from older stores are marked `long`. With lesson sync on, lessons are marked `(team)` (approved in the workspace), `(pending review)` (your proposal is waiting for the owner or to be sent), `(held — run learn push --resubmit <key>)` (your proposal conflicted with a change on the server) or `(local)`, and a "Team" section lists the team lessons.
 
 ```bash
 altimate-code learn show
@@ -383,7 +383,7 @@ altimate-code learn reject
 
 ### sync / push
 
-`sync` pulls the approved team lessons and sends queued proposals; `push` only sends. Both need [lesson sync](#sync-with-your-workspace) to be on. `sync` reports each store's pull, what was sent, and warns when the workspace still serves a learn-managed playbook skill.
+`sync` pulls the approved team lessons and sends queued proposals; `push` only sends. Both need [lesson sync](#sync-with-your-workspace) to be on. `sync` reports each store's pull, what was sent, and warns when the workspace still serves a learn-managed playbook skill. Both exit non-zero when sync was skipped, a pull failed, or a proposal was not sent; with `--json` the report is printed first.
 
 ```bash
 altimate-code learn sync
 
@@ -56,7 +56,8 @@ there is no network traffic and no team lesson is delivered.
   requests use the credential captured for the scope, and a result whose scope changed in flight is discarded.
   A route the server lacks (404 without a code) backs off for 24 hours.
 - `ledger.ts` owns the private per-store files `remote.json`, `outbox.json` and `sync.json` (ignored by the learn
-  `.gitignore`, written atomically with mode 0600).
+  `.gitignore`, written atomically with mode 0600). A malformed outbox is never read as empty: its valid entries are
+  kept and the file is set aside as `outbox.json.corrupt-<ms>` before it is replaced.
 - `effective.ts` overlays team lessons on local ones by `(repo_identity, store, lesson_key)`: remote wins, an
   exact-identity tombstone hides the local copy. `Store.loadApproved()` stays local-only.
 - `proposals.ts` turns local changes into proposals by the ledger (add, revision of your open proposal, or edit/remove
 
@@ -16,7 +16,7 @@ import { assertLearnLock } from "./lock"
 import { errText } from "./session-reflect"
 import type { AutoPromoteLimits } from "./config"
 import { readRemote, type Scope } from "./ledger"
-import { recordPromotion } from "./proposals"
+import { promotionHooks } from "./proposals"
 import { Log } from "@/util/log"
 
 const log = Log.create({ service: "learn.auto-promote" })
@@ -332,9 +332,7 @@ export async function autoPromote(input: AutoPromoteInput): Promise<AutoPromoteR
           publish: publishText,
           keepCandidate,
           onPublished: () => { published = true },
-          ...(input.sync ? { beforePublish: async (before: Lessons.Lesson[], after: Lessons.Lesson[]) => {
-            await recordPromotion(root, name, input.sync!.scope, before, after, "auto_promote")
-          } } : {}),
+          ...(input.sync ? promotionHooks(root, name, input.sync.scope, "auto_promote") : {}),
           // Every flagged lesson left in the candidate is unchanged from the approved set (checked above).
           allowFlagged: true,
           history: {
 
@@ -13,7 +13,7 @@ import { validateName } from "./playbook"
 import { estimateTokens, fileHookEnabled, lessonLine, renderSection, resolveLimits, retrieve, selectFile, selectStart, type Limits } from "./select"
 import * as Store from "./store"
 import { effectiveLessons, resolve, type Source } from "./effective"
-import { identityKey, readOutbox, readRemote, remoteFor, Scope, sameScope, scopeKey, writeOutbox, type Identity } from "./ledger"
+import { identityKey, readOutbox, readRemote, remoteFor, Scope, sameScope, scopeKey, usageBatches, uuidFrom, writeOutbox, type Identity } from "./ledger"
 import type { RemoteView } from "./sync"
 
 const log = Log.create({ service: "learn.delivery" })
@@ -91,12 +91,6 @@ function dropLines(text: string, lines: ReadonlySet<string>): string {
   return kept.length ? [heading, ...kept].join("\n") : ""
 }
 
-/** A stable RFC 4122-shaped id from a hex digest (the server only requires a UUID). */
-function uuidFrom(hex: string): string {
-  const variant = ((parseInt(hex[16], 16) & 0x3) | 0x8).toString(16)
-  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-4${hex.slice(13, 16)}-${variant}${hex.slice(17, 20)}-${hex.slice(20, 32)}`
-}
-
 async function read(root: string, file: string) {
   try {
     await SafeFS.assertSafePath(root, file)
@@ -582,11 +576,9 @@ export class Delivery {
     for (const [key, batch] of batches) {
       const ids = [...batch.items.keys()].sort()
       const batchId = uuidFrom(Store.sha256(JSON.stringify([session, name, key, ids])))
-      if (outbox.usage.some((usage) => usage.batch_id === batchId)) continue
-      outbox.usage.push({
-        batch_id: batchId, scope: batch.scope, created_at: new Date().toISOString(),
-        items: ids.map((id) => ({ public_id: id, applied: batch.items.get(id)!, helpful: 0, harmful: 0 })),
-      })
+      const items = ids.map((id) => ({ public_id: id, applied: batch.items.get(id)!, helpful: 0, harmful: 0 }))
+      for (const chunk of usageBatches(batchId, batch.scope, new Date().toISOString(), items))
+        if (!outbox.usage.some((usage) => usage.batch_id === chunk.batch_id)) outbox.usage.push(chunk)
     }
     await writeOutbox(this.root, name, outbox)
   }
 
@@ -13,11 +13,14 @@ import { createHash } from "node:crypto"
 import { constants } from "node:fs"
 import path from "node:path"
 import z from "zod"
+import { Log } from "@/util/log"
 import { canonical } from "./lesson"
 import { assertLearnLock } from "./lock"
 import * as SafeFS from "./safe-fs"
 import * as Store from "./store"
-import { BatchItem, LessonOut, Tombstone, UsageItem } from "../workspace/lesson-api"
+import { BatchItem, LessonOut, MAX_USAGE_ITEMS, Tombstone, UsageItem } from "../workspace/lesson-api"
+
+const log = Log.create({ service: "learn.ledger" })
 
 /** Where lessons sync to: an account (credential digest) on a host and tenant, a workspace, and a repository. */
 export const Scope = z.object({
@@ -75,7 +78,11 @@ export const SyncState = z.object({
   }).optional(),
   /** The server has no lesson routes: retry after this time. */
   unsupported_until: z.string().optional(),
+  /** Keyset cursor into the caller's submissions: the last row read, `(updated_at, public_id)`. */
   submissions_cursor: z.string().optional(),
+  submissions_after_id: z.string().optional(),
+  /** Where the next pull's `local_keys` window starts when the store has more keys than one request carries. */
+  local_keys_offset: z.number().int().nonnegative().default(0),
   backfill: z.object({ complete: z.boolean(), done: z.array(z.string()) }).default({ complete: false, done: [] }),
   ledger: z.record(z.string(), LedgerEntry).default({}),
   /** Local lessons the server would refuse as they are (for example grandfathered text over 140 characters). */
@@ -129,7 +136,7 @@ async function read(root: string, file: string): Promise<string | undefined> {
   }
 }
 
-/** A malformed state file reads as absent: sync state is a cache and a queue, rebuilt by the next pull. */
+/** A malformed cache or sync state reads as absent: both are rebuilt by the next pull. */
 async function readParsed<T>(root: string, file: string, schema: z.ZodType<T>): Promise<T | undefined> {
   const raw = await read(root, file)
   if (raw === undefined) return undefined
@@ -149,10 +156,52 @@ export async function removeRemote(root: string, store: string) {
   await SafeFS.remove(root, files(root, store).remote)
 }
 
-export async function readOutbox(root: string, store: string): Promise<Outbox> {
-  return (await readParsed(root, files(root, store).outbox, Outbox)) ?? { version: 1, proposals: [], usage: [] }
+/** The outbox in `raw`; when it is malformed (`intact: false`), every entry that is still valid. */
+function parseOutbox(raw: string): { outbox: Outbox; intact: boolean } {
+  let value: unknown
+  try { value = JSON.parse(raw) } catch { value = undefined }
+  const whole = Outbox.safeParse(value)
+  if (whole.success) return { outbox: whole.data, intact: true }
+  const fields = value && typeof value === "object" ? (value as Record<string, unknown>) : {}
+  const valid = <T>(list: unknown, schema: z.ZodType<T>) =>
+    Array.isArray(list) ? list.flatMap((item) => { const parsed = schema.safeParse(item); return parsed.success ? [parsed.data] : [] }) : []
+  return { outbox: { version: 1, proposals: valid(fields.proposals, Proposal), usage: valid(fields.usage, UsageBatch) }, intact: false }
+}
+
+/** An outbox as read: `intact: false` when the file was malformed and must be set aside before it is replaced. */
+export type ReadOutbox = Outbox & { intact: boolean }
+
+/**
+ * The outbox is the only copy of what is waiting to be sent, so a malformed file never reads as empty: it yields
+ * its valid entries, and `writeOutbox` sets the file aside (`outbox.json.corrupt-<ms>`) before replacing it.
+ * Usage batches over `MAX_USAGE_ITEMS` (queued by an earlier version) are split here, never sent whole.
+ */
+export async function readOutbox(root: string, store: string): Promise<ReadOutbox> {
+  const raw = await read(root, files(root, store).outbox)
+  if (raw === undefined) return { version: 1, proposals: [], usage: [], intact: true }
+  const { outbox, intact } = parseOutbox(raw)
+  if (!intact) log.warn("learn outbox is malformed; using its valid entries", { store, proposals: outbox.proposals.length, usage: outbox.usage.length })
+  const usage = outbox.usage.flatMap((batch) => usageBatches(batch.batch_id, batch.scope, batch.created_at, batch.items))
+  return { ...outbox, usage, intact }
+}
+
+/** `outbox` is what `readOutbox` returned, changed; a malformed file it was read from is kept aside first. */
+export async function writeOutbox(root: string, store: string, outbox: Outbox & { intact?: boolean }) {
+  await assertLearnLock(root)
+  const file = files(root, store).outbox
+  const { intact, ...value } = outbox
+  if (intact === false) {
+    const aside = `${file}.corrupt-${Date.now()}`
+    const moved = await SafeFS.rename(root, file, aside).then(() => true, (error: NodeJS.ErrnoException) => {
+      if (error.code === "ENOENT") return false
+      throw error
+    })
+    if (moved) log.warn("learn outbox was malformed; kept a copy and started a new one", { store, file: path.basename(aside) })
+  }
+  await write(root, store, file, value)
+  // Set aside once: a second write from the same read must not move the new file.
+  if (intact === false) outbox.intact = true
 }
-export const writeOutbox = (root: string, store: string, outbox: Outbox) => write(root, store, files(root, store).outbox, outbox)
 
 export async function readSyncState(root: string, store: string): Promise<SyncState> {
   return (await readParsed(root, files(root, store).sync, SyncState)) ?? SyncState.parse({ version: 1 })
@@ -162,7 +211,7 @@ export const writeSyncState = (root: string, store: string, state: SyncState) =>
 /** The ledger as it stands for `scope`; a different scope's ledger and backfill are discarded, never reused. */
 export function forScope(state: SyncState, scope: Scope): SyncState {
   if (state.scope === scopeKey(scope)) return state
-  return { ...state, scope: scopeKey(scope), submissions_cursor: undefined, backfill: { complete: false, done: [] }, ledger: {}, unsyncable: {} }
+  return { ...state, scope: scopeKey(scope), submissions_cursor: undefined, submissions_after_id: undefined, backfill: { complete: false, done: [] }, ledger: {}, unsyncable: {} }
 }
 
 /** The pulled cache, only when it was pulled for exactly this scope. */
@@ -196,7 +245,11 @@ const OPEN = "candidate"
 /** Record the caller's own proposals: the open one is the pending slot, every other status is a receipt. */
 export function applySubmissions(state: SyncState, store: string, rows: { public_id: string; lesson_key: string; repo_identity: string | null; store: string; status: string; version: number; submission_hash: string; content_hash?: string | null; updated_at: string }[]) {
   for (const row of rows) {
-    if (!state.submissions_cursor || row.updated_at > state.submissions_cursor) state.submissions_cursor = row.updated_at
+    const cursor = state.submissions_cursor
+    if (!cursor || row.updated_at > cursor || (row.updated_at === cursor && row.public_id > (state.submissions_after_id ?? ""))) {
+      state.submissions_cursor = row.updated_at
+      state.submissions_after_id = row.public_id
+    }
     if (row.store !== store) continue
     const value = entry(state, identityKey(row))
     const receipt = value.receipts.find((r) => r.submission_hash === row.submission_hash)
@@ -217,3 +270,24 @@ export function applyResult(state: SyncState, key: string, result: { submission_
   if (result.status === OPEN) value.pending = { public_id: result.public_id, version: result.version, submission_hash: result.submission_hash }
   else if (value.pending?.public_id === result.public_id) delete value.pending
 }
+
+/** A stable RFC 4122-shaped id from a hex digest (the server only requires a UUID for `batch_id`). */
+export function uuidFrom(hex: string): string {
+  const variant = ((parseInt(hex[16], 16) & 0x3) | 0x8).toString(16)
+  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-4${hex.slice(13, 16)}-${variant}${hex.slice(17, 20)}-${hex.slice(20, 32)}`
+}
+
+/**
+ * Usage items as outbox batches of at most `MAX_USAGE_ITEMS`. The server keys its receipt on `batch_id`, so
+ * each chunk gets its own id, derived from `id` and the chunk index; a single chunk keeps `id` itself.
+ */
+export function usageBatches(id: string, scope: Scope, created_at: string, items: readonly z.infer<typeof UsageItem>[]): UsageBatch[] {
+  if (items.length <= MAX_USAGE_ITEMS) return [{ batch_id: id, scope, created_at, items: [...items] }]
+  const batches: UsageBatch[] = []
+  for (let start = 0; start < items.length; start += MAX_USAGE_ITEMS)
+    batches.push({
+      batch_id: uuidFrom(createHash("sha256").update(`${id}\0${start / MAX_USAGE_ITEMS}`).digest("hex")),
+      scope, created_at, items: items.slice(start, start + MAX_USAGE_ITEMS),
+    })
+  return batches
+}