refactor(ingest): extract shared local ingestion core with concurrent-writer safety

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

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).
 

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)"