engine: idempotent ingest + self-diagnosing not-found (HAL-323)

Two-part fix for flaky bulk ingestion and the bench's duplicate uploads.

Idempotent ingest (root cause of "same doc ingested up to 6x"):
- New `idempotency_key` column on documents + partial unique index on
  (org_id, idempotency_key) — migration 0007. NULL key = no dedup
  (preserves current behavior for callers that don't opt in).
- All three ingest handlers (standalone/local in internal/api, multi-tenant
  REST in internal/handler, Connect RPC in internal/connecthandler) honor the
  `Idempotency-Key` header the SDK already sends: a repeat ingest returns the
  original document instead of creating a duplicate, and the concurrent
  same-key race is resolved by dropping the orphan source and returning the
  winner. The SDK retry-on-transient (which re-POSTs after a reset that landed
  AFTER the server committed) is now a no-op.
- db: ErrConflict sentinel mapped from SQLSTATE 23505; GetDocumentByIdempotencyKey.

Self-diagnosing not-found (resolves the pathless "object not found" mystery):
- Every storage backend's ErrNotFound now carries the resolved key/path
  (local Get+Delete, s3 Get+Delete, gcs Get+Delete) so a failure can always be
  attributed to a code path — pathless rows were pre-path-wrap leftovers.
- local.Get: the open-after-stat result is wrapped as retryable ErrNotFound
  (previously a raw error getSourceWithRetry would NOT retry), and a Windows-only
  bounded retry rides through the Defender scan window that transiently hides a
  freshly-written file.

Docs: OpenAPI documents the Idempotency-Key header; README documents the
mandatory Windows Defender exclusion for the storage root.

Verified: go build/vet clean; storage + db (live-Postgres) tests pass; and an
end-to-end double-POST with one Idempotency-Key returns one document_id while a
keyless control gets its own.

Author: hallelx2

README.md

@@ -164,6 +164,15 @@ llm:
 
 > **Anthropic-compatible gateways (GLM/Z.ai):** `base_url` **must include `/v1`** — the client posts to `${base_url}/messages`.
 
+> **Windows + local storage:** Windows Defender real-time protection scans
+> each freshly-written file and briefly hides it from `os.Stat`/`os.Open`,
+> which under heavy concurrent ingestion can surface as transient
+> `object not found` errors. The local backend rides through this window with
+> a short internal retry, but for large bulk loads **add a Defender exclusion
+> for your storage root** (`local.root` / `VLE_STORAGE_LOCAL_ROOT`):
+> `Add-MpPreference -ExclusionPath "C:\path\to\data\documents"`. Linux has no
+> such scan-hold and needs no exclusion.
+
 ### Supported formats
 
 PDF (positioned text + tables via [`pdftable`](https://github.com/hallelx2/pdftable)) · Markdown · HTML · DOCX · Text.

internal/api/server.go

@@ -231,6 +231,28 @@ func (d Deps) handleIngestDocument(w http.ResponseWriter, r *http.Request) {
 
 	docID := ingest.NewDocumentID()
 
+	// Idempotent ingest: if the caller supplied an Idempotency-Key and a
+	// document already exists under it (for the standalone org), return that
+	// document instead of creating a duplicate. A client — or the SDK's own
+	// transport retry-on-transient — that re-POSTs the same upload after a
+	// reset that landed AFTER the server committed the row is then a no-op.
+	// This is the standalone/local-mode half of the HAL-323 duplicate-upload
+	// fix (the multi-tenant handler in internal/handler mirrors it).
+	idemKey := r.Header.Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := d.DB.GetDocumentByIdempotencyKey(ctx, standaloneOrgID, idemKey); err == nil {
+			writeJSON(w, http.StatusAccepted, map[string]any{
+				"document_id": existing.ID,
+				"status":      string(existing.Status),
+			})
+			return
+		} else if !errors.Is(err, db.ErrNotFound) {
+			d.Logger.Error("ingest: idempotency lookup failed", "err", err)
+			writeErr(w, http.StatusInternalServerError, "db read failed")
+			return
+		}
+	}
+
 	var (
 		filename    string
 		contentType string
@@ -300,14 +322,27 @@ func (d Deps) handleIngestDocument(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err := d.DB.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       standaloneOrgID,
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          standaloneOrgID,
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// A concurrent same-key ingest won the race: drop the orphan source we
+		// just wrote and return the winner instead of a 500.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			_ = d.Storage.Delete(ctx, key)
+			if existing, lookupErr := d.DB.GetDocumentByIdempotencyKey(ctx, standaloneOrgID, idemKey); lookupErr == nil {
+				writeJSON(w, http.StatusAccepted, map[string]any{
+					"document_id": existing.ID,
+					"status":      string(existing.Status),
+				})
+				return
+			}
+		}
 		d.Logger.Error("ingest: db insert failed", "err", err)
 		writeErr(w, http.StatusInternalServerError, "db write failed")
 		return

internal/connecthandler/documents.go

@@ -82,6 +82,23 @@ func (s *DocumentsService) CreateDocument(
 		return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("content is required"))
 	}
 
+	// Idempotent ingest: a repeat CreateDocument carrying the same
+	// Idempotency-Key returns the original document rather than a duplicate
+	// (mirrors the REST path; closes the HAL-323 duplicate-upload bug for SDK
+	// callers that retry the RPC).
+	idemKey := req.Header().Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := s.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); err == nil {
+			return connect.NewResponse(&v1.CreateDocumentResponse{
+				DocumentId: string(existing.ID),
+				Status:     string(existing.Status),
+			}), nil
+		} else if !errors.Is(err, db.ErrNotFound) {
+			s.logger.Error("ingest: idempotency lookup failed", "err", err)
+			return nil, connect.NewError(connect.CodeInternal, errors.New("db read failed"))
+		}
+	}
+
 	docID := ingest.NewDocumentID()
 	contentType := msg.ContentType
 	if contentType == "" {
@@ -105,15 +122,27 @@ func (s *DocumentsService) CreateDocument(
 	}
 
 	if err := s.db.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       orgID,
-		StoreID:     storeIDFromConnect(req),
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          orgID,
+		StoreID:        storeIDFromConnect(req),
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// Concurrent same-key insert won the race: drop the orphan source and
+		// return the winner instead of erroring.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			_ = s.storage.Delete(ctx, key)
+			if existing, lookupErr := s.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); lookupErr == nil {
+				return connect.NewResponse(&v1.CreateDocumentResponse{
+					DocumentId: string(existing.ID),
+					Status:     string(existing.Status),
+				}), nil
+			}
+		}
 		s.logger.Error("ingest: db insert failed", "err", err)
 		return nil, connect.NewError(connect.CodeInternal, errors.New("db write failed"))
 	}

internal/handler/documents.go

@@ -158,6 +158,25 @@ func (h *DocumentsHandler) HandleIngestDocument(w http.ResponseWriter, r *http.R
 	ctx := r.Context()
 	docID := ingest.NewDocumentID()
 
+	// Idempotent ingest: if the client supplied an Idempotency-Key and we
+	// already have a document for this org under that key, return it instead
+	// of creating a duplicate. This makes a client/transport retry of the
+	// ingest POST (the HAL-323 duplicate-upload bug) a no-op.
+	idemKey := r.Header.Get("Idempotency-Key")
+	if idemKey != "" {
+		if existing, err := h.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); err == nil {
+			writeJSON(w, http.StatusAccepted, map[string]any{
+				"document_id": existing.ID,
+				"status":      string(existing.Status),
+			})
+			return
+		} else if !errors.Is(err, db.ErrNotFound) {
+			h.logger.Error("ingest: idempotency lookup failed", "err", err)
+			writeErr(w, http.StatusInternalServerError, "db read failed")
+			return
+		}
+	}
+
 	var (
 		filename    string
 		contentType string
@@ -227,15 +246,29 @@ func (h *DocumentsHandler) HandleIngestDocument(w http.ResponseWriter, r *http.R
 	}
 
 	if err := h.db.NewDocument(ctx, db.Document{
-		ID:          docID,
-		OrgID:       orgID,
-		StoreID:     storeID(r),
-		Title:       title,
-		ContentType: contentType,
-		SourceRef:   key,
-		Status:      db.StatusPending,
-		ByteSize:    size,
+		ID:             docID,
+		OrgID:          orgID,
+		StoreID:        storeID(r),
+		Title:          title,
+		ContentType:    contentType,
+		SourceRef:      key,
+		Status:         db.StatusPending,
+		ByteSize:       size,
+		IdempotencyKey: idemKey,
 	}); err != nil {
+		// A concurrent request with the same Idempotency-Key won the race and
+		// inserted first. Drop the orphan source we just wrote and return the
+		// winner's document rather than a 500.
+		if idemKey != "" && errors.Is(err, db.ErrConflict) {
+			_ = h.storage.Delete(ctx, key)
+			if existing, lookupErr := h.db.GetDocumentByIdempotencyKey(ctx, orgID, idemKey); lookupErr == nil {
+				writeJSON(w, http.StatusAccepted, map[string]any{
+					"document_id": existing.ID,
+					"status":      string(existing.Status),
+				})
+				return
+			}
+		}
 		h.logger.Error("ingest: db insert failed", "err", err)
 		writeErr(w, http.StatusInternalServerError, "db write failed")
 		return

openapi.yaml

@@ -113,6 +113,21 @@ paths:
         (field: "file") or JSON body with content. Returns immediately
         with a document_id in "pending" state. The ingest pipeline runs
         asynchronously: parse → build tree → summarize sections.
+
+        Pass an `Idempotency-Key` header to make ingestion idempotent: a
+        repeated request with the same key returns the original document
+        (and its current status) instead of creating a duplicate. Use this
+        to make a client/network retry of the upload safe.
+      parameters:
+        - name: Idempotency-Key
+          in: header
+          required: false
+          description: |
+            Opaque client-chosen key, unique per document. Repeating a
+            request with the same key returns the existing document rather
+            than ingesting a duplicate.
+          schema:
+            type: string
       requestBody:
         required: true
         content:

pkg/db/db.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
@@ -25,6 +26,10 @@ var migrationsFS embed.FS
 // ErrNotFound signals a missing row.
 var ErrNotFound = errors.New("db: not found")
 
+// ErrConflict signals a unique-constraint violation (SQLSTATE 23505), e.g.
+// a second insert with the same (org_id, idempotency_key).
+var ErrConflict = errors.New("db: conflict")
+
 // Pool wraps *pgxpool.Pool with engine-specific helpers.
 type Pool struct {
 	*pgxpool.Pool
@@ -136,5 +141,9 @@ func mapErr(err error) error {
 	if errors.Is(err, pgx.ErrNoRows) {
 		return ErrNotFound
 	}
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+		return fmt.Errorf("%w: %s", ErrConflict, pgErr.ConstraintName)
+	}
 	return err
 }

pkg/db/documents.go

@@ -41,6 +41,12 @@ type Document struct {
 	ErrorMessage string
 	ByteSize     int64
 	Metadata     map[string]string
+
+	// IdempotencyKey is the optional client-supplied Idempotency-Key.
+	// When non-empty it is unique per (org_id, idempotency_key): a repeat
+	// ingest with the same key returns the existing document instead of
+	// creating a duplicate. Empty means "no dedup" (column stored as NULL).
+	IdempotencyKey string
 	CreatedAt    time.Time
 	UpdatedAt    time.Time
 
@@ -76,14 +82,49 @@ func (p *Pool) NewDocument(ctx context.Context, d Document) error {
 	if storeID == "" {
 		storeID = NilScope
 	}
+	// NULL (not "") when no key, so the partial unique index ignores the row.
+	var idemKey any
+	if d.IdempotencyKey != "" {
+		idemKey = d.IdempotencyKey
+	}
 	_, err = p.Exec(ctx, `
-        INSERT INTO documents (id, org_id, store_id, title, content_type, source_ref, status, byte_size, metadata)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
-		string(d.ID), d.OrgID, storeID, d.Title, d.ContentType, d.SourceRef, string(d.Status), d.ByteSize, meta,
+        INSERT INTO documents (id, org_id, store_id, title, content_type, source_ref, status, byte_size, metadata, idempotency_key)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
+		string(d.ID), d.OrgID, storeID, d.Title, d.ContentType, d.SourceRef, string(d.Status), d.ByteSize, meta, idemKey,
 	)
 	return mapErr(err)
 }
 
+// GetDocumentByIdempotencyKey returns the document an org previously ingested
+// under key, or ErrNotFound if none exists. Used to short-circuit a repeat
+// ingest (client retry / SDK transport retry) back to the original document
+// instead of creating a duplicate.
+func (p *Pool) GetDocumentByIdempotencyKey(ctx context.Context, orgID, key string) (*Document, error) {
+	if orgID == "" {
+		return nil, fmt.Errorf("GetDocumentByIdempotencyKey: orgID is required")
+	}
+	if key == "" {
+		return nil, ErrNotFound
+	}
+	row := p.QueryRow(ctx, `
+        SELECT id, org_id, store_id, title, content_type, source_ref, status, error_message,
+               byte_size, metadata, created_at, updated_at, toc_tree
+        FROM documents WHERE org_id = $1 AND idempotency_key = $2`, orgID, key)
+
+	var d Document
+	var status string
+	var rawMeta, rawTOC []byte
+	if err := row.Scan(&d.ID, &d.OrgID, &d.StoreID, &d.Title, &d.ContentType, &d.SourceRef, &status,
+		&d.ErrorMessage, &d.ByteSize, &rawMeta, &d.CreatedAt, &d.UpdatedAt, &rawTOC); err != nil {
+		return nil, mapErr(err)
+	}
+	d.Status = DocumentStatus(status)
+	d.Metadata = unmarshalMeta(rawMeta)
+	d.TOCTree = rawTOC
+	d.IdempotencyKey = key
+	return &d, nil
+}
+
 // GetDocument fetches a document scoped to an org and (optionally) a
 // store. storeID == "" means "don't filter by store" — used by
 // header-less / pre-stores callers. A non-empty storeID restricts the