diff --git a/CHANGELOG.md b/CHANGELOG.md
index 583287f..33cbe72 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,21 @@ The project publishes 0.x prerelease versions; a stable release line is not yet
 - Migrate GitHub repository, Release, issue, badge, and raw-content coordinates
   to the canonical `bytefolk` organization while retaining the published npm
   scope, MCP identity, and existing cache paths.
+- Internal: the local ingestion mechanics used by
+  `mem ingest qoder` — deterministic recursive transcript walk, per-path line
+  cursors (atomic rename write, reset when a file is rewritten shorter), the
+  `--dry-run` / `--limit` semantics, per-file degradation on an idempotency
+  conflict, run-report aggregation and the closed failure-code vocabulary —
+  moved out of `server/cmd/mem` into a new `server/internal/ingest` package
+  (`#111`). The connector is now a thin call site that supplies the Qoder parser,
+  the memory payload and the HTTP upload. Memories payloads, the stdout summary,
+  and the cursor file format and location under `~/.mem/ingest/qoder` are
+  unchanged, and a root already given as a canonical absolute path keeps the
+  cursor keys and `Idempotency-Key` values it had before the move. A root given
+  relative, or one reached through a symlink, is now identified by its canonical
+  absolute path, so its cursor key and per-line `Idempotency-Key` differ from the
+  pre-refactor spelling. The package is the shared core that `put --watch`
+  (`#110`) consumes instead of writing a second state layer.
 
 ### Security
 
@@ -45,6 +60,28 @@ The project publishes 0.x prerelease versions; a stable release line is not yet
 
 ### Fixed
 
+- `mem ingest qoder` derives a transcript's checkpoint key, its per-line
+  `Idempotency-Key` values and its project/session memory path from one canonical
+  root identity. Previously the root was used exactly as the caller spelled it, so
+  two working directories that each contained `sessions/p.jsonl` and shared one
+  checkpoint directory collided on a single cursor: the second run saw an
+  up-to-date checkpoint and posted nothing, silently dropping that store. A
+  relative or symlinked root now re-keys its existing cursors, which replays those
+  files once instead of skipping them. Checkpoint writes stage through a distinct
+  temporary file per save and no longer rewind a checkpoint another run already
+  advanced, so a slower run finishing second can neither fail on a shared staging
+  name nor undo the faster run's progress.
+- An ingest cycle that aborted on a rejected write tallied the failure under the
+  network code whatever the server had answered, because the connector mapped the
+  typed API error to a CLI error before the shared core could classify it. The
+  core now sees the typed error and classifies authentication, plan, quota,
+  provider and timeout responses correctly, while the SPEC §7.1 exit codes stay
+  mapped at the command boundary that owns them. The same tally was wrong for
+  local reads: a failed `open` returns a `syscall.Errno`, which satisfies
+  `net.Error`, so an absent or unreadable transcript was reported as a network
+  failure instead of `root_missing` / `read_denied`, and a run that aborted while
+  reading recorded no failure at all. Both paths now classify before the transport
+  default and the aborted cycle reports the code it died on.
 - The npm installer now verifies the selected Release binary against the
   release's SHA-256 manifest before making it executable, rejects malformed or
   ambiguous manifest entries, verifies cached binaries, and removes partial or
diff --git a/docs/integrations/qoder-ingest.md b/docs/integrations/qoder-ingest.md
index 8bf00fb..3b49708 100644
--- a/docs/integrations/qoder-ingest.md
+++ b/docs/integrations/qoder-ingest.md
@@ -20,7 +20,7 @@ mem ingest qoder --limit 200         # stop after 200 memories
 
 | Flag | Default | Meaning |
 | --- | --- | --- |
-| `--root` | `~/.qoder/projects` | glob base scanned recursively for `*.jsonl` |
+| `--root` | `~/.qoder/projects` | glob base scanned recursively for `*.jsonl`, resolved to a canonical absolute path first |
 | `--path-root` | `/AgentTranscripts` | virtual path prefix for ingested memories |
 | `--state-dir` | `~/.mem/ingest/qoder` | checkpoint cursor directory |
 | `--dry-run` | `false` | parse and plan only; do not write |
@@ -35,7 +35,8 @@ Each parseable line becomes one memory:
 - **path** — `<path-root>/<project>/<session>`, where `<project>` is the first
   path segment under the ingest root and `<session>` is the transcript file name
   minus `.jsonl`
-- **source** — `{"type":"qoder","ref":<abs path>,"locator":{"line":N}}`
+- **source** — `{"type":"qoder","ref":<abs>,"locator":{"line":N}}`, where `<abs>`
+  is the transcript's canonical absolute path (symlinks resolved)
 - **producer** — `session_id` (session slug) and `agent_id` (the model/agent id
   recorded on the line, when present)
 - **event_at** — the message timestamp (RFC 3339 or epoch), when present
@@ -62,9 +63,14 @@ Ingestion is **incremental and idempotent**:
 
 - A per-file cursor (`~/.mem/ingest/qoder/<sha1(abs)>.json`) records the highest
   already-ingested line. A re-run parses only lines appended since the last run.
+  Because the key is the canonical path, one store shares one cursor however it is
+  reached, and two directories that each contain `sessions/p.jsonl` do not.
 - Even if the cursor is lost, the stable `Idempotency-Key` per line makes a
   re-post an idempotent replay (`replayed` responses are counted, not
   duplicated).
+- A `--root` previously spelled relative, or one behind a symlink, is keyed
+  differently than before: its existing cursor is orphaned and that store is
+  re-posted once under new keys.
 
 Deleting `~/.mem/ingest/qoder` resets all cursors (safe: ids remain idempotent).
 
diff --git a/server/cmd/mem/cmds_ingest.go b/server/cmd/mem/cmds_ingest.go
index 2ed5b20..ca9f6b6 100644
--- a/server/cmd/mem/cmds_ingest.go
+++ b/server/cmd/mem/cmds_ingest.go
@@ -9,10 +9,10 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
-	"sort"
 	"strings"
 
 	"github.com/PeterGuy326/mem/server/internal/apiclient"
+	"github.com/PeterGuy326/mem/server/internal/ingest"
 	"github.com/spf13/cobra"
 )
 
@@ -112,30 +112,14 @@ func cliStateRoot() string {
 	return filepath.Join(home, ".mem")
 }
 
-// expandTranscriptGlob recursively collects *.jsonl transcripts under base and
-// sorts the results, so ingestion order is deterministic across runs and
-// checkpoints are stable. (Go's filepath.Glob does not treat ** as recursive, so
-// we walk the tree explicitly.)
+// expandTranscriptGlob collects the transcripts to offer for ingestion. The walk
+// and its ordering belong to the shared core; this wrapper keeps the CLI's
+// exit-code contract for an unwalkable root.
 func expandTranscriptGlob(base string) ([]string, error) {
-	// A bare base that names a single existing file (not a directory) is
-	// accepted as a one-off transcript.
-	if fi, e := os.Stat(base); e == nil && !fi.IsDir() {
-		return []string{base}, nil
-	}
-	var paths []string
-	err := filepath.WalkDir(base, func(p string, d os.DirEntry, err error) error {
-		if err != nil {
-			return nil // skip unreadable entries rather than failing the walk
-		}
-		if !d.IsDir() && strings.EqualFold(filepath.Ext(p), ".jsonl") {
-			paths = append(paths, p)
-		}
-		return nil
-	})
+	paths, err := ingest.Walk(base, ingest.HasJSONLExtension)
 	if err != nil {
 		return nil, newCliError(1, fmt.Sprintf("walk %s: %v", base, err), "")
 	}
-	sort.Strings(paths)
 	return paths, nil
 }
 
@@ -144,6 +128,13 @@ func runIngestQoder(cmd *cobra.Command, o ingestOptions) error {
 	if base == "" {
 		return newCliError(1, "cannot determine session store root", "set --root or $HOME")
 	}
+	// Walk, the checkpoint key and the project/session split must all see one
+	// identity. Left as the caller spelled it, a relative --root would make two
+	// working directories that each hold sessions/p.jsonl share a checkpoint.
+	base, err := ingest.CanonicalRoot(base)
+	if err != nil {
+		return newCliError(1, err.Error(), "set --root to an accessible path")
+	}
 	paths, err := expandTranscriptGlob(base)
 	if err != nil {
 		return err
@@ -161,106 +152,90 @@ func runIngestQoder(cmd *cobra.Command, o ingestOptions) error {
 		return newCliError(3, "not logged in", "run `mem auth login` first")
 	}
 	client := newHTTPClient(cfg)
-	stateDir := o.checkpointDir()
+	warn := func(format string, args ...any) {
+		fmt.Fprintf(cmd.ErrOrStderr(), format, args...)
+	}
 
-	var (
-		files       = 0
-		memories    int // written this run
-		replayed    int // server-reported idempotent replays
-		unparseable int // parsed lines yielded no ingestible text
-		remaining   = o.limit
+	report, err := ingest.Run(
+		context.Background(),
+		paths,
+		ingest.Options{
+			StateDir: o.checkpointDir(),
+			DryRun:   o.dryRun,
+			Limit:    o.limit,
+			Log:      warn,
+		},
+		o.parseTranscript(base),
+		o.uploadMemory(client, warn),
 	)
+	if err != nil {
+		// Classify runs inside the core on the typed error, so the exit-code
+		// mapping belongs here, at the boundary that owns it.
+		return fromAPIError(err)
+	}
+
+	fmt.Fprintf(cmd.OutOrStdout(),
+		"qoder ingest: %d file(s), %d memory written, %d server-replay, %d unparseable line%s\n",
+		report.Scanned, report.Ingested, report.Deduped, report.Unparseable,
+		ingestModeNote(o.dryRun))
+	return nil
+}
 
-	for _, abs := range paths {
-		files++
-		cp := loadQoderCheckpoint(stateDir, abs)
-		turns, skipped, perr := parseQoderTranscript(abs, cp.LastLine)
-		if perr != nil {
-			return perr
+// parseTranscript adapts the Qoder transcript reader to the core's ParseFunc:
+// every turn becomes a unit carrying its own request body and stable key.
+func (o ingestOptions) parseTranscript(base string) ingest.ParseFunc {
+	return func(abs string, skipBefore int) ([]ingest.Unit, int, error) {
+		turns, skipped, err := parseQoderTranscript(abs, skipBefore)
+		if err != nil {
+			return nil, skipped, err
 		}
-		unparseable += skipped
 		project, session := splitTranscriptPath(base, abs)
-
-		newLast := cp.LastLine
+		units := make([]ingest.Unit, 0, len(turns))
 		for _, turn := range turns {
-			if o.limit > 0 && remaining <= 0 {
-				break
-			}
-			// parseQoderTranscript already skips <= cp.LastLine, so this
-			// guard is a belt-and-suspenders check against anomalies.
-			if turn.Line <= newLast {
-				continue
-			}
-
-			key := ingestIdempotencyKey(abs, turn.Line)
-			body := ingestMemoryBody(o, abs, project, session, turn)
-
-			if o.dryRun {
-				memories++
-				if remaining > 0 {
-					remaining--
-				}
-				continue
-			}
-
-			var resp map[string]any
-			// Use the raw client to detect 409 (Idempotency-Key conflict)
-			// without wrapping into cliError, which would lose the kind.
-			err := client.api.DoJSONWithHeaders(
-				context.Background(),
-				http.MethodPost,
-				"/v1/memories",
-				body,
-				&resp,
-				map[string]string{"Idempotency-Key": key},
-			)
-			if err != nil {
-				// 409 Idempotency-Key conflict: the file was rewritten with
-				// different content at the same line — skip the remainder of
-				// this file and continue with others rather than aborting the
-				// whole run. The checkpoint is NOT advanced for this file, so
-				// the operator can investigate and retry.
-				var ae *apiclient.APIError
-				if errors.As(err, &ae) && ae.Kind() == apiclient.KindConflict {
-					fmt.Fprintf(cmd.ErrOrStderr(),
-						"warn: %s line %d: idempotency conflict (file rewritten?); skipping remaining lines in %s\n",
-						abs, turn.Line, filepath.Base(abs))
-					break
-				}
-				return fromAPIError(err)
-			}
-			if r, _ := resp["replayed"].(bool); r {
-				replayed++
-			} else {
-				memories++
-			}
-			if remaining > 0 {
-				remaining--
-			}
-			newLast = turn.Line
+			units = append(units, ingest.Unit{
+				Line:           turn.Line,
+				Body:           ingestMemoryBody(o, abs, project, session, turn),
+				IdempotencyKey: ingestIdempotencyKey(abs, turn.Line),
+			})
 		}
+		return units, skipped, nil
+	}
+}
 
-		if !o.dryRun {
-			if size, mtime, serr := fileState(abs); serr == nil {
-				if err := saveQoderCheckpoint(stateDir, qoderCheckpoint{
-					Abs:      abs,
-					Size:     size,
-					ModTime:  mtime,
-					LastLine: newLast,
-				}); err != nil {
-					fmt.Fprintf(cmd.ErrOrStderr(), "warn: save checkpoint for %s: %v\n", abs, err)
-				}
+// uploadMemory posts one unit through the standard memories endpoint. Errors
+// stay typed on purpose: the core's Classify dispatches on *apiclient.APIError
+// for both the per-file degradation decision and the failure code in the report,
+// and the SPEC §7.1 exit-code mapping is applied by the calling command.
+func (o ingestOptions) uploadMemory(client *httpClient, warn func(string, ...any)) ingest.UploadFunc {
+	return func(ctx context.Context, abs string, u ingest.Unit) (ingest.Outcome, error) {
+		var resp map[string]any
+		err := client.api.DoJSONWithHeaders(
+			ctx,
+			http.MethodPost,
+			"/v1/memories",
+			u.Body,
+			&resp,
+			map[string]string{"Idempotency-Key": u.IdempotencyKey},
+		)
+		if err != nil {
+			var ae *apiclient.APIError
+			if errors.As(err, &ae) && ae.Kind() == apiclient.KindConflict {
+				// The file was rewritten with different content at the same
+				// line, so every later line keeps colliding on its stable key.
+				warn("warn: %s line %d: idempotency conflict (file rewritten?); skipping remaining lines in %s\n",
+					abs, u.Line, filepath.Base(abs))
+				return ingest.Outcome{}, fmt.Errorf("%w: %s:%d", ingest.ErrDegradeFile, abs, u.Line)
 			}
+			return ingest.Outcome{}, err
 		}
+		if r, _ := resp["replayed"].(bool); r {
+			return ingest.Outcome{Deduplicated: true}, nil
+		}
+		return ingest.Outcome{}, nil
 	}
-
-	fmt.Fprintf(cmd.OutOrStdout(),
-		"qoder ingest: %d file(s), %d memory written, %d server-replay, %d unparseable line%s\n",
-		files, memories, replayed, unparseable,
-		ingestModeNote(o.dryRun))
-	return nil
 }
 
+// ingestModeNote is the stdout suffix that separates a plan from a write.
 func ingestModeNote(dryRun bool) string {
 	if dryRun {
 		return " (dry-run: no writes)"
diff --git a/server/cmd/mem/cmds_ingest_test.go b/server/cmd/mem/cmds_ingest_test.go
index 3f720f5..b9c240d 100644
--- a/server/cmd/mem/cmds_ingest_test.go
+++ b/server/cmd/mem/cmds_ingest_test.go
@@ -2,17 +2,23 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
+
+	"github.com/PeterGuy326/mem/server/internal/apiclient"
+	"github.com/PeterGuy326/mem/server/internal/ingest"
 )
 
 // writeTranscript writes a small valid JSONL transcript containing three
@@ -376,11 +382,11 @@ func TestIngestQoder409ConflictDegradesPerFile(t *testing.T) {
 	if !strings.Contains(stdout.String(), "idempotency conflict") {
 		t.Fatalf("expected conflict warning, got stdout = %q", stdout.String())
 	}
-	cp := loadQoderCheckpoint(cpDir, abs1)
+	cp := ingest.LoadCursor(cpDir, abs1)
 	if cp.LastLine != 1 {
 		t.Fatalf("project-a checkpoint LastLine = %d, want 1 (line 2 failed)", cp.LastLine)
 	}
-	cp2 := loadQoderCheckpoint(cpDir, abs2)
+	cp2 := ingest.LoadCursor(cpDir, abs2)
 	if cp2.LastLine != 3 {
 		t.Fatalf("project-b checkpoint LastLine = %d, want 3", cp2.LastLine)
 	}
@@ -418,7 +424,7 @@ func TestIngestQoderLimitCheckpoint(t *testing.T) {
 	if requests.Load() != 2 {
 		t.Fatalf("first run requests = %d, want 2", requests.Load())
 	}
-	cp := loadQoderCheckpoint(cpDir, abs)
+	cp := ingest.LoadCursor(cpDir, abs)
 	if cp.LastLine != 2 {
 		t.Fatalf("after limit checkpoint LastLine = %d, want 2", cp.LastLine)
 	}
@@ -433,8 +439,300 @@ func TestIngestQoderLimitCheckpoint(t *testing.T) {
 	if requests.Load() != 1 {
 		t.Fatalf("second run requests = %d, want 1", requests.Load())
 	}
-	cp2 := loadQoderCheckpoint(cpDir, abs)
+	cp2 := ingest.LoadCursor(cpDir, abs)
 	if cp2.LastLine != 3 {
 		t.Fatalf("after second run LastLine = %d, want 3", cp2.LastLine)
 	}
 }
+
+// TestIngestQoderUploadErrorsStayTyped covers the adapter's error contract: the
+// core derives the report's failure code from the error it is handed, so
+// translating it inside uploadMemory would report every API failure as a network
+// failure.
+func TestIngestQoderUploadErrorsStayTyped(t *testing.T) {
+	cases := []struct {
+		name   string
+		status int
+		// apiError expects a *apiclient.APIError with this status to survive.
+		apiError    bool
+		degrade     bool
+		want        ingest.Code
+		unreachable bool
+	}{
+		{name: "400", status: http.StatusBadRequest, apiError: true, want: ingest.CodeUploadRejected},
+		{name: "401", status: http.StatusUnauthorized, apiError: true, want: ingest.CodeAuth},
+		{name: "402", status: http.StatusPaymentRequired, apiError: true, want: ingest.CodePlanQuota},
+		{name: "403", status: http.StatusForbidden, apiError: true, want: ingest.CodeAuth},
+		{name: "409", status: http.StatusConflict, degrade: true, want: ingest.CodeUploadRejected},
+		{name: "429", status: http.StatusTooManyRequests, apiError: true, want: ingest.CodePlanQuota},
+		{name: "502", status: http.StatusBadGateway, apiError: true, want: ingest.CodeProviderTimeout},
+		{name: "503", status: http.StatusServiceUnavailable, apiError: true, want: ingest.CodeProviderTimeout},
+		{name: "504", status: http.StatusGatewayTimeout, apiError: true, want: ingest.CodeProviderTimeout},
+		{name: "network", unreachable: true, want: ingest.CodeNetwork},
+	}
+	for _, tc := range cases {
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(tc.status)
+			_, _ = fmt.Fprintf(w, `{"error":"%d","hint":"h"}`, tc.status)
+		}))
+		url := srv.URL
+		if tc.unreachable {
+			srv.Close()
+		} else {
+			defer srv.Close()
+		}
+
+		client := newHTTPClient(&cliConfig{Server: url, Token: "tok"})
+		upload := (ingestOptions{}).uploadMemory(client, func(string, ...any) {})
+		_, err := upload(context.Background(), "/store/p/s.jsonl", ingest.Unit{
+			Line: 2, Body: map[string]any{"kind": "observation", "content": "c"},
+		})
+		if err == nil {
+			t.Fatalf("%s: want an error", tc.name)
+		}
+
+		var ae *apiclient.APIError
+		switch {
+		case tc.apiError && !errors.As(err, &ae):
+			t.Errorf("%s: err = %v, want the typed APIError to survive the adapter", tc.name, err)
+		case tc.apiError && ae.StatusCode != tc.status:
+			t.Errorf("%s: APIError.StatusCode = %d, want %d", tc.name, ae.StatusCode, tc.status)
+		case tc.degrade && !errors.Is(err, ingest.ErrDegradeFile):
+			t.Errorf("%s: err = %v, want per-file degradation", tc.name, err)
+		case tc.unreachable && errors.As(err, &ae):
+			t.Errorf("%s: err = %v, want a transport error", tc.name, err)
+		}
+		if got := ingest.Classify(err); got != tc.want {
+			t.Errorf("%s: Classify = %q, want %q", tc.name, got, tc.want)
+		}
+	}
+}
+
+// TestIngestQoderReadFailuresClassify covers the other side of a run: a source
+// that cannot be read must reach the core as the OS error it is, so the report
+// names the read state instead of claiming a transport failure.
+func TestIngestQoderReadFailuresClassify(t *testing.T) {
+	dir := t.TempDir()
+	parse := ingestOptions{}.parseTranscript(dir)
+
+	run := func(path string, wantCode ingest.Code) {
+		t.Helper()
+		report, err := ingest.Run(context.Background(), []string{path},
+			ingest.Options{StateDir: filepath.Join(dir, "state")},
+			parse,
+			func(context.Context, string, ingest.Unit) (ingest.Outcome, error) {
+				t.Error("upload must not be reached for an unreadable source")
+				return ingest.Outcome{}, nil
+			})
+		if err == nil {
+			t.Fatalf("%s: run succeeded", path)
+		}
+		if got := ingest.Classify(err); got != wantCode {
+			t.Errorf("%s: Classify(%v) = %q, want %q", path, err, got, wantCode)
+		}
+		if report.Failures[wantCode] != 1 {
+			t.Errorf("%s: report tally = %+v, want one %q", path, report.Failures, wantCode)
+		}
+	}
+
+	run(filepath.Join(dir, "gone", "p.jsonl"), ingest.CodeRootMissing)
+
+	if runtime.GOOS == "windows" || os.Geteuid() == 0 {
+		t.Skip("mode bits do not deny a read here: the permission case needs a POSIX non-root user")
+	}
+	denied := writeTranscript(t, dir, "denied/s.jsonl")
+	if err := os.Chmod(denied, 0o000); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = os.Chmod(denied, 0o600) }()
+	run(denied, ingest.CodeReadDenied)
+}
+
+// TestIngestQoderMapsExitCodesAtTheBoundary pins the SPEC §7.1 codes that the
+// command owns, which is where the APIError is allowed to become a cliError.
+func TestIngestQoderMapsExitCodesAtTheBoundary(t *testing.T) {
+	cases := []struct {
+		status int
+		want   int
+	}{
+		{http.StatusBadRequest, 1},
+		{http.StatusUnauthorized, 3},
+		{http.StatusForbidden, 3},
+		{http.StatusPaymentRequired, 4},
+		{http.StatusTooManyRequests, 4},
+		{http.StatusBadGateway, 5},
+		{http.StatusServiceUnavailable, 5},
+		{http.StatusGatewayTimeout, 5},
+		{http.StatusInternalServerError, 1},
+	}
+	for _, tc := range cases {
+		dir := t.TempDir()
+		transcriptDir := filepath.Join(dir, "store")
+		writeTranscript(t, transcriptDir, "p/s.jsonl")
+
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(tc.status)
+			_, _ = fmt.Fprintf(w, `{"error":"%d","hint":"h"}`, tc.status)
+		}))
+
+		t.Setenv("MEM_CONFIG", filepath.Join(dir, "missing.yaml"))
+		t.Setenv("MEM_SERVER", srv.URL)
+		t.Setenv("MEM_TOKEN", "tok")
+		t.Setenv("MEM_STATE_DIR", filepath.Join(dir, "state"))
+
+		root := newRootCmd()
+		root.SetOut(&bytes.Buffer{})
+		root.SetErr(&bytes.Buffer{})
+		root.SetArgs([]string{"ingest", "qoder", "--root", transcriptDir})
+		err := root.Execute()
+		srv.Close()
+
+		var ce *cliError
+		if !errors.As(err, &ce) {
+			t.Fatalf("status %d: err = %v, want a cliError", tc.status, err)
+		}
+		if ce.code != tc.want {
+			t.Errorf("status %d: exit code = %d, want %d", tc.status, ce.code, tc.want)
+		}
+	}
+}
+
+// TestIngestQoderRelativeRootKeepsSeparateCheckpoints is the cross-working-
+// directory fixture: two directories each holding an identical store/p/s.jsonl
+// and sharing one checkpoint directory must not look already-ingested to the
+// second run.
+func TestIngestQoderRelativeRootKeepsSeparateCheckpoints(t *testing.T) {
+	var requests atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_, _ = w.Write([]byte(`{"memory":{"id":"m-1"},"replayed":false}`))
+	}))
+	defer srv.Close()
+
+	base := t.TempDir()
+	dirA := filepath.Join(base, "a")
+	dirB := filepath.Join(base, "b")
+	for _, dir := range []string{dirA, dirB} {
+		writeTranscript(t, filepath.Join(dir, "store"), "p/s.jsonl")
+	}
+
+	t.Setenv("MEM_CONFIG", filepath.Join(base, "missing.yaml"))
+	t.Setenv("MEM_SERVER", srv.URL)
+	t.Setenv("MEM_TOKEN", "tok")
+	t.Setenv("MEM_STATE_DIR", filepath.Join(base, "state"))
+
+	run := func() int {
+		t.Helper()
+		requests.Store(0)
+		root := newRootCmd()
+		root.SetOut(&bytes.Buffer{})
+		root.SetErr(&bytes.Buffer{})
+		root.SetArgs([]string{"ingest", "qoder", "--root", "store"})
+		if err := root.Execute(); err != nil {
+			t.Fatal(err)
+		}
+		return int(requests.Load())
+	}
+
+	t.Chdir(dirA)
+	if got := run(); got != 3 {
+		t.Fatalf("first run in directory A posted %d, want 3", got)
+	}
+
+	// Same relative spelling, different files: the cursor must not be shared.
+	t.Chdir(dirB)
+	if got := run(); got != 3 {
+		t.Fatalf("run in directory B posted %d, want 3 (both stores collided on one checkpoint)", got)
+	}
+
+	// Directory A is still complete under its own identity.
+	t.Chdir(dirA)
+	if got := run(); got != 0 {
+		t.Fatalf("re-run in directory A posted %d, want 0", got)
+	}
+	// And the transcript the server was told about is an absolute path, so
+	// provenance does not depend on where the CLI happened to run.
+	absB, err := ingest.CanonicalRoot(filepath.Join(dirB, "store", "p", "s.jsonl"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := ingest.LoadCursor(filepath.Join(base, "state", "ingest", "qoder"), absB).LastLine; got != 3 {
+		t.Fatalf("directory B checkpoint LastLine = %d, want 3 keyed by %s", got, absB)
+	}
+}
+
+// TestIngestQoderRelativeRootKeepsProjectSplit pins the other half of root
+// canonicalization: the base used to derive a memory's project and session must
+// be spelled the same way as the paths the walk returned.
+func TestIngestQoderRelativeRootKeepsProjectSplit(t *testing.T) {
+	dir := t.TempDir()
+	abs, err := ingest.CanonicalRoot(filepath.Join(dir, "store", "campus-2027", "sessions", "recruit-s3e0a.jsonl"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	writeTranscript(t, filepath.Join(dir, "store"), "campus-2027/sessions/recruit-s3e0a.jsonl")
+
+	var (
+		mu     sync.Mutex
+		paths  []string
+		refs   []string
+		keys   []string
+		posted int
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		raw, _ := io.ReadAll(r.Body)
+		var b map[string]any
+		_ = json.Unmarshal(raw, &b)
+		src, _ := b["source"].(map[string]any)
+		ref, _ := src["ref"].(string)
+		mu.Lock()
+		paths = append(paths, fmt.Sprint(b["path"]))
+		refs = append(refs, ref)
+		keys = append(keys, r.Header.Get("Idempotency-Key"))
+		posted++
+		mu.Unlock()
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_, _ = w.Write([]byte(`{"memory":{"id":"m-1"},"replayed":false}`))
+	}))
+	defer srv.Close()
+
+	t.Setenv("MEM_CONFIG", filepath.Join(dir, "missing.yaml"))
+	t.Setenv("MEM_SERVER", srv.URL)
+	t.Setenv("MEM_TOKEN", "tok")
+	t.Setenv("MEM_STATE_DIR", filepath.Join(dir, "state"))
+
+	t.Chdir(dir)
+	root := newRootCmd()
+	root.SetOut(&bytes.Buffer{})
+	root.SetErr(&bytes.Buffer{})
+	root.SetArgs([]string{"ingest", "qoder", "--root", "store"})
+	if err := root.Execute(); err != nil {
+		t.Fatal(err)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if posted != 3 {
+		t.Fatalf("posted %d memories, want 3", posted)
+	}
+	for i := range paths {
+		if !strings.HasPrefix(paths[i], "/AgentTranscripts/campus-2027/recruit-s3e0a") {
+			t.Errorf("path[%d] = %q, want the project taken from the store root", i, paths[i])
+		}
+		if refs[i] != abs {
+			t.Errorf("source.ref[%d] = %q, want the canonical path %q", i, refs[i], abs)
+		}
+		if !strings.HasPrefix(keys[i], "qoder:") {
+			t.Errorf("key[%d] = %q", i, keys[i])
+		}
+	}
+	// The key is derived from the same canonical identity, so it cannot change
+	// when the same store is reached from another working directory.
+	wantKey := ingestIdempotencyKey(abs, 2)
+	if keys[1] != wantKey {
+		t.Errorf("key[1] = %q, want %q", keys[1], wantKey)
+	}
+}
diff --git a/server/cmd/mem/qoder_checkpoint.go b/server/cmd/mem/qoder_checkpoint.go
deleted file mode 100644
index a7fd069..0000000
--- a/server/cmd/mem/qoder_checkpoint.go
+++ /dev/null
@@ -1,89 +0,0 @@
-package main
-
-import (
-	"crypto/sha1"
-	"encoding/hex"
-	"encoding/json"
-	"fmt"
-	"os"
-	"path/filepath"
-)
-
-// qoderCheckpoint is the persisted per-transcript cursor that makes `mem ingest
-// qoder` incremental: it records how many leading lines were already ingested so
-// a re-run processes only newly appended messages. Combined with the stable
-// Idempotency-Key per line, re-runs are both fast and idempotent.
-type qoderCheckpoint struct {
-	Abs      string `json:"abs"`       // absolute path of the transcript
-	Size     int64  `json:"size"`      // file size at write time (diagnostic)
-	ModTime  string `json:"mtime"`     // file mtime at write time (diagnostic)
-	LastLine int    `json:"last_line"` // highest 1-based line already ingested
-}
-
-// qoderCheckpointPath returns the state-dir-relative path for a transcript's
-// cursor, keyed by a content-stable hash of its absolute path.
-func qoderCheckpointPath(stateDir, abs string) string {
-	sum := sha1.Sum([]byte(abs))
-	return filepath.Join(stateDir, hex.EncodeToString(sum[:])+".json")
-}
-
-// loadQoderCheckpoint reads a transcript cursor. A missing or malformed cursor
-// yields the zero value (LastLine 0), meaning "nothing ingested yet" — never a
-// hard error, so a corrupt cursor cannot block ingest.
-//
-// If the on-disk file is now smaller than when the checkpoint was written, the
-// file was truncated and rewritten — reset LastLine so re-ingestion does not
-// skip the new content at formerly-ingested line numbers.
-func loadQoderCheckpoint(stateDir, abs string) qoderCheckpoint {
-	var cp qoderCheckpoint
-	p := qoderCheckpointPath(stateDir, abs)
-	b, err := os.ReadFile(p)
-	if err != nil {
-		return cp
-	}
-	if err := json.Unmarshal(b, &cp); err != nil {
-		return qoderCheckpoint{Abs: abs}
-	}
-	if cp.Abs == "" {
-		cp.Abs = abs
-	}
-	// Detect truncation: if the file was rewritten and is now smaller, reset
-	// the cursor so the new content at formerly-ingested line numbers is not
-	// silently skipped.
-	if cp.Size > 0 {
-		if fi, err := os.Stat(abs); err == nil && fi.Size() < cp.Size {
-			cp.LastLine = 0
-		}
-	}
-	return cp
-}
-
-// saveQoderCheckpoint atomically persists a transcript cursor. Errors are
-// returned (callers may warn without failing the whole ingest).
-func saveQoderCheckpoint(stateDir string, cp qoderCheckpoint) error {
-	p := qoderCheckpointPath(stateDir, cp.Abs)
-	if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
-		return fmt.Errorf("create checkpoint dir: %w", err)
-	}
-	b, err := json.Marshal(cp)
-	if err != nil {
-		return fmt.Errorf("encode checkpoint: %w", err)
-	}
-	tmp := p + ".tmp"
-	if err := os.WriteFile(tmp, b, 0o600); err != nil {
-		return fmt.Errorf("write checkpoint: %w", err)
-	}
-	if err := os.Rename(tmp, p); err != nil {
-		return fmt.Errorf("commit checkpoint: %w", err)
-	}
-	return nil
-}
-
-// fileState returns the size and mtime of a transcript (for diagnostics).
-func fileState(abs string) (size int64, mtime string, err error) {
-	fi, err := os.Stat(abs)
-	if err != nil {
-		return 0, "", err
-	}
-	return fi.Size(), fi.ModTime().UTC().Format("2006-01-02T15:04:05Z07:00"), nil
-}
diff --git a/server/internal/ingest/cursor_lock.go b/server/internal/ingest/cursor_lock.go
new file mode 100644
index 0000000..3091d8d
--- /dev/null
+++ b/server/internal/ingest/cursor_lock.go
@@ -0,0 +1,42 @@
+package ingest
+
+import (
+	"fmt"
+	"os"
+)
+
+// cursorLock holds an advisory lock on one cursor sidecar. The sidecar
+// deliberately remains on disk after release: unlinking a locked file can
+// create a second inode that another process locks independently. The OS
+// releases the advisory lock when this descriptor, or its owning process, exits.
+type cursorLock struct {
+	file *os.File
+}
+
+func acquireCursorLock(cursorPath string) (*cursorLock, error) {
+	file, err := os.OpenFile(cursorPath+".lock", os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, fmt.Errorf("open lock file: %w", err)
+	}
+	if err := lockCursorFile(file); err != nil {
+		_ = file.Close()
+		return nil, fmt.Errorf("acquire OS lock: %w", err)
+	}
+	return &cursorLock{file: file}, nil
+}
+
+func (l *cursorLock) release() error {
+	if l == nil || l.file == nil {
+		return nil
+	}
+	unlockErr := unlockCursorFile(l.file)
+	closeErr := l.file.Close()
+	l.file = nil
+	if unlockErr != nil {
+		return fmt.Errorf("unlock OS lock: %w", unlockErr)
+	}
+	if closeErr != nil {
+		return fmt.Errorf("close lock file: %w", closeErr)
+	}
+	return nil
+}
diff --git a/server/internal/ingest/cursor_lock_aix.go b/server/internal/ingest/cursor_lock_aix.go
new file mode 100644
index 0000000..eaccd7d
--- /dev/null
+++ b/server/internal/ingest/cursor_lock_aix.go
@@ -0,0 +1,25 @@
+//go:build aix
+
+package ingest
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// AIX does not expose flock(2) through x/sys, so use the blocking fcntl record
+// lock equivalent for the first byte of the persistent sidecar inode.
+func lockCursorFile(file *os.File) error {
+	return unix.FcntlFlock(file.Fd(), unix.F_SETLKW, &unix.Flock_t{
+		Type: unix.F_WRLCK,
+		Len:  1,
+	})
+}
+
+func unlockCursorFile(file *os.File) error {
+	return unix.FcntlFlock(file.Fd(), unix.F_SETLK, &unix.Flock_t{
+		Type: unix.F_UNLCK,
+		Len:  1,
+	})
+}
diff --git a/server/internal/ingest/cursor_lock_other.go b/server/internal/ingest/cursor_lock_other.go
new file mode 100644
index 0000000..ef18b73
--- /dev/null
+++ b/server/internal/ingest/cursor_lock_other.go
@@ -0,0 +1,16 @@
+//go:build !(aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris || windows)
+
+package ingest
+
+import (
+	"fmt"
+	"os"
+)
+
+func lockCursorFile(_ *os.File) error {
+	return fmt.Errorf("cursor locks are not supported on this operating system")
+}
+
+func unlockCursorFile(_ *os.File) error {
+	return nil
+}
diff --git a/server/internal/ingest/cursor_lock_unix.go b/server/internal/ingest/cursor_lock_unix.go
new file mode 100644
index 0000000..5d87009
--- /dev/null
+++ b/server/internal/ingest/cursor_lock_unix.go
@@ -0,0 +1,17 @@
+//go:build darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
+
+package ingest
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+func lockCursorFile(file *os.File) error {
+	return unix.Flock(int(file.Fd()), unix.LOCK_EX)
+}
+
+func unlockCursorFile(file *os.File) error {
+	return unix.Flock(int(file.Fd()), unix.LOCK_UN)
+}
diff --git a/server/internal/ingest/cursor_lock_windows.go b/server/internal/ingest/cursor_lock_windows.go
new file mode 100644
index 0000000..533f241
--- /dev/null
+++ b/server/internal/ingest/cursor_lock_windows.go
@@ -0,0 +1,26 @@
+//go:build windows
+
+package ingest
+
+import (
+	"os"
+
+	"golang.org/x/sys/windows"
+)
+
+// Lock a one-byte range. Windows releases a LockFileEx lock when the owning
+// process or file handle exits, matching the Unix advisory-lock lifecycle.
+func lockCursorFile(file *os.File) error {
+	return windows.LockFileEx(
+		windows.Handle(file.Fd()),
+		windows.LOCKFILE_EXCLUSIVE_LOCK,
+		0,
+		1,
+		0,
+		&windows.Overlapped{},
+	)
+}
+
+func unlockCursorFile(file *os.File) error {
+	return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &windows.Overlapped{})
+}
diff --git a/server/internal/ingest/ingest.go b/server/internal/ingest/ingest.go
new file mode 100644
index 0000000..e852629
--- /dev/null
+++ b/server/internal/ingest/ingest.go
@@ -0,0 +1,497 @@
+// Package ingest owns the mechanics that every local→mem ingestion connector
+// would otherwise re-implement: a deterministic recursive walk, a per-file
+// incremental cursor store, the change decision, a closed failure-code
+// vocabulary, and cycle report aggregation.
+//
+// The package is deliberately unaware of any particular input format, of HTTP,
+// and of the command surface. A connector supplies two functions:
+//
+//	Parse  turns one local file into ordered Units, given the leading line
+//	       count already ingested, and reports how many lines it could not use.
+//	Upload persists one Unit and reports whether the server replayed it. Its
+//	       error is one that Classify understands.
+//
+// Call sites stay thin: `mem ingest qoder` today wires a transcript parser and
+// a /v1/memories POST to Run, and a future `mem put --watch` wires a different
+// source and sink to the same Run, so cursor layout, change detection and the
+// report vocabulary are written once.
+//
+// Run returns a Report; printing it is the caller's job, which is why nothing
+// here touches cobra or an io.Writer directly. Diagnostics go through
+// Options.Log when the caller wants them.
+//
+// Contract notes that matter for new call sites:
+//
+//   - The change decision is size-based, not content-hashed. A cursor records
+//     the file size at write time, and a file that has since become smaller is
+//     treated as rewritten so its cursor resets. Nothing compares content, so a
+//     same-size in-place edit is not detected here; adding a content gate is a
+//     decision to make, not an implementation detail of a call site.
+//   - Cursors are keyed by the canonical absolute path (see CanonicalRoot,
+//     CursorPath), which Walk establishes for every path it returns. Keying on
+//     path-plus-device identity would invalidate existing on-disk cursors.
+//   - --dry-run neither writes a request nor advances a cursor. Callers must
+//     not "optimize" by saving a cursor after a dry run.
+package ingest
+
+import (
+	"context"
+	"crypto/sha1"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"github.com/PeterGuy326/mem/server/internal/apiclient"
+)
+
+// Code is the closed set of failure classifications a run may report. Names are
+// shared vocabulary: report consumers (CLI text, watch daemon, JSON output)
+// must not invent per-call-site aliases for the same condition.
+type Code string
+
+const (
+	// CodeAuth: the server rejected our credentials (401/403).
+	CodeAuth Code = "auth"
+	// CodePlanQuota: plan or quota blocked the write (402/429).
+	CodePlanQuota Code = "plan_quota"
+	// CodeProviderTimeout: an upstream provider stage failed or timed out
+	// (502/503/504).
+	CodeProviderTimeout Code = "provider_timeout"
+	// CodeNetwork: the request never reached the server.
+	CodeNetwork Code = "network"
+	// CodeReadDenied: a local path could not be read.
+	CodeReadDenied Code = "read_denied"
+	// CodeUploadRejected: the server refused this specific unit (409 on a
+	// stable idempotency key, or a rejected payload).
+	CodeUploadRejected Code = "upload_rejected"
+	// CodeRootMissing: the configured source root does not exist.
+	CodeRootMissing Code = "root_missing"
+	// CodeStateCorrupt: a cursor could not be decoded, so it was treated as
+	// "nothing ingested yet" rather than blocking the run.
+	CodeStateCorrupt Code = "state_corrupt"
+)
+
+// ErrDegradeFile lets an UploadFunc say "stop this file, keep the run going".
+// A rewritten file can conflict with its own stable per-line keys forever, so
+// aborting the whole cycle would let one bad file block every other source.
+// The cursor for that file is not advanced, which keeps a retry meaningful.
+var ErrDegradeFile = errors.New("ingest: degrade file")
+
+// Unit is one ingestible item produced by a connector's Parse function.
+type Unit struct {
+	// Line is the 1-based position in the source file that this unit came
+	// from. Run records it in the cursor as the high-water mark.
+	Line int
+	// Body is the request payload, opaque to this package.
+	Body any
+	// IdempotencyKey is the connector's stable retry key for this unit.
+	IdempotencyKey string
+}
+
+// ParseFunc converts one local file into the units that have not been ingested
+// yet. skipBefore is the cursor's high-water mark: units at or below it must
+// not be returned. The second result counts lines that were readable but
+// produced no unit (malformed, empty, or out of scope for the format).
+type ParseFunc func(abs string, skipBefore int) ([]Unit, int, error)
+
+// Outcome is what an Upload call reports back about one unit.
+type Outcome struct {
+	// Deduplicated marks a server-reported idempotent replay: the memory
+	// already existed for this key, so nothing new was written. Counting
+	// replays as ingested would overstate a re-run.
+	Deduplicated bool
+}
+
+// UploadFunc persists one unit and reports whether the server treated it as a
+// replay. Return an error wrapping ErrDegradeFile to skip the rest of the
+// current file, or any other error to end the run.
+type UploadFunc func(ctx context.Context, abs string, u Unit) (Outcome, error)
+
+// Cursor is the persisted per-file checkpoint. The field order and JSON names
+// are the on-disk format: changing either would strand cursors that existing
+// users already have.
+type Cursor struct {
+	Abs      string `json:"abs"`
+	Size     int64  `json:"size"`
+	ModTime  string `json:"mtime"`
+	LastLine int    `json:"last_line"`
+
+	// Corrupt is set in memory when a stored cursor failed to decode. It is
+	// never persisted.
+	Corrupt bool `json:"-"`
+}
+
+// Options configures one run.
+type Options struct {
+	// StateDir holds the cursors. Required.
+	StateDir string
+	// DryRun plans only: no Upload call, no cursor write.
+	DryRun bool
+	// Limit stops ingesting units after this many (0 = no limit).
+	Limit int
+	// Log receives diagnostics. Nil discards them.
+	Log func(format string, args ...any)
+}
+
+// Report aggregates one cycle. Run populates Scanned, Ingested, Deduped,
+// Changed, Failed and Unparseable. Unchanged and LocalGone are observations a
+// caller makes, not states Run detects: comparing content is out of scope here
+// (see the size-based contract note above) and Run never deletes a cursor, so
+// both stay zero unless a watcher fills them. They exist so a watcher and a
+// one-shot importer report the same names.
+type Report struct {
+	Scanned     int          // files walked and offered to Parse
+	Ingested    int          // units persisted by this run (or planned, in dry-run)
+	Deduped     int          // units the server reported as replays
+	Unchanged   int          // reserved: files observed as already ingested
+	Changed     int          // files that had at least one unit accepted
+	LocalGone   int          // reserved: cursor records whose file disappeared
+	Failed      int          // files degraded rather than aborted
+	Unparseable int          // readable lines that yielded no unit
+	Failures    map[Code]int // per-code tally, including cursor degradation
+}
+
+// Add folds another report into this one, for callers that run several batches
+// and report once.
+func (r *Report) Add(other Report) {
+	r.Scanned += other.Scanned
+	r.Ingested += other.Ingested
+	r.Deduped += other.Deduped
+	r.Unchanged += other.Unchanged
+	r.Changed += other.Changed
+	r.LocalGone += other.LocalGone
+	r.Failed += other.Failed
+	r.Unparseable += other.Unparseable
+	for code, n := range other.Failures {
+		if r.Failures == nil {
+			r.Failures = map[Code]int{}
+		}
+		r.Failures[code] += n
+	}
+}
+
+func (r *Report) fail(code Code) {
+	if r.Failures == nil {
+		r.Failures = map[Code]int{}
+	}
+	r.Failures[code]++
+}
+
+// CanonicalRoot turns any caller-supplied path into the identity that cursors
+// and idempotency keys are derived from. Without it a relative --root makes two
+// different working directories that each contain the same relative source path
+// collide on one cursor, so the second run sees an up-to-date checkpoint and
+// silently ingests nothing.
+//
+// Symlinks are resolved so the same store reached by two spellings shares one
+// identity. A root that does not exist yet keeps its absolute form rather than
+// erroring, which is what Walk's documented "missing root yields no paths, no
+// error" behavior needs.
+func CanonicalRoot(p string) (string, error) {
+	abs, err := filepath.Abs(p)
+	if err != nil {
+		return "", fmt.Errorf("resolve %s: %w", p, err)
+	}
+	if resolved, err := filepath.EvalSymlinks(abs); err == nil {
+		return resolved, nil
+	}
+	return abs, nil
+}
+
+// Walk collects candidate files under base in lexical order, so cursor
+// high-water marks mean the same thing from one run to the next. Go's
+// filepath.Glob does not treat ** as recursive, hence the explicit walk.
+// Unreadable entries are skipped rather than failing the walk: a session store
+// routinely contains directories the caller cannot enter.
+//
+// base is canonicalized, so every returned path is absolute and cursor identity
+// cannot depend on the caller's working directory.
+//
+// A base that does not exist yields no paths and no error, which is what the
+// existing connector does with it (it reports "no transcripts matched").
+// Whether a missing root is worth reporting is therefore the call site's
+// decision; CodeRootMissing exists for the sites that report it, such as a
+// watch daemon that must not sit idle on a typo'd path.
+//
+// accept is called with each candidate path; a nil accept takes every file.
+func Walk(base string, accept func(abs string) bool) ([]string, error) {
+	root, err := CanonicalRoot(base)
+	if err != nil {
+		return nil, err
+	}
+	if fi, err := os.Stat(root); err == nil && !fi.IsDir() {
+		return []string{root}, nil
+	}
+	var paths []string
+	err = filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
+		if err != nil {
+			return nil
+		}
+		if d.IsDir() {
+			return nil
+		}
+		if accept == nil || accept(p) {
+			paths = append(paths, p)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(paths)
+	return paths, nil
+}
+
+// CursorPath returns the cursor file for one source path, keyed by a hash of
+// its canonical absolute path (see CanonicalRoot) so any source layout is safe
+// to store in one directory.
+func CursorPath(stateDir, abs string) string {
+	sum := sha1.Sum([]byte(abs))
+	return filepath.Join(stateDir, hex.EncodeToString(sum[:])+".json")
+}
+
+// LoadCursor reads a cursor. A missing or undecodable cursor yields LastLine 0,
+// meaning "nothing ingested yet", and is never a hard error: broken state must
+// not block ingestion. Load reports that case through Cursor.Corrupt instead.
+//
+// A file that is now smaller than when its cursor was written was truncated
+// and rewritten, so LastLine resets; otherwise new content at already-ingested
+// line numbers would be skipped forever.
+func LoadCursor(stateDir, abs string) Cursor {
+	var cp Cursor
+	b, err := os.ReadFile(CursorPath(stateDir, abs))
+	if err != nil {
+		return cp
+	}
+	if err := json.Unmarshal(b, &cp); err != nil {
+		return Cursor{Abs: abs, Corrupt: true}
+	}
+	if cp.Abs == "" {
+		cp.Abs = abs
+	}
+	if cp.Size > 0 {
+		if fi, err := os.Stat(abs); err == nil && fi.Size() < cp.Size {
+			cp.LastLine = 0
+		}
+	}
+	return cp
+}
+
+// SaveCursor writes a cursor through a temporary file and an atomic rename, so
+// an interrupted run cannot leave a half-written checkpoint behind. Each save
+// gets its own temporary name: two runs may reach the same cursor concurrently
+// and a shared name would interleave their writes before the rename.
+//
+// A save never rewinds a checkpoint that is already further along. A stored
+// cursor with a smaller Size is the truncation signal LoadCursor resets on, so
+// that case must still write; anything else that is ahead stays.
+//
+// The per-cursor OS-backed lock covers the read/merge/write sequence so
+// independent ingest processes cannot move LastLine backwards or share a
+// staging path. Errors are returned (callers may warn without failing the
+// whole ingest).
+func SaveCursor(stateDir string, cp Cursor) (err error) {
+	p := CursorPath(stateDir, cp.Abs)
+	dir := filepath.Dir(p)
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+		return fmt.Errorf("create checkpoint dir: %w", err)
+	}
+	lock, err := acquireCursorLock(p)
+	if err != nil {
+		return fmt.Errorf("lock cursor: %w", err)
+	}
+	defer func() {
+		if releaseErr := lock.release(); err == nil && releaseErr != nil {
+			err = fmt.Errorf("release cursor lock: %w", releaseErr)
+		}
+	}()
+
+	return saveCursorLocked(stateDir, p, cp)
+}
+
+// saveCursorLocked commits cp while the caller owns p's cursor lock. Keeping
+// this small inner operation separate lets the lock span the current-cursor
+// read as well as the atomic replacement.
+func saveCursorLocked(stateDir, p string, cp Cursor) error {
+	current := LoadCursor(stateDir, cp.Abs)
+	if current.LastLine > cp.LastLine && current.Size <= cp.Size {
+		return nil
+	}
+	b, err := json.Marshal(cp)
+	if err != nil {
+		return fmt.Errorf("encode checkpoint: %w", err)
+	}
+	tmp, err := os.CreateTemp(filepath.Dir(p), ".cursor-*.tmp")
+	if err != nil {
+		return fmt.Errorf("create checkpoint: %w", err)
+	}
+	tmpName := tmp.Name()
+	defer func() {
+		_ = tmp.Close()
+		_ = os.Remove(tmpName)
+	}()
+	if err := tmp.Chmod(0o600); err != nil {
+		return fmt.Errorf("secure checkpoint staging file: %w", err)
+	}
+	if _, err := tmp.Write(b); err != nil {
+		return fmt.Errorf("write checkpoint: %w", err)
+	}
+	if err := tmp.Sync(); err != nil {
+		return fmt.Errorf("sync checkpoint staging file: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return fmt.Errorf("close checkpoint: %w", err)
+	}
+	if err := os.Rename(tmpName, p); err != nil {
+		return fmt.Errorf("commit checkpoint: %w", err)
+	}
+	return nil
+}
+
+// FileState returns the size and UTC mtime recorded alongside a cursor. Both
+// are diagnostics: only size participates in the change decision.
+func FileState(abs string) (size int64, mtime string, err error) {
+	fi, err := os.Stat(abs)
+	if err != nil {
+		return 0, "", err
+	}
+	return fi.Size(), fi.ModTime().UTC().Format("2006-01-02T15:04:05Z07:00"), nil
+}
+
+// Classify maps an Upload or Parse failure onto the shared vocabulary. Callers
+// use it to report a code without re-deriving it from statuses, and exit-code
+// mapping stays in the command surface that owns it.
+func Classify(err error) Code {
+	if err == nil {
+		return ""
+	}
+	var ae *apiclient.APIError
+	if errors.As(err, &ae) {
+		switch ae.Kind() {
+		case apiclient.KindAuth:
+			return CodeAuth
+		case apiclient.KindPlan, apiclient.KindQuota:
+			return CodePlanQuota
+		case apiclient.KindProvider, apiclient.KindTimeout:
+			return CodeProviderTimeout
+		case apiclient.KindConflict, apiclient.KindBadInput:
+			return CodeUploadRejected
+		}
+	}
+	switch {
+	case errors.Is(err, ErrDegradeFile):
+		return CodeUploadRejected
+	// The local-file cases must come before any net.Error-shaped probe:
+	// syscall.Errno implements Timeout and Temporary, so the *fs.PathError a
+	// failed open or stat returns satisfies net.Error, and treating it as a
+	// transport failure would hide which local path was unreadable or gone.
+	case errors.Is(err, os.ErrPermission):
+		return CodeReadDenied
+	case errors.Is(err, os.ErrNotExist):
+		return CodeRootMissing
+	}
+	// Anything else is a request that never reached the server, including a
+	// transport error and a deadline exceeded mid-call.
+	return CodeNetwork
+}
+
+// Run walks the given paths, parses each against its cursor, uploads units
+// through upload, and persists cursors for the files it actually wrote.
+//
+// A parse or unit error other than ErrDegradeFile ends the run: the report
+// tallies its classified code and the error is returned as-is, so the caller
+// keeps its own error surface. An ErrDegradeFile error ends the current file
+// only; its cursor stays put and the report records the failure.
+func Run(ctx context.Context, paths []string, opts Options, parse ParseFunc, upload UploadFunc) (Report, error) {
+	var report Report
+	remaining := opts.Limit
+
+	for _, abs := range paths {
+		report.Scanned++
+
+		cp := LoadCursor(opts.StateDir, abs)
+		if cp.Corrupt {
+			report.fail(CodeStateCorrupt)
+		}
+		units, unparseable, err := parse(abs, cp.LastLine)
+		report.Unparseable += unparseable
+		if err != nil {
+			report.fail(Classify(err))
+			return report, err
+		}
+
+		newLast := cp.LastLine
+		moved := false
+		for _, u := range units {
+			if opts.Limit > 0 && remaining <= 0 {
+				break
+			}
+			// Parse already skips at or below the cursor; this guard keeps an
+			// inconsistent parser from rewinding a cursor.
+			if u.Line <= newLast {
+				continue
+			}
+
+			if opts.DryRun {
+				report.Ingested++
+				if remaining > 0 {
+					remaining--
+				}
+				continue
+			}
+
+			outcome, err := upload(ctx, abs, u)
+			if err != nil {
+				if errors.Is(err, ErrDegradeFile) {
+					report.Failed++
+					report.fail(CodeUploadRejected)
+					break
+				}
+				report.fail(Classify(err))
+				return report, err
+			}
+			if outcome.Deduplicated {
+				report.Deduped++
+			} else {
+				report.Ingested++
+			}
+			if remaining > 0 {
+				remaining--
+			}
+			newLast = u.Line
+			moved = true
+		}
+		if moved {
+			report.Changed++
+		}
+
+		if opts.DryRun {
+			continue
+		}
+		if size, mtime, err := FileState(abs); err == nil {
+			if err := SaveCursor(opts.StateDir, Cursor{
+				Abs:      abs,
+				Size:     size,
+				ModTime:  mtime,
+				LastLine: newLast,
+			}); err != nil {
+				if opts.Log != nil {
+					opts.Log("warn: save checkpoint for %s: %v\n", abs, err)
+				}
+				report.fail(CodeStateCorrupt)
+			}
+		}
+	}
+	return report, nil
+}
+
+// HasJSONLExtension reports whether a path looks like a JSON-lines file,
+// ignoring case. It is the accept predicate the transcript connectors use.
+func HasJSONLExtension(p string) bool {
+	return strings.EqualFold(filepath.Ext(p), ".jsonl")
+}
diff --git a/server/internal/ingest/ingest_test.go b/server/internal/ingest/ingest_test.go
new file mode 100644
index 0000000..3cbaaeb
--- /dev/null
+++ b/server/internal/ingest/ingest_test.go
@@ -0,0 +1,463 @@
+package ingest
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/PeterGuy326/mem/server/internal/apiclient"
+)
+
+func writeSource(t *testing.T, dir, name, content string) string {
+	t.Helper()
+	p := filepath.Join(dir, name)
+	if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(p, []byte(content), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return p
+}
+
+// unitsForLines builds a parser that yields one unit per line above the cursor.
+func unitsForLines(lines int) ParseFunc {
+	return func(abs string, skipBefore int) ([]Unit, int, error) {
+		var units []Unit
+		for line := skipBefore + 1; line <= lines; line++ {
+			units = append(units, Unit{Line: line, Body: fmt.Sprintf("%s#%d", filepath.Base(abs), line)})
+		}
+		return units, 0, nil
+	}
+}
+
+func TestRunDryRunWritesNothing(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	abs := writeSource(t, src, "a.jsonl", "x")
+
+	var uploads int
+	report, err := Run(context.Background(), []string{abs}, Options{StateDir: states, DryRun: true},
+		unitsForLines(3),
+		func(context.Context, string, Unit) (Outcome, error) {
+			uploads++
+			return Outcome{}, errors.New("dry-run must not upload")
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if uploads != 0 {
+		t.Fatalf("uploads = %d, want 0 (dry-run performed writes)", uploads)
+	}
+	if report.Ingested != 3 || report.Scanned != 1 {
+		t.Fatalf("report = %+v, want 3 planned units in 1 scanned file", report)
+	}
+	entries, err := os.ReadDir(states)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("dry-run left cursor state behind: %v", entries)
+	}
+}
+
+func TestRunRespectsLimitAndReportsDedupSeparately(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	abs := writeSource(t, src, "a.jsonl", "x")
+
+	var uploaded []int
+	report, err := Run(context.Background(), []string{abs}, Options{StateDir: states, Limit: 2},
+		unitsForLines(5),
+		func(_ context.Context, _ string, u Unit) (Outcome, error) {
+			uploaded = append(uploaded, u.Line)
+			// The second unit is a server-reported replay.
+			return Outcome{Deduplicated: u.Line == 2}, nil
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(uploaded) != 2 || uploaded[0] != 1 || uploaded[1] != 2 {
+		t.Fatalf("uploaded lines = %v, want [1 2]", uploaded)
+	}
+	if report.Ingested != 1 || report.Deduped != 1 {
+		t.Fatalf("report = %+v, want 1 ingested and 1 deduped", report)
+	}
+	cp := LoadCursor(states, abs)
+	if cp.LastLine != 2 {
+		t.Fatalf("cursor LastLine = %d, want 2", cp.LastLine)
+	}
+}
+
+func TestRunDegradesOneFileAndContinuesOthers(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	a := writeSource(t, src, "a.jsonl", "x")
+	b := writeSource(t, src, "b.jsonl", "x")
+
+	// Seed a's cursor at line 1 so the conflict starts from line 2.
+	if err := SaveCursor(states, Cursor{Abs: a, Size: 1, ModTime: "2026-01-01T00:00:00Z", LastLine: 1}); err != nil {
+		t.Fatal(err)
+	}
+
+	var uploaded []string
+	report, err := Run(context.Background(), []string{a, b}, Options{StateDir: states},
+		unitsForLines(3),
+		func(_ context.Context, abs string, u Unit) (Outcome, error) {
+			uploaded = append(uploaded, fmt.Sprintf("%s:%d", filepath.Base(abs), u.Line))
+			if abs == a {
+				return Outcome{}, fmt.Errorf("%w: %s:%d", ErrDegradeFile, abs, u.Line)
+			}
+			return Outcome{}, nil
+		})
+	if err != nil {
+		t.Fatalf("a degraded file must not abort the run: %v", err)
+	}
+	want := []string{"a.jsonl:2", "b.jsonl:1", "b.jsonl:2", "b.jsonl:3"}
+	if strings.Join(uploaded, ",") != strings.Join(want, ",") {
+		t.Fatalf("uploads = %v, want %v", uploaded, want)
+	}
+	if report.Failed != 1 || report.Failures[CodeUploadRejected] != 1 {
+		t.Fatalf("report = %+v, want one upload_rejected failure", report)
+	}
+	if got := LoadCursor(states, a).LastLine; got != 1 {
+		t.Fatalf("degraded file cursor LastLine = %d, want 1 (retry must stay meaningful)", got)
+	}
+	if got := LoadCursor(states, b).LastLine; got != 3 {
+		t.Fatalf("healthy file cursor LastLine = %d, want 3", got)
+	}
+}
+
+func TestRunAbortsOnUnclassifiedUploadError(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	abs := writeSource(t, src, "a.jsonl", "x")
+
+	authErr := &apiclient.APIError{StatusCode: http.StatusUnauthorized}
+	_, err := Run(context.Background(), []string{abs}, Options{StateDir: states},
+		unitsForLines(3),
+		func(context.Context, string, Unit) (Outcome, error) {
+			return Outcome{}, authErr
+		})
+	if !errors.Is(err, authErr) {
+		t.Fatalf("err = %v, want the caller's error returned unchanged", err)
+	}
+	if _, err := os.Stat(CursorPath(states, abs)); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("aborted run must not advance the cursor, stat err = %v", err)
+	}
+}
+
+func TestCorruptCursorDegradesWithoutBlocking(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	abs := writeSource(t, src, "a.jsonl", "x")
+
+	p := CursorPath(states, abs)
+	if err := os.WriteFile(p, []byte("{not json"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	cp := LoadCursor(states, abs)
+	if !cp.Corrupt || cp.LastLine != 0 {
+		t.Fatalf("cursor = %+v, want Corrupt with a reset high-water mark", cp)
+	}
+
+	report, err := Run(context.Background(), []string{abs}, Options{StateDir: states},
+		unitsForLines(2),
+		func(context.Context, string, Unit) (Outcome, error) { return Outcome{}, nil })
+	if err != nil {
+		t.Fatalf("a corrupt cursor must not block the run: %v", err)
+	}
+	if report.Failures[CodeStateCorrupt] != 1 {
+		t.Fatalf("report = %+v, want one state_corrupt failure", report)
+	}
+	if report.Ingested != 2 {
+		t.Fatalf("report.Ingested = %d, want 2 (re-planned from line 1)", report.Ingested)
+	}
+}
+
+func TestLoadCursorResetsWhenFileShrank(t *testing.T) {
+	src := t.TempDir()
+	states := t.TempDir()
+	abs := writeSource(t, src, "a.jsonl", strings.Repeat("x", 20))
+
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 999, ModTime: "2026-01-01T00:00:00Z", LastLine: 7}); err != nil {
+		t.Fatal(err)
+	}
+	if got := LoadCursor(states, abs).LastLine; got != 0 {
+		t.Fatalf("LastLine = %d, want 0 after the file shrank below the recorded size", got)
+	}
+
+	// Growing the file keeps the high-water mark.
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 1, ModTime: "2026-01-01T00:00:00Z", LastLine: 7}); err != nil {
+		t.Fatal(err)
+	}
+	if got := LoadCursor(states, abs).LastLine; got != 7 {
+		t.Fatalf("LastLine = %d, want 7 while the file only grew", got)
+	}
+}
+
+// TestCursorOnDiskFormat pins the persisted layout: existing users' cursors
+// must stay readable, so the key names and their order are part of the contract.
+func TestCursorOnDiskFormat(t *testing.T) {
+	states := t.TempDir()
+	abs := filepath.Join(t.TempDir(), "a.jsonl")
+	want := `{"abs":"` + abs + `","size":12,"mtime":"2026-08-30T06:14:01Z","last_line":3}`
+
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 12, ModTime: "2026-08-30T06:14:01Z", LastLine: 3}); err != nil {
+		t.Fatal(err)
+	}
+	b, err := os.ReadFile(CursorPath(states, abs))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(b) != want {
+		t.Fatalf("cursor bytes =\n%s\nwant\n%s", b, want)
+	}
+	fi, err := os.Stat(CursorPath(states, abs))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if perm := fi.Mode().Perm(); perm != 0o600 {
+		t.Fatalf("cursor mode = %o, want 600", perm)
+	}
+}
+
+// TestWalkCanonicalizesRelativeBase pins the identity rule: every path Walk
+// returns is absolute and symlink-resolved, so a caller-supplied relative root
+// cannot make two working directories share one cursor.
+func TestWalkCanonicalizesRelativeBase(t *testing.T) {
+	store := t.TempDir()
+	writeSource(t, store, "sessions/a.jsonl", "x")
+	absBase, err := CanonicalRoot(store)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Chdir(store)
+	got, err := Walk("sessions", HasJSONLExtension)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := filepath.Join(absBase, "sessions", "a.jsonl")
+	if len(got) != 1 || got[0] != want {
+		t.Fatalf("walk returned %v, want [%s]", got, want)
+	}
+
+	// A same-named source in another working directory is a different identity,
+	// so the two never share a cursor file.
+	storeB := t.TempDir()
+	writeSource(t, storeB, "sessions/a.jsonl", "x")
+	t.Chdir(storeB)
+	second, err := Walk("sessions", HasJSONLExtension)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(second) != 1 || second[0] == want {
+		t.Fatalf("relative bases from two directories collided on %v", second)
+	}
+	if CursorPath("states", want) == CursorPath("states", second[0]) {
+		t.Fatalf("cursor keys collide for %s and %s", want, second[0])
+	}
+
+	// A root that does not exist yet stays absolute instead of erroring, which
+	// is what the missing-root contract above needs.
+	absent := filepath.Join(store, "nope", "deep")
+	resolved, err := CanonicalRoot(absent)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !filepath.IsAbs(resolved) {
+		t.Fatalf("CanonicalRoot of an absent path = %q, want absolute", resolved)
+	}
+	t.Chdir(store)
+	if paths, err := Walk("nope/deep", HasJSONLExtension); err != nil || len(paths) != 0 {
+		t.Fatalf("missing root: paths = %v, err = %v; want none, no error", paths, err)
+	}
+}
+
+func TestSaveCursorKeepsCommittedProgressAndLeavesNoTempFile(t *testing.T) {
+	states := t.TempDir()
+	abs := filepath.Join(t.TempDir(), "a.jsonl")
+
+	// A run that read the file earlier must not rewind one that finished first.
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 200, ModTime: "2026-08-30T06:14:01Z", LastLine: 10}); err != nil {
+		t.Fatal(err)
+	}
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 200, ModTime: "2026-08-30T06:14:02Z", LastLine: 7}); err != nil {
+		t.Fatal(err)
+	}
+	if got := LoadCursor(states, abs); got.LastLine != 10 || got.Size != 200 {
+		t.Fatalf("cursor = %+v, want the committed line 10 kept", got)
+	}
+
+	// A rewrite that shrank the file is the one case allowed to rewind.
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 40, ModTime: "2026-08-30T06:14:03Z", LastLine: 2}); err != nil {
+		t.Fatal(err)
+	}
+	if got := LoadCursor(states, abs).LastLine; got != 2 {
+		t.Fatalf("LastLine = %d, want 2 after the source shrank", got)
+	}
+
+	entries, err := os.ReadDir(states)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(entries) != 2 {
+		t.Fatalf("state dir = %v, want cursor file and lock sidecar", entries)
+	}
+	var hasJSON, hasLock bool
+	for _, e := range entries {
+		if strings.HasSuffix(e.Name(), ".json") {
+			hasJSON = true
+		}
+		if strings.HasSuffix(e.Name(), ".json.lock") {
+			hasLock = true
+		}
+	}
+	if !hasJSON || !hasLock {
+		t.Fatalf("state dir = %v, want .json cursor and .json.lock sidecar", entries)
+	}
+}
+
+// TestSaveCursorDoesNotStageInASharedSlot pins the temporary naming. Reusing
+// <cursor>.tmp gives every process writing that cursor the same staging file,
+// so one run's write can land inside another's rename.
+func TestSaveCursorDoesNotStageInASharedSlot(t *testing.T) {
+	states := t.TempDir()
+	abs := filepath.Join(t.TempDir(), "a.jsonl")
+	leftover := CursorPath(states, abs) + ".tmp"
+	if err := os.WriteFile(leftover, []byte("another run's staging file"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := SaveCursor(states, Cursor{Abs: abs, Size: 20, ModTime: "2026-08-30T06:14:01Z", LastLine: 4}); err != nil {
+		t.Fatal(err)
+	}
+	if got := LoadCursor(states, abs).LastLine; got != 4 {
+		t.Fatalf("LastLine = %d, want 4", got)
+	}
+	if b, err := os.ReadFile(leftover); err != nil || string(b) != "another run's staging file" {
+		t.Fatalf("save consumed the shared staging file: %q, err %v", b, err)
+	}
+}
+
+func TestConcurrentSaveCursorPublishesWholeCursors(t *testing.T) {
+	states := t.TempDir()
+	abs := filepath.Join(t.TempDir(), "a.jsonl")
+
+	var wg sync.WaitGroup
+	for i := 1; i <= 8; i++ {
+		wg.Add(1)
+		go func(line int) {
+			defer wg.Done()
+			if err := SaveCursor(states, Cursor{Abs: abs, Size: 100, ModTime: "2026-08-30T06:14:01Z", LastLine: line}); err != nil {
+				t.Error(err)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	b, err := os.ReadFile(CursorPath(states, abs))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var cp Cursor
+	if err := json.Unmarshal(b, &cp); err != nil {
+		t.Fatalf("cursor published half-written: %v (%s)", err, b)
+	}
+	if cp.LastLine < 1 || cp.LastLine > 8 {
+		t.Fatalf("cursor = %+v", cp)
+	}
+	if entries, err := os.ReadDir(states); err != nil || len(entries) != 2 {
+		t.Fatalf("state dir = %v, err = %v; want cursor file and lock sidecar", entries, err)
+	}
+}
+
+func TestWalkIsDeterministicAndSkipsUnreadable(t *testing.T) {
+	root := t.TempDir()
+	for _, name := range []string{"c.jsonl", "nested/b.jsonl", "nested/deep/a.jsonl", "ignore.txt"} {
+		writeSource(t, root, name, "x")
+	}
+	got, err := Walk(root, HasJSONLExtension)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 3 {
+		t.Fatalf("walk returned %d paths, want 3 (*.jsonl only): %v", len(got), got)
+	}
+	if !sort.StringsAreSorted(got) {
+		t.Fatalf("walk order must be lexical for stable cursors: %v", got)
+	}
+
+	// A base naming one existing file is accepted as a one-off source.
+	one := got[0]
+	single, err := Walk(one, HasJSONLExtension)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(single) != 1 || single[0] != one {
+		t.Fatalf("single-file base returned %v, want [%s]", single, one)
+	}
+
+	// A missing root is not an error today: the connector reports "no
+	// transcripts matched" instead. Pin that, because surfacing it as a
+	// failure here would change `mem ingest qoder`'s observable behaviour.
+	missing, err := Walk(filepath.Join(root, "nope"), HasJSONLExtension)
+	if err != nil || len(missing) != 0 {
+		t.Fatalf("missing root: paths = %v, err = %v; want none, no error", missing, err)
+	}
+}
+
+func TestClassifyCoversSharedCodes(t *testing.T) {
+	cases := []struct {
+		err  error
+		want Code
+	}{
+		{&apiclient.APIError{StatusCode: http.StatusForbidden}, CodeAuth},
+		{&apiclient.APIError{StatusCode: http.StatusPaymentRequired}, CodePlanQuota},
+		{&apiclient.APIError{StatusCode: http.StatusTooManyRequests}, CodePlanQuota},
+		{&apiclient.APIError{StatusCode: http.StatusServiceUnavailable}, CodeProviderTimeout},
+		{&apiclient.APIError{StatusCode: http.StatusGatewayTimeout}, CodeProviderTimeout},
+		{&apiclient.APIError{StatusCode: http.StatusConflict}, CodeUploadRejected},
+		{fmt.Errorf("wrapped: %w", ErrDegradeFile), CodeUploadRejected},
+		{&os.PathError{Op: "open", Err: os.ErrPermission}, CodeReadDenied},
+		{&os.PathError{Op: "stat", Err: os.ErrNotExist}, CodeRootMissing},
+		{nil, ""},
+	}
+	for _, tc := range cases {
+		if got := Classify(tc.err); got != tc.want {
+			t.Errorf("Classify(%v) = %q, want %q", tc.err, got, tc.want)
+		}
+	}
+
+	// The rows above build PathErrors around sentinel errors. A real failed open
+	// carries a syscall.Errno instead, and Errno implements net.Error, so only
+	// this shape reproduces a file error being reported as a transport failure.
+	if _, err := os.Open(filepath.Join(t.TempDir(), "absent.jsonl")); err != nil {
+		if got := Classify(fmt.Errorf("open transcript: %w", err)); got != CodeRootMissing {
+			t.Errorf("Classify(real open failure) = %q, want %q", got, CodeRootMissing)
+		}
+	} else {
+		t.Fatal("opening a file that does not exist succeeded")
+	}
+}
+
+func TestReportAddAggregatesAcrossCycles(t *testing.T) {
+	var total Report
+	total.Add(Report{Scanned: 2, Ingested: 5, Deduped: 1, Failed: 1, Failures: map[Code]int{CodeUploadRejected: 1}})
+	total.Add(Report{Scanned: 1, Ingested: 2, Unchanged: 1, Failures: map[Code]int{CodeStateCorrupt: 1}})
+	if total.Scanned != 3 || total.Ingested != 7 || total.Deduped != 1 || total.Failed != 1 {
+		t.Fatalf("total = %+v", total)
+	}
+	if total.Failures[CodeUploadRejected] != 1 || total.Failures[CodeStateCorrupt] != 1 {
+		t.Fatalf("failure tally = %+v", total.Failures)
+	}
+}