diff --git a/README.md b/README.md
index c8ec319b..e197b557 100644
--- a/README.md
+++ b/README.md
@@ -67,8 +67,8 @@ Early-stage, under active development.
 | | |
 |---|---|
 | **Done** | `types`, `proto`, `store`, `sqlite`, `postgres`, `core`, `validate`, `fhirpath`, `registry`, `sync`, `modules`, `view`, `ai`, `auth`, `jobs`, `audit`, `smart`, `subscriptions`, `http`, `runtime`, `client` — CRUD, history, transaction bundles, atomic writes, structural validation, FHIRPath, FHIR definition catalog, device-to-hub push/pull, manifest-driven module installer, ViewDefinition execution, policy-governed AI tool harness, shared identity and policy library, shared job runtime, shared audit event library, optional SMART on FHIR (scopes, tokens, auth adapters), change-triggered workflows with webhook/local delivery, FHIR REST HTTP adapter, runtime composition, and Go client SDK |
-| **Partial** | `cli` (operator surface usable; backup/restore and conflict drill-down remain) |
-| **Next (Stage 1)** | `testkit`, CLI backup/restore and conflict drill-down |
+| **Partial** | `cli` (operator surface usable; conflict drill-down remains) |
+| **Next (Stage 1)** | `testkit`, CLI conflict drill-down |
 
 Durable job and audit persistence is available via `pkg/postgres` and `pkg/sqlite`. Shared runtime helpers live in `pkg/jobs` and `pkg/audit`. See [Roadmap](#roadmap) for the full plan.
 
diff --git a/cmd/haistack/README.md b/cmd/haistack/README.md
index 25b179b8..2d86f63d 100644
--- a/cmd/haistack/README.md
+++ b/cmd/haistack/README.md
@@ -19,7 +19,7 @@ It does **not**:
 - Replace `pkg/runtime` composition logic (commands call the runtime builder)
 - Implement FHIR business rules or persistence (those live in `pkg/*`)
 - Call remote sync hubs for health checks (`sync status` is local-store only)
-- Provide backup/restore or conflict drill-down (planned later)
+- Provide conflict drill-down (planned later)
 
 ## When to use it
 
@@ -71,6 +71,8 @@ haistack serve
 | `haistack serve` | Build `pkg/runtime`, start HTTP, print bound address. |
 | `haistack validate <file>` | Structural validation via `validate.Engine`. Exits non-zero when invalid. |
 | `haistack import <file>` | Import one JSON resource; use `--create-only` or `--update-only` to control conflicts. |
+| `haistack backup [dir]` | Write NDJSON files plus `manifest.json` for stored resources (default dir `backup`). |
+| `haistack restore [dir]` | Reload an NDJSON backup directory with create-or-update. |
 | `haistack read <ResourceType/id>` | Read one stored resource. |
 | `haistack delete <ResourceType/id> --force` | Delete one stored resource. |
 | `haistack export <ResourceType[/id]>` | Export one resource or a resource collection as JSON. |
diff --git a/cmd/haistack/command/backup.go b/cmd/haistack/command/backup.go
new file mode 100644
index 00000000..2c4cad63
--- /dev/null
+++ b/cmd/haistack/command/backup.go
@@ -0,0 +1,74 @@
+package command
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/degoke/health-ai-stack/cmd/haistack/internal/app"
+	"github.com/spf13/cobra"
+)
+
+func newBackupCommand(opts *Options, printer *app.Printer) *cobra.Command {
+	return &cobra.Command{
+		Use:   "backup [dir]",
+		Short: "Write an NDJSON backup of stored FHIR resources",
+		Args:  cobra.MaximumNArgs(1),
+		Example: `  haistack backup
+  haistack backup ./snapshots/today --output json`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			dir := "backup"
+			if len(args) == 1 {
+				dir = args[0]
+			}
+			ctx := context.Background()
+			session, err := openSession(opts, printer, ctx)
+			if err != nil {
+				return err
+			}
+			defer func() { _ = session.Close(ctx) }()
+			report, err := session.Backup(ctx, dir)
+			if err != nil {
+				return exitErr(printer, err)
+			}
+			if printer.Format == app.OutputJSON {
+				return printer.Print(report)
+			}
+			writeStdout(printer, fmt.Sprintf("backed up %d resource type(s) to %s", len(report.Files), report.Directory))
+			for _, file := range report.Files {
+				writeStdout(printer, fmt.Sprintf("  %s %d", file.File, file.Count))
+			}
+			return nil
+		},
+	}
+}
+
+func newRestoreCommand(opts *Options, printer *app.Printer) *cobra.Command {
+	return &cobra.Command{
+		Use:   "restore [dir]",
+		Short: "Restore FHIR resources from an NDJSON backup directory",
+		Args:  cobra.MaximumNArgs(1),
+		Example: `  haistack restore
+  haistack restore ./snapshots/today`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			dir := "backup"
+			if len(args) == 1 {
+				dir = args[0]
+			}
+			ctx := context.Background()
+			session, err := openSession(opts, printer, ctx)
+			if err != nil {
+				return err
+			}
+			defer func() { _ = session.Close(ctx) }()
+			report, err := session.Restore(ctx, dir)
+			if err != nil {
+				return exitErr(printer, err)
+			}
+			if printer.Format == app.OutputJSON {
+				return printer.Print(report)
+			}
+			writeStdout(printer, fmt.Sprintf("restored %s: created %d, updated %d", report.Directory, report.Created, report.Updated))
+			return nil
+		},
+	}
+}
diff --git a/cmd/haistack/command/command_test.go b/cmd/haistack/command/command_test.go
index daffbd05..0e96c27b 100644
--- a/cmd/haistack/command/command_test.go
+++ b/cmd/haistack/command/command_test.go
@@ -235,6 +235,50 @@ func TestImportCreatesThenUpdates(t *testing.T) {
 	}
 }
 
+func TestBackupRestoreRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	wd, _ := os.Getwd()
+	defer func() { _ = os.Chdir(wd) }()
+
+	coreModule := repoCoreModule(t)
+	writeConfig(t, dir, filepath.Join(dir, "test.db"), coreModule)
+
+	patientPath := filepath.Join(dir, "patient.json")
+	if err := os.WriteFile(patientPath, []byte(`{"resourceType":"Patient","id":"bak-1","name":[{"family":"Backup"}]}`), 0o644); err != nil {
+		t.Fatalf("write patient: %v", err)
+	}
+	if _, _, err := runCLI(t, dir, "import", patientPath); err != nil {
+		t.Fatalf("import: %v", err)
+	}
+
+	backupDir := filepath.Join(dir, "snap")
+	stdout, _, err := runCLI(t, dir, "backup", backupDir, "--output", "json")
+	if err != nil {
+		t.Fatalf("backup: %v", err)
+	}
+	if !strings.Contains(stdout, "Patient.ndjson") {
+		t.Fatalf("backup stdout = %q", stdout)
+	}
+	if _, err := os.Stat(filepath.Join(backupDir, "manifest.json")); err != nil {
+		t.Fatalf("manifest: %v", err)
+	}
+
+	if _, _, err := runCLI(t, dir, "delete", "Patient/bak-1", "--force"); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	restoreOut, _, err := runCLI(t, dir, "restore", backupDir, "--output", "json")
+	if err != nil {
+		t.Fatalf("restore: %v", err)
+	}
+	if !strings.Contains(restoreOut, `"created"`) {
+		t.Fatalf("restore stdout = %q", restoreOut)
+	}
+	readOut, _, err := runCLI(t, dir, "read", "Patient/bak-1", "--output", "json")
+	if err != nil || !strings.Contains(readOut, "Backup") {
+		t.Fatalf("read after restore: output=%q err=%v", readOut, err)
+	}
+}
+
 func TestSearchParsesKeyValueArgs(t *testing.T) {
 	dir := t.TempDir()
 	wd, _ := os.Getwd()
diff --git a/cmd/haistack/command/doc.go b/cmd/haistack/command/doc.go
index f2f50f77..c0dffc73 100644
--- a/cmd/haistack/command/doc.go
+++ b/cmd/haistack/command/doc.go
@@ -10,6 +10,7 @@
 //   - serve — start managed HTTP runtime
 //   - validate — structural FHIR validation
 //   - import — create or update one JSON resource file with an explicit conflict policy
+//   - backup, restore — NDJSON dump and reload of stored resources
 //   - read, delete, export — resource inspection and file-safe data operations
 //   - search — FHIR search with key=value parameters
 //   - fhirpath eval — evaluate a FHIRPath expression
diff --git a/cmd/haistack/command/root.go b/cmd/haistack/command/root.go
index b5b7901a..a2611429 100644
--- a/cmd/haistack/command/root.go
+++ b/cmd/haistack/command/root.go
@@ -90,6 +90,8 @@ against a Health AI Stack runtime. Configure storage and capabilities in haistac
 		newReadCommand(opts, printer),
 		newDeleteCommand(opts, printer),
 		newExportCommand(opts, printer),
+		newBackupCommand(opts, printer),
+		newRestoreCommand(opts, printer),
 		newSearchCommand(opts, printer),
 		newFHIRPathCommand(opts, printer),
 		newSyncCommand(opts, printer),
diff --git a/cmd/haistack/internal/app/operator.go b/cmd/haistack/internal/app/operator.go
index 688def36..b9a815bb 100644
--- a/cmd/haistack/internal/app/operator.go
+++ b/cmd/haistack/internal/app/operator.go
@@ -1,9 +1,15 @@
 package app
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
+	"os"
+	"path/filepath"
 	"sort"
+	"strings"
+	"time"
 
 	"github.com/degoke/health-ai-stack/pkg/store"
 	"github.com/degoke/health-ai-stack/pkg/types"
@@ -72,6 +78,191 @@ func (s *Session) ExportResources(ctx context.Context, resourceType string, limi
 	return out, nil
 }
 
+// BackupManifest describes a directory of NDJSON resource dumps.
+type BackupManifest struct {
+	Format    string               `json:"format"`
+	CreatedAt time.Time            `json:"createdAt"`
+	Files     []BackupManifestFile `json:"files"`
+}
+
+// BackupManifestFile is one resource-type NDJSON file in a backup.
+type BackupManifestFile struct {
+	Type  string `json:"type"`
+	Count int    `json:"count"`
+	File  string `json:"file"`
+}
+
+// BackupReport is the CLI result of writing a backup directory.
+type BackupReport struct {
+	Directory string               `json:"directory"`
+	Files     []BackupManifestFile `json:"files"`
+}
+
+const backupFormatV1 = "haistack-backup/v1"
+
+// Backup writes one NDJSON file per resource type plus manifest.json.
+func (s *Session) Backup(ctx context.Context, dir string) (*BackupReport, error) {
+	if dir == "" {
+		dir = "backup"
+	}
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, fmt.Errorf("create backup directory: %w", err)
+	}
+	manifest := BackupManifest{Format: backupFormatV1, CreatedAt: time.Now().UTC()}
+	for _, resourceType := range s.backupResourceTypes() {
+		resources, err := s.ExportResources(ctx, resourceType, 0)
+		if err != nil {
+			return nil, err
+		}
+		if len(resources) == 0 {
+			continue
+		}
+		filename := resourceType + ".ndjson"
+		var buf bytes.Buffer
+		for i, resource := range resources {
+			if i > 0 {
+				_ = buf.WriteByte('\n')
+			}
+			_, _ = buf.Write(bytes.TrimSpace(resource.JSON))
+		}
+		if err := os.WriteFile(filepath.Join(dir, filename), buf.Bytes(), 0o644); err != nil {
+			return nil, fmt.Errorf("write %s: %w", filename, err)
+		}
+		manifest.Files = append(manifest.Files, BackupManifestFile{
+			Type:  resourceType,
+			Count: len(resources),
+			File:  filename,
+		})
+	}
+	data, err := json.MarshalIndent(manifest, "", "  ")
+	if err != nil {
+		return nil, err
+	}
+	if err := os.WriteFile(filepath.Join(dir, "manifest.json"), append(data, '\n'), 0o644); err != nil {
+		return nil, fmt.Errorf("write manifest: %w", err)
+	}
+	return &BackupReport{Directory: dir, Files: manifest.Files}, nil
+}
+
+// RestoreReport is the CLI result of loading a backup directory.
+type RestoreReport struct {
+	Directory string `json:"directory"`
+	Created   int    `json:"created"`
+	Updated   int    `json:"updated"`
+	Files     int    `json:"files"`
+}
+
+// Restore loads NDJSON files from a backup directory using create-or-update.
+func (s *Session) Restore(ctx context.Context, dir string) (*RestoreReport, error) {
+	if dir == "" {
+		dir = "backup"
+	}
+	svc := s.Runtime.Services().ResourceService
+	if svc == nil {
+		return nil, fmt.Errorf("resource service is not available")
+	}
+	files, err := s.restoreFiles(dir)
+	if err != nil {
+		return nil, err
+	}
+	report := &RestoreReport{Directory: dir, Files: len(files)}
+	for _, file := range files {
+		data, err := os.ReadFile(filepath.Join(dir, file.File))
+		if err != nil {
+			return nil, fmt.Errorf("read %s: %w", file.File, err)
+		}
+		for i, line := range bytes.Split(data, []byte("\n")) {
+			line = bytes.TrimSpace(line)
+			if len(line) == 0 {
+				continue
+			}
+			env, err := types.NewJSONCodec().ParseJSON(file.Type, line)
+			if err != nil {
+				return nil, fmt.Errorf("%s line %d: %w", file.File, i+1, err)
+			}
+			cleaned, err := types.SetMeta(env.JSON, types.Meta{})
+			if err != nil {
+				return nil, fmt.Errorf("%s line %d: clear meta: %w", file.File, i+1, err)
+			}
+			env.JSON = cleaned
+			env.VersionID = ""
+			action, _, err := ImportResource(ctx, svc, env, false, false)
+			if err != nil {
+				return nil, fmt.Errorf("%s %s/%s: %w", file.File, env.ResourceType, env.ID, err)
+			}
+			switch action {
+			case "create":
+				report.Created++
+			case "update":
+				report.Updated++
+			}
+		}
+	}
+	return report, nil
+}
+
+func (s *Session) backupResourceTypes() []string {
+	if s != nil && s.Runtime != nil && s.Runtime.Services() != nil && s.Runtime.Services().RegistrySnapshot != nil {
+		types := s.Runtime.Services().RegistrySnapshot.EnabledResourceTypes()
+		if len(types) > 0 {
+			return types
+		}
+	}
+	return []string{"Patient", "Observation", "Encounter", "Practitioner", "Organization", "Appointment"}
+}
+
+func (s *Session) restoreFiles(dir string) ([]BackupManifestFile, error) {
+	manifestPath := filepath.Join(dir, "manifest.json")
+	if data, err := os.ReadFile(manifestPath); err == nil {
+		var manifest BackupManifest
+		if err := json.Unmarshal(data, &manifest); err != nil {
+			return nil, fmt.Errorf("parse manifest: %w", err)
+		}
+		return orderRestoreFiles(manifest.Files), nil
+	}
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		return nil, fmt.Errorf("read backup directory: %w", err)
+	}
+	var files []BackupManifestFile
+	for _, entry := range entries {
+		name := entry.Name()
+		if entry.IsDir() || !strings.HasSuffix(name, ".ndjson") {
+			continue
+		}
+		files = append(files, BackupManifestFile{
+			Type: strings.TrimSuffix(name, ".ndjson"),
+			File: name,
+		})
+	}
+	return orderRestoreFiles(files), nil
+}
+
+func orderRestoreFiles(files []BackupManifestFile) []BackupManifestFile {
+	priority := map[string]int{
+		"Organization": 0,
+		"Practitioner": 1,
+		"Patient":      2,
+		"Encounter":    3,
+	}
+	out := append([]BackupManifestFile(nil), files...)
+	sort.SliceStable(out, func(i, j int) bool {
+		pi, okI := priority[out[i].Type]
+		if !okI {
+			pi = 50
+		}
+		pj, okJ := priority[out[j].Type]
+		if !okJ {
+			pj = 50
+		}
+		if pi != pj {
+			return pi < pj
+		}
+		return out[i].Type < out[j].Type
+	})
+	return out
+}
+
 // ReindexPlan counts current resources for the selected enabled type(s) without
 // modifying search indexes.
 func (s *Session) ReindexPlan(ctx context.Context, resourceType string) ([]ReindexPlanItem, error) {
diff --git a/pkg/bulkimport/doc.go b/pkg/bulkimport/doc.go
new file mode 100644
index 00000000..678d8d8e
--- /dev/null
+++ b/pkg/bulkimport/doc.go
@@ -0,0 +1,19 @@
+// Package bulkimport implements FHIR Bulk Data $import orchestration for HAIStack.
+//
+// The import service accepts a Parameters kickoff (Prefer: respond-async),
+// reads application/fhir+ndjson inputs, and creates or updates resources
+// through a ResourceWriter. Background execution integrates with pkg/jobs
+// via TypeImportBulk.
+//
+// Remote input.url values are fetched only when a URLLoader is configured.
+// The runtime wires HTTPLoader, which GETs http and https URLs with a 60s
+// timeout and a 64MiB size limit. file:// and other non-http(s) schemes are
+// rejected; redirects to those schemes are also refused. Tests may omit the
+// loader so URL-only kickoffs fail closed.
+//
+// HTTP adapters in pkg/http expose:
+//   - POST /fhir/$import
+//   - GET /fhir/$import/status/{jobId}
+//   - DELETE /fhir/$import/status/{jobId}
+//   - GET /fhir/$import/files/{jobId}/{filename} — download import error NDJSON
+package bulkimport
diff --git a/pkg/bulkimport/executor.go b/pkg/bulkimport/executor.go
new file mode 100644
index 00000000..f7b23133
--- /dev/null
+++ b/pkg/bulkimport/executor.go
@@ -0,0 +1,202 @@
+package bulkimport
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/degoke/health-ai-stack/pkg/core"
+	"github.com/degoke/health-ai-stack/pkg/types"
+)
+
+// ResourceWriter persists imported FHIR resources.
+type ResourceWriter interface {
+	Create(ctx context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error)
+	Update(ctx context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error)
+	Read(ctx context.Context, resourceType, id string) (*types.ResourceEnvelope, error)
+}
+
+// Executor reads NDJSON inputs and upserts resources.
+type Executor struct {
+	Resources ResourceWriter
+	Files     FileStore
+	Codec     types.ResourceCodec
+}
+
+// ExecuteRequest configures one import execution pass.
+type ExecuteRequest struct {
+	JobID       string
+	Inputs      []InputFile
+	BaseFileURL string
+	IsCancelled func() bool
+	OnProgress  func(done, total int)
+}
+
+// ExecuteResult summarizes imported counts and error artifacts.
+type ExecuteResult struct {
+	Output []CountFile
+	Errors []ErrorFile
+}
+
+// Execute imports NDJSON inputs stored for the job.
+func (e *Executor) Execute(ctx context.Context, req ExecuteRequest) (*ExecuteResult, error) {
+	if e == nil || e.Resources == nil || e.Files == nil {
+		return nil, fmt.Errorf("import: executor is not configured")
+	}
+	codec := e.Codec
+	if codec == nil {
+		codec = types.NewJSONCodec()
+	}
+	result := &ExecuteResult{}
+	total := len(req.Inputs)
+	for i, input := range req.Inputs {
+		if req.IsCancelled != nil && req.IsCancelled() {
+			return nil, fmt.Errorf("import cancelled")
+		}
+		if req.OnProgress != nil {
+			req.OnProgress(i, total)
+		}
+		output, errFile, err := e.importInput(ctx, codec, req, i, input)
+		if err != nil {
+			return nil, err
+		}
+		if output != nil {
+			result.Output = append(result.Output, *output)
+		}
+		if errFile != nil {
+			result.Errors = append(result.Errors, *errFile)
+		}
+	}
+	if req.OnProgress != nil {
+		req.OnProgress(total, total)
+	}
+	return result, nil
+}
+
+func (e *Executor) importInput(ctx context.Context, codec types.ResourceCodec, req ExecuteRequest, index int, input InputFile) (*CountFile, *ErrorFile, error) {
+	data, _, err := e.Files.Get(ctx, inputPath(req.JobID, index, input.Type))
+	if err != nil {
+		return nil, nil, err
+	}
+	var errBuf bytes.Buffer
+	errorsWritten := 0
+	imported := 0
+	for _, line := range bytes.Split(data, []byte("\n")) {
+		line = bytes.TrimSpace(line)
+		if len(line) == 0 {
+			continue
+		}
+		env, parseErr := codec.ParseJSON(input.Type, line)
+		if parseErr != nil {
+			writeImportError(&errBuf, &errorsWritten, fmt.Sprintf("parse %s: %v", input.Type, parseErr))
+			continue
+		}
+		if input.Type != "" && env.ResourceType != "" && env.ResourceType != input.Type {
+			writeImportError(&errBuf, &errorsWritten, fmt.Sprintf("resourceType %s does not match input type %s", env.ResourceType, input.Type))
+			continue
+		}
+		if _, err := upsertResource(ctx, e.Resources, env); err != nil {
+			writeImportError(&errBuf, &errorsWritten, fmt.Sprintf("persist %s/%s: %v", env.ResourceType, env.ID, err))
+			continue
+		}
+		imported++
+	}
+	output := &CountFile{Type: input.Type, Count: imported}
+	if errorsWritten == 0 {
+		return output, nil, nil
+	}
+	errPath := errorPath(req.JobID, index, input.Type)
+	if err := e.Files.Put(ctx, errPath, errBuf.Bytes(), InputFormatNDJSON); err != nil {
+		return nil, nil, err
+	}
+	return output, &ErrorFile{Type: input.Type, URL: errorFileURL(req.BaseFileURL, req.JobID, index, input.Type)}, nil
+}
+
+func upsertResource(ctx context.Context, writer ResourceWriter, env *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	if err := stripServerAssignedMeta(env); err != nil {
+		return nil, err
+	}
+	if env.ID == "" {
+		return writer.Create(ctx, env)
+	}
+	_, err := writer.Read(ctx, env.ResourceType, env.ID)
+	if err == nil {
+		return writer.Update(ctx, env)
+	}
+	if !core.IsNotFound(err) {
+		return nil, err
+	}
+	return writer.Create(ctx, env)
+}
+
+// stripServerAssignedMeta matches CLI restore: types.SetMeta(env.JSON, types.Meta{})
+// clears meta.versionId and meta.lastUpdated while preserving profile/tag, then
+// VersionID (and LastUpdated) are cleared on the envelope before persist.
+func stripServerAssignedMeta(env *types.ResourceEnvelope) error {
+	if env == nil {
+		return fmt.Errorf("import: resource envelope is required")
+	}
+	cleaned, err := types.SetMeta(env.JSON, types.Meta{})
+	if err != nil {
+		return fmt.Errorf("import: clear meta: %w", err)
+	}
+	env.JSON = cleaned
+	env.VersionID = ""
+	env.LastUpdated = time.Time{}
+	return nil
+}
+
+func writeImportError(buf *bytes.Buffer, count *int, message string) {
+	line, _ := json.Marshal(map[string]any{
+		"resourceType": "OperationOutcome",
+		"issue": []map[string]string{{
+			"severity":    "error",
+			"code":        "processing",
+			"diagnostics": message,
+		}},
+	})
+	if *count > 0 {
+		_ = buf.WriteByte('\n')
+	}
+	_, _ = buf.Write(line)
+	*count++
+}
+
+func inputPath(jobID string, index int, resourceType string) string {
+	if resourceType == "" {
+		resourceType = "Resource"
+	}
+	return fmt.Sprintf("%s/input-%d-%s.ndjson", jobID, index, sanitizeType(resourceType))
+}
+
+func errorFilename(index int, resourceType string) string {
+	if resourceType == "" {
+		resourceType = "Resource"
+	}
+	return fmt.Sprintf("error-%d-%s.ndjson", index, sanitizeType(resourceType))
+}
+
+func errorPath(jobID string, index int, resourceType string) string {
+	return jobID + "/" + errorFilename(index, resourceType)
+}
+
+func errorFileURL(baseFileURL, jobID string, index int, resourceType string) string {
+	name := errorFilename(index, resourceType)
+	base := strings.TrimSuffix(baseFileURL, "/")
+	if base == "" {
+		return errorPath(jobID, index, resourceType)
+	}
+	return base + "/" + name
+}
+
+func sanitizeType(resourceType string) string {
+	return strings.Map(func(r rune) rune {
+		if (r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '_' || r == '-' {
+			return r
+		}
+		return '-'
+	}, resourceType)
+}
diff --git a/pkg/bulkimport/loader.go b/pkg/bulkimport/loader.go
new file mode 100644
index 00000000..43f5344b
--- /dev/null
+++ b/pkg/bulkimport/loader.go
@@ -0,0 +1,117 @@
+package bulkimport
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+const (
+	// HTTPLoaderTimeout is the default client timeout for remote NDJSON fetches.
+	HTTPLoaderTimeout = 60 * time.Second
+	// HTTPLoaderMaxBytes is the default maximum size of a fetched NDJSON payload.
+	HTTPLoaderMaxBytes int64 = 64 << 20
+)
+
+// HTTPLoader fetches import input.url values over HTTP(S).
+//
+// Only http and https URLs are accepted. file://, data:, and other schemes are
+// rejected before the request is issued, and redirects to those schemes are
+// refused. The default client uses a 60s timeout and reads at most 64MiB.
+type HTTPLoader struct {
+	Client   *http.Client
+	MaxBytes int64
+}
+
+// NewHTTPLoader returns an HTTPLoader with a 60s timeout.
+func NewHTTPLoader() *HTTPLoader {
+	return &HTTPLoader{
+		Client: &http.Client{
+			Timeout:       HTTPLoaderTimeout,
+			CheckRedirect: httpLoaderRedirect,
+		},
+		MaxBytes: HTTPLoaderMaxBytes,
+	}
+}
+
+var _ URLLoader = (*HTTPLoader)(nil)
+
+// Load GETs an http or https URL and returns the response body.
+func (l *HTTPLoader) Load(ctx context.Context, rawURL string) ([]byte, error) {
+	if l == nil {
+		return nil, fmt.Errorf("import: HTTP loader is nil")
+	}
+	parsed, err := url.Parse(rawURL)
+	if err != nil {
+		return nil, fmt.Errorf("import: invalid url: %w", err)
+	}
+	if err := allowedImportURLScheme(parsed.Scheme); err != nil {
+		return nil, err
+	}
+	if parsed.Host == "" {
+		return nil, fmt.Errorf("import: url host is required")
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, parsed.String(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("import: build request: %w", err)
+	}
+	resp, err := l.client().Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("import: get %s: %w", parsed.Redacted(), err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1024))
+		return nil, fmt.Errorf("import: get %s: unexpected status %d", parsed.Redacted(), resp.StatusCode)
+	}
+	maxBytes := l.MaxBytes
+	if maxBytes <= 0 {
+		maxBytes = HTTPLoaderMaxBytes
+	}
+	data, err := io.ReadAll(io.LimitReader(resp.Body, maxBytes+1))
+	if err != nil {
+		return nil, fmt.Errorf("import: read %s: %w", parsed.Redacted(), err)
+	}
+	if int64(len(data)) > maxBytes {
+		return nil, fmt.Errorf("import: response from %s exceeds %d bytes", parsed.Redacted(), maxBytes)
+	}
+	return data, nil
+}
+
+func (l *HTTPLoader) client() *http.Client {
+	if l != nil && l.Client != nil {
+		return l.Client
+	}
+	return &http.Client{
+		Timeout:       HTTPLoaderTimeout,
+		CheckRedirect: httpLoaderRedirect,
+	}
+}
+
+func httpLoaderRedirect(req *http.Request, via []*http.Request) error {
+	if req == nil || req.URL == nil {
+		return fmt.Errorf("import: redirect missing url")
+	}
+	if err := allowedImportURLScheme(req.URL.Scheme); err != nil {
+		return err
+	}
+	if len(via) >= 10 {
+		return fmt.Errorf("import: too many redirects")
+	}
+	return nil
+}
+
+func allowedImportURLScheme(scheme string) error {
+	switch strings.ToLower(strings.TrimSpace(scheme)) {
+	case "http", "https":
+		return nil
+	case "":
+		return fmt.Errorf("import: url scheme is required; only http and https are allowed")
+	default:
+		return fmt.Errorf("import: url scheme %q is not allowed; only http and https are allowed", scheme)
+	}
+}
diff --git a/pkg/bulkimport/loader_test.go b/pkg/bulkimport/loader_test.go
new file mode 100644
index 00000000..d6608b7f
--- /dev/null
+++ b/pkg/bulkimport/loader_test.go
@@ -0,0 +1,53 @@
+package bulkimport_test
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
+)
+
+func TestHTTPLoaderRejectsFileURL(t *testing.T) {
+	loader := bulkimport.NewHTTPLoader()
+	_, err := loader.Load(context.Background(), "file:///etc/passwd")
+	if err == nil {
+		t.Fatal("expected file:// to be rejected")
+	}
+	if !strings.Contains(err.Error(), "not allowed") {
+		t.Fatalf("error = %v, want scheme rejection", err)
+	}
+}
+
+func TestHTTPLoaderRejectsUnexpectedSchemes(t *testing.T) {
+	loader := bulkimport.NewHTTPLoader()
+	for _, raw := range []string{"ftp://example.test/Patient.ndjson", "data:text/plain,hi", "/etc/passwd"} {
+		if _, err := loader.Load(context.Background(), raw); err == nil {
+			t.Fatalf("expected %q to be rejected", raw)
+		}
+	}
+}
+
+func TestHTTPLoaderFetchesHTTP(t *testing.T) {
+	want := `{"resourceType":"Patient","id":"p1"}`
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			t.Errorf("method = %s", r.Method)
+		}
+		w.Header().Set("Content-Type", "application/fhir+ndjson")
+		_, _ = io.WriteString(w, want)
+	}))
+	t.Cleanup(server.Close)
+
+	loader := bulkimport.NewHTTPLoader()
+	got, err := loader.Load(context.Background(), server.URL+"/Patient.ndjson")
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if string(got) != want {
+		t.Fatalf("body = %q, want %q", got, want)
+	}
+}
diff --git a/pkg/bulkimport/parameters.go b/pkg/bulkimport/parameters.go
new file mode 100644
index 00000000..448ca5be
--- /dev/null
+++ b/pkg/bulkimport/parameters.go
@@ -0,0 +1,98 @@
+package bulkimport
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+)
+
+// ParseParametersKickoff decodes a FHIR Parameters resource into a kickoff request.
+// Each repeating `input` parameter must include a `type` part and either inline
+// NDJSON (`valueString` / `ndjson`) or a `url` part.
+func ParseParametersKickoff(data []byte) (KickoffRequest, error) {
+	var raw map[string]any
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return KickoffRequest{}, fmt.Errorf("import: parse Parameters: %w", err)
+	}
+	if rt, _ := raw["resourceType"].(string); rt != "" && rt != "Parameters" {
+		return KickoffRequest{}, fmt.Errorf("import: expected Parameters resource, got %s", rt)
+	}
+	req := KickoffRequest{InputFormat: InputFormatNDJSON}
+	params, _ := raw["parameter"].([]any)
+	for _, item := range params {
+		obj, ok := item.(map[string]any)
+		if !ok {
+			continue
+		}
+		name, _ := obj["name"].(string)
+		switch name {
+		case "inputFormat":
+			if v := parameterString(obj); v != "" {
+				req.InputFormat = v
+			}
+		case "input":
+			input, err := parseInputParameter(obj)
+			if err != nil {
+				return KickoffRequest{}, err
+			}
+			req.Inputs = append(req.Inputs, input)
+		}
+	}
+	if len(req.Inputs) == 0 {
+		return KickoffRequest{}, fmt.Errorf("import: Parameters must include at least one input")
+	}
+	return req, nil
+}
+
+func parseInputParameter(obj map[string]any) (InputFile, error) {
+	input := InputFile{}
+	parts, _ := obj["part"].([]any)
+	for _, raw := range parts {
+		part, ok := raw.(map[string]any)
+		if !ok {
+			continue
+		}
+		name, _ := part["name"].(string)
+		switch strings.ToLower(strings.TrimSpace(name)) {
+		case "type":
+			input.Type = parameterString(part)
+		case "url":
+			input.URL = parameterString(part)
+		case "valuestring", "ndjson", "resource":
+			input.NDJSON = []byte(parameterString(part))
+		}
+	}
+	if input.Type == "" && len(input.NDJSON) > 0 {
+		var peek struct {
+			ResourceType string `json:"resourceType"`
+		}
+		first := input.NDJSON
+		if i := indexByte(first, '\n'); i >= 0 {
+			first = first[:i]
+		}
+		_ = json.Unmarshal(first, &peek)
+		input.Type = peek.ResourceType
+	}
+	if input.Type == "" && input.URL == "" && len(input.NDJSON) == 0 {
+		return InputFile{}, fmt.Errorf("import: input parameter is missing type and payload")
+	}
+	return input, nil
+}
+
+func parameterString(obj map[string]any) string {
+	for _, key := range []string{"valueCode", "valueString", "valueUri", "valueUrl", "valueCanonical"} {
+		if v, ok := obj[key].(string); ok && strings.TrimSpace(v) != "" {
+			return strings.TrimSpace(v)
+		}
+	}
+	return ""
+}
+
+func indexByte(data []byte, c byte) int {
+	for i, b := range data {
+		if b == c {
+			return i
+		}
+	}
+	return -1
+}
diff --git a/pkg/bulkimport/service.go b/pkg/bulkimport/service.go
new file mode 100644
index 00000000..7325302f
--- /dev/null
+++ b/pkg/bulkimport/service.go
@@ -0,0 +1,314 @@
+package bulkimport
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/degoke/health-ai-stack/pkg/core"
+	"github.com/degoke/health-ai-stack/pkg/jobs"
+	"github.com/degoke/health-ai-stack/pkg/store"
+	"github.com/google/uuid"
+)
+
+// URLLoader optionally fetches remote NDJSON listed in Parameters input.url.
+type URLLoader interface {
+	Load(ctx context.Context, url string) ([]byte, error)
+}
+
+// Config configures a Service.
+type Config struct {
+	Jobs      JobStore
+	Files     FileStore
+	Executor  *Executor
+	JobQueue  store.JobStore
+	Loader    URLLoader
+	BasePath  string
+	PublicURL string
+	Now       func() time.Time
+	NewID     func() string
+}
+
+// Service orchestrates bulk import kickoff, polling, cancellation, and execution.
+type Service struct {
+	jobs      JobStore
+	files     FileStore
+	executor  *Executor
+	jobQueue  store.JobStore
+	loader    URLLoader
+	basePath  string
+	publicURL string
+	now       func() time.Time
+	newID     func() string
+}
+
+// NewService validates configuration and returns a Service.
+func NewService(cfg Config) (*Service, error) {
+	if cfg.Jobs == nil {
+		return nil, fmt.Errorf("import: JobStore is required")
+	}
+	if cfg.Files == nil {
+		return nil, fmt.Errorf("import: FileStore is required")
+	}
+	if cfg.Executor == nil {
+		return nil, fmt.Errorf("import: Executor is required")
+	}
+	if cfg.BasePath == "" {
+		cfg.BasePath = "/fhir"
+	}
+	if cfg.PublicURL == "" {
+		cfg.PublicURL = cfg.BasePath
+	}
+	if cfg.Now == nil {
+		cfg.Now = func() time.Time { return time.Now().UTC() }
+	}
+	if cfg.NewID == nil {
+		cfg.NewID = uuid.NewString
+	}
+	return &Service{
+		jobs:      cfg.Jobs,
+		files:     cfg.Files,
+		executor:  cfg.Executor,
+		jobQueue:  cfg.JobQueue,
+		loader:    cfg.Loader,
+		basePath:  strings.TrimSuffix(cfg.BasePath, "/"),
+		publicURL: strings.TrimSuffix(cfg.PublicURL, "/"),
+		now:       cfg.Now,
+		newID:     cfg.NewID,
+	}, nil
+}
+
+// Kickoff creates an async import job and optionally enqueues background execution.
+func (s *Service) Kickoff(ctx context.Context, req KickoffRequest) (*Job, error) {
+	if s == nil {
+		return nil, fmt.Errorf("import: service is nil")
+	}
+	if err := s.prepareInputs(ctx, &req); err != nil {
+		return nil, err
+	}
+	id := s.newID()
+	now := nowUTC(s.now)
+	for i, input := range req.Inputs {
+		if err := s.files.Put(ctx, inputPath(id, i, input.Type), input.NDJSON, InputFormatNDJSON); err != nil {
+			return nil, err
+		}
+		req.Inputs[i].NDJSON = nil
+	}
+	job := Job{
+		ID:        id,
+		Status:    StatusInProgress,
+		Request:   req,
+		CreatedAt: now,
+		Progress:  "0%",
+	}
+	if err := s.jobs.Create(ctx, job); err != nil {
+		return nil, err
+	}
+	if s.jobQueue != nil {
+		_, err := jobs.Enqueue(ctx, s.jobQueue, jobs.TypeImportBulk, JobPayload{JobID: id}, jobs.EnqueueOptions{
+			Now: s.now,
+		})
+		if err != nil {
+			return nil, err
+		}
+	} else if err := s.RunJob(ctx, id); err != nil {
+		return nil, err
+	}
+	return s.jobs.Get(ctx, id)
+}
+
+func (s *Service) prepareInputs(ctx context.Context, req *KickoffRequest) error {
+	if strings.TrimSpace(req.InputFormat) != "" && !strings.EqualFold(req.InputFormat, InputFormatNDJSON) {
+		return invalidImport(fmt.Sprintf("import: unsupported inputFormat %q", req.InputFormat))
+	}
+	if len(req.Inputs) == 0 {
+		return invalidImport("import: at least one input is required")
+	}
+	for i := range req.Inputs {
+		input := &req.Inputs[i]
+		if len(bytes.TrimSpace(input.NDJSON)) > 0 {
+			continue
+		}
+		if strings.TrimSpace(input.URL) == "" {
+			return invalidImport(fmt.Sprintf("import: input %d requires NDJSON or url", i))
+		}
+		if s.loader == nil {
+			return invalidImport(fmt.Sprintf("import: input url %q cannot be fetched; provide inline NDJSON", input.URL))
+		}
+		data, err := s.loader.Load(ctx, input.URL)
+		if err != nil {
+			return invalidImport(fmt.Sprintf("import: load %s: %v", input.URL, err))
+		}
+		input.NDJSON = data
+	}
+	return nil
+}
+
+// GetJob returns the current job state.
+func (s *Service) GetJob(ctx context.Context, id string) (*Job, error) {
+	return s.jobs.Get(ctx, id)
+}
+
+// Cancel requests cancellation of an in-progress import job.
+func (s *Service) Cancel(ctx context.Context, id string) error {
+	job, err := s.jobs.Get(ctx, id)
+	if err != nil {
+		return err
+	}
+	if job == nil {
+		return fmt.Errorf("import: job %q not found", id)
+	}
+	switch job.Status {
+	case StatusComplete, StatusError, StatusCancelled:
+		return nil
+	}
+	job.CancelRequested = true
+	job.Status = StatusCancelled
+	job.CompletedAt = nowUTC(s.now)
+	return s.jobs.Update(ctx, *job)
+}
+
+// Manifest builds the import manifest for a completed job.
+func (s *Service) Manifest(job *Job) *Manifest {
+	if job == nil {
+		return nil
+	}
+	return &Manifest{
+		TransactionTime:     job.TransactionTime,
+		Request:             job.Request.RequestURL,
+		RequiresAccessToken: job.Request.RequiresAccess,
+		Output:              append([]CountFile(nil), job.Output...),
+		Error:               append([]ErrorFile(nil), job.Errors...),
+	}
+}
+
+// StatusURL returns the polling URL for a job.
+func (s *Service) StatusURL(jobID string) string {
+	return s.publicURL + "/$import/status/" + jobID
+}
+
+// FileURL returns the public URL prefix for import error artifacts.
+func (s *Service) FileURL(jobID string) string {
+	return s.publicURL + "/$import/files/" + jobID
+}
+
+// GetFile serves one import error artifact.
+func (s *Service) GetFile(ctx context.Context, jobID, filename string) ([]byte, string, error) {
+	path := jobID + "/" + strings.TrimPrefix(filename, "/")
+	return s.files.Get(ctx, path)
+}
+
+// RunJob executes one import job synchronously.
+func (s *Service) RunJob(ctx context.Context, jobID string) error {
+	job, err := s.jobs.Get(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	if job == nil {
+		return fmt.Errorf("import: job %q not found", jobID)
+	}
+	if job.Status != StatusInProgress {
+		return nil
+	}
+
+	isCancelled := func() bool {
+		current, getErr := s.jobs.Get(ctx, jobID)
+		if getErr != nil || current == nil {
+			return false
+		}
+		return current.CancelRequested || current.Status == StatusCancelled
+	}
+
+	result, err := s.executor.Execute(ctx, ExecuteRequest{
+		JobID:       jobID,
+		Inputs:      job.Request.Inputs,
+		BaseFileURL: s.FileURL(jobID),
+		IsCancelled: isCancelled,
+		OnProgress: func(done, total int) {
+			if total <= 0 {
+				return
+			}
+			current, getErr := s.jobs.Get(ctx, jobID)
+			if getErr != nil || current == nil {
+				return
+			}
+			if current.Status == StatusCancelled || current.CancelRequested {
+				return
+			}
+			current.Progress = fmt.Sprintf("%d%%", (done*100)/total)
+			_ = s.jobs.Update(ctx, *current)
+		},
+	})
+	if err != nil {
+		current, getErr := s.jobs.Get(ctx, jobID)
+		if getErr != nil {
+			return getErr
+		}
+		if current == nil {
+			return fmt.Errorf("import: job %q not found", jobID)
+		}
+		if current.CancelRequested || current.Status == StatusCancelled {
+			current.Status = StatusCancelled
+			current.CompletedAt = nowUTC(s.now)
+			return s.jobs.Update(ctx, *current)
+		}
+		return s.failJob(ctx, current, err)
+	}
+
+	return s.completeJob(ctx, jobID, result)
+}
+
+func (s *Service) completeJob(ctx context.Context, jobID string, result *ExecuteResult) error {
+	current, err := s.jobs.Get(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	if current == nil {
+		return fmt.Errorf("import: job %q not found", jobID)
+	}
+	if current.Status == StatusCancelled || current.CancelRequested {
+		current.Status = StatusCancelled
+		current.CompletedAt = nowUTC(s.now)
+		return s.jobs.Update(ctx, *current)
+	}
+	current.Status = StatusComplete
+	current.TransactionTime = nowUTC(s.now)
+	current.CompletedAt = current.TransactionTime
+	current.Progress = "100%"
+	if result != nil {
+		current.Output = result.Output
+		current.Errors = result.Errors
+	}
+	return s.jobs.Update(ctx, *current)
+}
+
+func (s *Service) failJob(ctx context.Context, job *Job, cause error) error {
+	job.Status = StatusError
+	job.LastError = cause.Error()
+	job.CompletedAt = nowUTC(s.now)
+	return s.jobs.Update(ctx, *job)
+}
+
+// JobHandler returns a jobs.Handler that executes bulk import jobs.
+func (s *Service) JobHandler() jobs.Handler {
+	return jobs.HandlerFunc(func(ctx context.Context, job store.JobRecord) error {
+		var payload JobPayload
+		if err := jobs.UnmarshalPayload(job.Payload, &payload); err != nil {
+			return err
+		}
+		return s.RunJob(ctx, payload.JobID)
+	})
+}
+
+func nowUTC(now func() time.Time) time.Time {
+	if now == nil {
+		return time.Now().UTC()
+	}
+	return now().UTC()
+}
+
+func invalidImport(message string) error {
+	return &core.ServiceError{Kind: core.ErrorKindInvalid, Message: message}
+}
diff --git a/pkg/bulkimport/service_test.go b/pkg/bulkimport/service_test.go
new file mode 100644
index 00000000..c163e43c
--- /dev/null
+++ b/pkg/bulkimport/service_test.go
@@ -0,0 +1,309 @@
+package bulkimport_test
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
+	"github.com/degoke/health-ai-stack/pkg/core"
+	"github.com/degoke/health-ai-stack/pkg/types"
+)
+
+type memoryWriter struct {
+	resources map[string]*types.ResourceEnvelope
+	created   []*types.ResourceEnvelope
+	updated   []*types.ResourceEnvelope
+}
+
+func (m *memoryWriter) key(resourceType, id string) string { return resourceType + "/" + id }
+
+func cloneEnvelope(resource *types.ResourceEnvelope) *types.ResourceEnvelope {
+	cp := *resource
+	if resource.JSON != nil {
+		cp.JSON = append([]byte(nil), resource.JSON...)
+	}
+	return &cp
+}
+
+func (m *memoryWriter) Create(_ context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	cp := cloneEnvelope(resource)
+	m.created = append(m.created, cp)
+	m.resources[m.key(resource.ResourceType, resource.ID)] = cp
+	return cp, nil
+}
+
+func (m *memoryWriter) Update(_ context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	cp := cloneEnvelope(resource)
+	m.updated = append(m.updated, cp)
+	m.resources[m.key(resource.ResourceType, resource.ID)] = cp
+	return cp, nil
+}
+
+func (m *memoryWriter) Read(_ context.Context, resourceType, id string) (*types.ResourceEnvelope, error) {
+	if env := m.resources[m.key(resourceType, id)]; env != nil {
+		return env, nil
+	}
+	return nil, &core.ServiceError{Kind: core.ErrorKindNotFound, Message: "not found"}
+}
+
+func TestBulkImportRoundTrip(t *testing.T) {
+	now := time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC)
+	writer := &memoryWriter{resources: map[string]*types.ResourceEnvelope{}}
+	files := bulkimport.NewInMemoryFileStore()
+	jobs := bulkimport.NewInMemoryJobStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     jobs,
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: writer, Files: files},
+		BasePath: "/fhir",
+		Now:      func() time.Time { return now },
+		NewID:    func() string { return "job-1" },
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+
+	ndjson := []byte(`{"resourceType":"Patient","id":"p1","name":[{"family":"Doe"}]}` + "\n" +
+		`{"resourceType":"Patient","id":"p2","name":[{"family":"Roe"}]}`)
+	job, err := svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		InputFormat: bulkimport.InputFormatNDJSON,
+		Inputs:      []bulkimport.InputFile{{Type: "Patient", NDJSON: ndjson}},
+		RequestURL:  "POST /fhir/$import",
+	})
+	if err != nil {
+		t.Fatalf("Kickoff: %v", err)
+	}
+	if job.Status != bulkimport.StatusComplete {
+		t.Fatalf("status = %s error=%s", job.Status, job.LastError)
+	}
+	manifest := svc.Manifest(job)
+	if manifest == nil || len(manifest.Output) != 1 || manifest.Output[0].Count != 2 {
+		t.Fatalf("manifest = %#v", manifest)
+	}
+	if _, err := writer.Read(context.Background(), "Patient", "p1"); err != nil {
+		t.Fatalf("read p1: %v", err)
+	}
+	if _, err := writer.Read(context.Background(), "Patient", "p2"); err != nil {
+		t.Fatalf("read p2: %v", err)
+	}
+}
+
+func TestParseParametersKickoffInlineNDJSON(t *testing.T) {
+	body := []byte(`{
+		"resourceType":"Parameters",
+		"parameter":[
+			{"name":"inputFormat","valueCode":"application/fhir+ndjson"},
+			{"name":"input","part":[
+				{"name":"type","valueCode":"Patient"},
+				{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}"}
+			]}
+		]
+	}`)
+	req, err := bulkimport.ParseParametersKickoff(body)
+	if err != nil {
+		t.Fatalf("ParseParametersKickoff: %v", err)
+	}
+	if req.InputFormat != bulkimport.InputFormatNDJSON {
+		t.Fatalf("format = %q", req.InputFormat)
+	}
+	if len(req.Inputs) != 1 || req.Inputs[0].Type != "Patient" {
+		t.Fatalf("inputs = %#v", req.Inputs)
+	}
+	if !strings.Contains(string(req.Inputs[0].NDJSON), `"p1"`) {
+		t.Fatalf("ndjson = %s", req.Inputs[0].NDJSON)
+	}
+}
+
+func TestParseParametersKickoffRequiresInput(t *testing.T) {
+	_, err := bulkimport.ParseParametersKickoff([]byte(`{"resourceType":"Parameters"}`))
+	if err == nil {
+		t.Fatal("expected error")
+	}
+}
+
+func TestKickoffRejectsURLWithoutLoader(t *testing.T) {
+	files := bulkimport.NewInMemoryFileStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: &memoryWriter{resources: map[string]*types.ResourceEnvelope{}}, Files: files},
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	_, err = svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		Inputs: []bulkimport.InputFile{{Type: "Patient", URL: "https://example.test/Patient.ndjson"}},
+	})
+	if err == nil {
+		t.Fatal("expected url loader error")
+	}
+	if core.KindOf(err) != core.ErrorKindInvalid {
+		t.Fatalf("kind = %s, want invalid", core.KindOf(err))
+	}
+}
+
+func TestImportUpdatesExistingResource(t *testing.T) {
+	writer := &memoryWriter{resources: map[string]*types.ResourceEnvelope{
+		"Patient/p1": {ResourceType: "Patient", ID: "p1", JSON: []byte(`{"resourceType":"Patient","id":"p1"}`)},
+	}}
+	files := bulkimport.NewInMemoryFileStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: writer, Files: files},
+		NewID:    func() string { return "job-upd" },
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	job, err := svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		Inputs: []bulkimport.InputFile{{
+			Type:   "Patient",
+			NDJSON: []byte(`{"resourceType":"Patient","id":"p1","name":[{"family":"Updated"}]}`),
+		}},
+	})
+	if err != nil {
+		t.Fatalf("Kickoff: %v", err)
+	}
+	if job.Status != bulkimport.StatusComplete {
+		t.Fatalf("status = %s", job.Status)
+	}
+	got, err := writer.Read(context.Background(), "Patient", "p1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var obj map[string]any
+	if err := json.Unmarshal(got.JSON, &obj); err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(got.JSON), "Updated") {
+		t.Fatalf("json = %s", got.JSON)
+	}
+}
+
+func TestImportStripsMetaVersionIDBeforePersist(t *testing.T) {
+	writer := &memoryWriter{resources: map[string]*types.ResourceEnvelope{}}
+	files := bulkimport.NewInMemoryFileStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: writer, Files: files},
+		NewID:    func() string { return "job-meta" },
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	ndjson := []byte(`{"resourceType":"Patient","id":"p1","meta":{"versionId":"9","lastUpdated":"2024-01-01T00:00:00Z","profile":["http://example.org/StructureDefinition/Patient"],"tag":[{"system":"http://example.org/tags","code":"imported"}]}}`)
+	job, err := svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		Inputs: []bulkimport.InputFile{{Type: "Patient", NDJSON: ndjson}},
+	})
+	if err != nil {
+		t.Fatalf("Kickoff: %v", err)
+	}
+	if job.Status != bulkimport.StatusComplete {
+		t.Fatalf("status = %s error=%s", job.Status, job.LastError)
+	}
+	if len(writer.created) != 1 {
+		t.Fatalf("created = %d", len(writer.created))
+	}
+	got := writer.created[0]
+	if got.VersionID != "" {
+		t.Fatalf("envelope VersionID = %q, want empty", got.VersionID)
+	}
+	if !got.LastUpdated.IsZero() {
+		t.Fatalf("envelope LastUpdated = %v, want zero", got.LastUpdated)
+	}
+	var obj map[string]any
+	if err := json.Unmarshal(got.JSON, &obj); err != nil {
+		t.Fatal(err)
+	}
+	meta, _ := obj["meta"].(map[string]any)
+	if meta == nil {
+		t.Fatalf("expected profile/tag meta to be preserved, json = %s", got.JSON)
+	}
+	if _, ok := meta["versionId"]; ok {
+		t.Fatalf("versionId still present: %s", got.JSON)
+	}
+	if _, ok := meta["lastUpdated"]; ok {
+		t.Fatalf("lastUpdated still present: %s", got.JSON)
+	}
+	if _, ok := meta["profile"]; !ok {
+		t.Fatalf("profile dropped: %s", got.JSON)
+	}
+	if _, ok := meta["tag"]; !ok {
+		t.Fatalf("tag dropped: %s", got.JSON)
+	}
+}
+
+func TestKickoffClientErrorsAreInvalid(t *testing.T) {
+	files := bulkimport.NewInMemoryFileStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: &memoryWriter{resources: map[string]*types.ResourceEnvelope{}}, Files: files},
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	_, err = svc.Kickoff(context.Background(), bulkimport.KickoffRequest{})
+	if core.KindOf(err) != core.ErrorKindInvalid {
+		t.Fatalf("empty inputs kind = %s", core.KindOf(err))
+	}
+	_, err = svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		InputFormat: "text/csv",
+		Inputs:      []bulkimport.InputFile{{Type: "Patient", NDJSON: []byte(`{"resourceType":"Patient","id":"p1"}`)}},
+	})
+	if core.KindOf(err) != core.ErrorKindInvalid {
+		t.Fatalf("format kind = %s", core.KindOf(err))
+	}
+}
+
+type cancelOnCreateWriter struct {
+	memoryWriter
+	cancel func()
+	once   bool
+}
+
+func (w *cancelOnCreateWriter) Create(ctx context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	if w.cancel != nil && !w.once {
+		w.once = true
+		w.cancel()
+	}
+	return w.memoryWriter.Create(ctx, resource)
+}
+
+func TestRunJobDoesNotCompleteCancelledJob(t *testing.T) {
+	writer := &cancelOnCreateWriter{memoryWriter: memoryWriter{resources: map[string]*types.ResourceEnvelope{}}}
+	files := bulkimport.NewInMemoryFileStore()
+	var svc *bulkimport.Service
+	writer.cancel = func() {
+		if err := svc.Cancel(context.Background(), "job-cancel"); err != nil {
+			t.Errorf("Cancel: %v", err)
+		}
+	}
+	var err error
+	svc, err = bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: writer, Files: files},
+		NewID:    func() string { return "job-cancel" },
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	job, err := svc.Kickoff(context.Background(), bulkimport.KickoffRequest{
+		Inputs: []bulkimport.InputFile{{
+			Type:   "Patient",
+			NDJSON: []byte(`{"resourceType":"Patient","id":"p1"}`),
+		}},
+	})
+	if err != nil {
+		t.Fatalf("Kickoff: %v", err)
+	}
+	if job.Status != bulkimport.StatusCancelled {
+		t.Fatalf("status = %s, want cancelled", job.Status)
+	}
+}
diff --git a/pkg/bulkimport/store.go b/pkg/bulkimport/store.go
new file mode 100644
index 00000000..a31c286d
--- /dev/null
+++ b/pkg/bulkimport/store.go
@@ -0,0 +1,125 @@
+package bulkimport
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+// JobStore persists bulk import job state.
+type JobStore interface {
+	Create(ctx context.Context, job Job) error
+	Get(ctx context.Context, id string) (*Job, error)
+	Update(ctx context.Context, job Job) error
+}
+
+// FileStore stores import input and error artifact bytes keyed by relative path.
+type FileStore interface {
+	Put(ctx context.Context, path string, data []byte, contentType string) error
+	Get(ctx context.Context, path string) ([]byte, string, error)
+	Delete(ctx context.Context, path string) error
+}
+
+// InMemoryJobStore is a concurrent-safe JobStore for tests and local use.
+type InMemoryJobStore struct {
+	mu   sync.RWMutex
+	jobs map[string]Job
+}
+
+// NewInMemoryJobStore constructs an empty in-memory job store.
+func NewInMemoryJobStore() *InMemoryJobStore {
+	return &InMemoryJobStore{jobs: make(map[string]Job)}
+}
+
+func (s *InMemoryJobStore) Create(_ context.Context, job Job) error {
+	if job.ID == "" {
+		return fmt.Errorf("import: job id is required")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, exists := s.jobs[job.ID]; exists {
+		return fmt.Errorf("import: duplicate job %q", job.ID)
+	}
+	s.jobs[job.ID] = job
+	return nil
+}
+
+func (s *InMemoryJobStore) Get(_ context.Context, id string) (*Job, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	job, ok := s.jobs[id]
+	if !ok {
+		return nil, nil
+	}
+	copy := job
+	return &copy, nil
+}
+
+func (s *InMemoryJobStore) Update(_ context.Context, job Job) error {
+	if job.ID == "" {
+		return fmt.Errorf("import: job id is required")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	existing, exists := s.jobs[job.ID]
+	if !exists {
+		return fmt.Errorf("import: job %q not found", job.ID)
+	}
+	// Cancellation wins over complete/in-progress writes that raced after a
+	// stale Get. completeJob re-reads, but this store-level guard closes TOCTOU.
+	if existing.Status == StatusCancelled || existing.CancelRequested {
+		if job.Status != StatusCancelled {
+			existing.Status = StatusCancelled
+			existing.CancelRequested = true
+			s.jobs[job.ID] = existing
+			return nil
+		}
+		job.Status = StatusCancelled
+		job.CancelRequested = true
+	}
+	s.jobs[job.ID] = job
+	return nil
+}
+
+// InMemoryFileStore is a concurrent-safe FileStore for tests and local use.
+type InMemoryFileStore struct {
+	mu    sync.RWMutex
+	files map[string]storedFile
+}
+
+type storedFile struct {
+	data        []byte
+	contentType string
+}
+
+// NewInMemoryFileStore constructs an empty in-memory file store.
+func NewInMemoryFileStore() *InMemoryFileStore {
+	return &InMemoryFileStore{files: make(map[string]storedFile)}
+}
+
+func (s *InMemoryFileStore) Put(_ context.Context, path string, data []byte, contentType string) error {
+	if path == "" {
+		return fmt.Errorf("import: file path is required")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.files[path] = storedFile{data: append([]byte(nil), data...), contentType: contentType}
+	return nil
+}
+
+func (s *InMemoryFileStore) Get(_ context.Context, path string) ([]byte, string, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	file, ok := s.files[path]
+	if !ok {
+		return nil, "", fmt.Errorf("import: file %q not found", path)
+	}
+	return append([]byte(nil), file.data...), file.contentType, nil
+}
+
+func (s *InMemoryFileStore) Delete(_ context.Context, path string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.files, path)
+	return nil
+}
diff --git a/pkg/bulkimport/store_test.go b/pkg/bulkimport/store_test.go
new file mode 100644
index 00000000..e9c887c6
--- /dev/null
+++ b/pkg/bulkimport/store_test.go
@@ -0,0 +1,77 @@
+package bulkimport_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
+)
+
+func TestInMemoryJobStoreUpdateDoesNotUncancel(t *testing.T) {
+	ctx := context.Background()
+	store := bulkimport.NewInMemoryJobStore()
+	job := bulkimport.Job{ID: "job-1", Status: bulkimport.StatusInProgress}
+	if err := store.Create(ctx, job); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	cancelled := job
+	cancelled.Status = bulkimport.StatusCancelled
+	cancelled.CancelRequested = true
+	if err := store.Update(ctx, cancelled); err != nil {
+		t.Fatalf("Update cancelled: %v", err)
+	}
+
+	complete := job
+	complete.Status = bulkimport.StatusComplete
+	complete.Progress = "100%"
+	complete.CancelRequested = false
+	if err := store.Update(ctx, complete); err != nil {
+		t.Fatalf("Update complete: %v", err)
+	}
+
+	got, err := store.Get(ctx, job.ID)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	if got == nil {
+		t.Fatal("expected job")
+	}
+	if got.Status != bulkimport.StatusCancelled {
+		t.Fatalf("status = %s, want cancelled", got.Status)
+	}
+	if !got.CancelRequested {
+		t.Fatal("expected CancelRequested to remain set")
+	}
+}
+
+func TestInMemoryJobStoreUpdateHonorsCancelRequested(t *testing.T) {
+	ctx := context.Background()
+	store := bulkimport.NewInMemoryJobStore()
+	job := bulkimport.Job{
+		ID:              "job-2",
+		Status:          bulkimport.StatusInProgress,
+		CancelRequested: true,
+	}
+	if err := store.Create(ctx, job); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	complete := job
+	complete.Status = bulkimport.StatusComplete
+	complete.CancelRequested = false
+	if err := store.Update(ctx, complete); err != nil {
+		t.Fatalf("Update complete: %v", err)
+	}
+
+	got, err := store.Get(ctx, job.ID)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	if got.Status != bulkimport.StatusCancelled {
+		t.Fatalf("status = %s, want cancelled", got.Status)
+	}
+	if !got.CancelRequested {
+		t.Fatal("expected CancelRequested to remain set")
+	}
+}
diff --git a/pkg/bulkimport/types.go b/pkg/bulkimport/types.go
new file mode 100644
index 00000000..2372bc4d
--- /dev/null
+++ b/pkg/bulkimport/types.go
@@ -0,0 +1,74 @@
+package bulkimport
+
+import "time"
+
+// JobStatus tracks bulk import lifecycle state.
+type JobStatus string
+
+const (
+	StatusInProgress JobStatus = "in-progress"
+	StatusComplete   JobStatus = "complete"
+	StatusError      JobStatus = "error"
+	StatusCancelled  JobStatus = "cancelled"
+)
+
+const InputFormatNDJSON = "application/fhir+ndjson"
+
+// InputFile is one NDJSON source in a kickoff request.
+type InputFile struct {
+	Type   string `json:"type"`
+	URL    string `json:"url,omitempty"`
+	NDJSON []byte `json:"-"`
+}
+
+// CountFile summarizes imported resources of one type.
+type CountFile struct {
+	Type  string `json:"type"`
+	Count int    `json:"count"`
+	URL   string `json:"url,omitempty"`
+}
+
+// ErrorFile describes one error NDJSON artifact.
+type ErrorFile struct {
+	Type string `json:"type"`
+	URL  string `json:"url"`
+}
+
+// KickoffRequest captures import parameters from an HTTP kickoff.
+type KickoffRequest struct {
+	TenantID       string
+	PrincipalID    string
+	InputFormat    string
+	Inputs         []InputFile
+	RequestURL     string
+	RequiresAccess bool
+}
+
+// Job is a durable bulk import job record.
+type Job struct {
+	ID              string
+	Status          JobStatus
+	Request         KickoffRequest
+	TransactionTime time.Time
+	Progress        string
+	Output          []CountFile
+	Errors          []ErrorFile
+	CreatedAt       time.Time
+	CompletedAt     time.Time
+	LastError       string
+	CancelRequested bool
+}
+
+// Manifest is the completed import manifest returned to clients.
+type Manifest struct {
+	TransactionTime     time.Time   `json:"transactionTime"`
+	Request             string      `json:"request"`
+	RequiresAccessToken bool        `json:"requiresAccessToken"`
+	Output              []CountFile `json:"output"`
+	Error               []ErrorFile `json:"error,omitempty"`
+}
+
+// JobPayload is the background job payload for bulk import execution.
+type JobPayload struct {
+	JobID string `json:"jobId"`
+}
diff --git a/pkg/http/README.md b/pkg/http/README.md
index 0b331163..3263ad3f 100644
--- a/pkg/http/README.md
+++ b/pkg/http/README.md
@@ -64,6 +64,10 @@ Base path defaults to `/fhir` (configurable via `Config.BasePath`).
 | `GET` | `/fhir/$export/status/{jobId}` | Poll status or fetch manifest | 202 in progress, 200 complete |
 | `DELETE` | `/fhir/$export/status/{jobId}` | Cancel export | 202 |
 | `GET` | `/fhir/$export/files/{jobId}/{file}` | Download NDJSON artifact | 200 |
+| `POST` | `/fhir/$import` | System bulk import kickoff (`Prefer: respond-async`, Parameters + NDJSON) | 202 + `Content-Location` when `BulkImportService` configured |
+| `GET` | `/fhir/$import/status/{jobId}` | Poll import status or fetch manifest | 202 in progress, 200 complete |
+| `DELETE` | `/fhir/$import/status/{jobId}` | Cancel import | 202 |
+| `GET` | `/fhir/$import/files/{jobId}/{file}` | Download import error NDJSON artifact | 200 |
 | `GET`/`POST` | `/fhir/$operation` or resource operation path | Custom operation | 200 + returned resource |
 | `POST` | `/sync/push` | Sync push (via `NewRootHandlerWithSyncMiddleware`) | 200 + results |
 | `GET` | `/sync/pull` | Sync pull (via `NewRootHandlerWithSyncMiddleware`) | 200 + events |
@@ -87,6 +91,7 @@ SDC operations (`$populate`, `$assemble`, `$validate`, `$extract`, adaptive ques
 ### Deferred
 
 - Configure `BulkExportService` (typically `pkg/export.Service` wired by `pkg/runtime`) to enable Bulk Data export
+- Configure `BulkImportService` (typically `pkg/bulkimport.Service` wired by `pkg/runtime`) to enable Bulk Data import
 - Full CapabilityStatement conformance coverage
 - SMART metadata and built-in token runtime
 
diff --git a/pkg/http/bulk.go b/pkg/http/bulk.go
index 975ef7fa..63dcf361 100644
--- a/pkg/http/bulk.go
+++ b/pkg/http/bulk.go
@@ -34,7 +34,7 @@ func (h *handler) handleBulkExport(w http.ResponseWriter, r *http.Request, route
 			writeError(w, err)
 			return
 		}
-		if strings.ToLower(r.Header.Get("Prefer")) != "respond-async" {
+		if !prefersRespondAsync(r.Header.Get("Prefer")) {
 			writeError(w, invalidRequest("Prefer: respond-async is required for bulk export kickoff", nil))
 			return
 		}
@@ -167,6 +167,22 @@ func parseCSVParam(raw string) []string {
 	return out
 }
 
+// prefersRespondAsync reports whether a Prefer header requests async processing.
+// It accepts the FHIR Bulk Data token by itself or as one comma-separated
+// preference, for example "respond-async, wait=10".
+func prefersRespondAsync(header string) bool {
+	for _, part := range strings.Split(header, ",") {
+		token := strings.TrimSpace(part)
+		if i := strings.IndexAny(token, "; "); i >= 0 {
+			token = token[:i]
+		}
+		if strings.EqualFold(token, "respond-async") {
+			return true
+		}
+	}
+	return false
+}
+
 func notFound(message string, args ...any) error {
 	return &core.ServiceError{
 		Kind:    core.ErrorKindNotFound,
diff --git a/pkg/http/bulk_import.go b/pkg/http/bulk_import.go
new file mode 100644
index 00000000..35f6d6d6
--- /dev/null
+++ b/pkg/http/bulk_import.go
@@ -0,0 +1,179 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
+)
+
+// BulkImportService handles FHIR Bulk Data import kickoff, polling, cancellation,
+// and error-artifact download.
+type BulkImportService interface {
+	Kickoff(ctx context.Context, req bulkimport.KickoffRequest) (*bulkimport.Job, error)
+	GetJob(ctx context.Context, jobID string) (*bulkimport.Job, error)
+	Cancel(ctx context.Context, jobID string) error
+	Manifest(job *bulkimport.Job) *bulkimport.Manifest
+	StatusURL(jobID string) string
+	GetFile(ctx context.Context, jobID, filename string) ([]byte, string, error)
+}
+
+func (h *handler) handleBulkImport(w http.ResponseWriter, r *http.Request) {
+	if h.cfg.BulkImportService == nil {
+		writeError(w, notImplementedEndpoint(r.URL.Path))
+		return
+	}
+	if r.Method != http.MethodPost {
+		writeMethodNotAllowed(w, r.Method, http.MethodPost)
+		return
+	}
+	if !prefersRespondAsync(r.Header.Get("Prefer")) {
+		writeError(w, invalidRequest("Prefer: respond-async is required for bulk import kickoff", nil))
+		return
+	}
+	body, err := readBody(r)
+	if err != nil {
+		writeError(w, err)
+		return
+	}
+	req, err := bulkimport.ParseParametersKickoff(body)
+	if err != nil {
+		writeError(w, invalidRequest(err.Error(), err))
+		return
+	}
+	if err := h.authorizeImportKickoff(r.Context(), req); err != nil {
+		writeError(w, err)
+		return
+	}
+	req.RequestURL = r.Method + " " + r.URL.RequestURI()
+	req.RequiresAccess = h.cfg.AuthChecker != nil
+	if principal, tenant, ok := identityFromContext(r.Context()); ok {
+		req.TenantID = tenant.TenantID
+		req.PrincipalID = principal.ID
+	}
+	job, err := h.cfg.BulkImportService.Kickoff(r.Context(), req)
+	if err != nil {
+		writeError(w, err)
+		return
+	}
+	w.Header().Set("Content-Location", h.cfg.BulkImportService.StatusURL(job.ID))
+	w.WriteHeader(http.StatusAccepted)
+}
+
+func (h *handler) handleBulkImportStatus(w http.ResponseWriter, r *http.Request, jobID string) {
+	if h.cfg.BulkImportService == nil {
+		writeError(w, notImplementedEndpoint(r.URL.Path))
+		return
+	}
+	switch r.Method {
+	case http.MethodGet:
+		if err := h.authorizeRead(r.Context(), "Parameters", ""); err != nil {
+			writeError(w, err)
+			return
+		}
+		job, err := h.cfg.BulkImportService.GetJob(r.Context(), jobID)
+		if err != nil {
+			writeError(w, err)
+			return
+		}
+		if job == nil {
+			writeError(w, notFound("import job %q not found", jobID))
+			return
+		}
+		switch job.Status {
+		case bulkimport.StatusComplete:
+			manifest := h.cfg.BulkImportService.Manifest(job)
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_ = json.NewEncoder(w).Encode(manifest)
+		case bulkimport.StatusInProgress:
+			if job.Progress != "" {
+				w.Header().Set("X-Progress", job.Progress)
+			}
+			w.WriteHeader(http.StatusAccepted)
+		case bulkimport.StatusCancelled:
+			writeError(w, invalidRequest("import job cancelled", nil))
+		default:
+			writeError(w, invalidRequest(job.LastError, nil))
+		}
+	case http.MethodDelete:
+		if err := h.authorizeImportWrite(r.Context()); err != nil {
+			writeError(w, err)
+			return
+		}
+		if err := h.cfg.BulkImportService.Cancel(r.Context(), jobID); err != nil {
+			writeError(w, err)
+			return
+		}
+		w.WriteHeader(http.StatusAccepted)
+	default:
+		writeMethodNotAllowed(w, r.Method, http.MethodGet, http.MethodDelete)
+	}
+}
+
+func (h *handler) handleBulkImportFile(w http.ResponseWriter, r *http.Request, jobID, filename string) {
+	if h.cfg.BulkImportService == nil {
+		writeError(w, notImplementedEndpoint(r.URL.Path))
+		return
+	}
+	if r.Method != http.MethodGet {
+		writeMethodNotAllowed(w, r.Method, http.MethodGet)
+		return
+	}
+	if err := h.authorizeRead(r.Context(), "Binary", ""); err != nil {
+		writeError(w, err)
+		return
+	}
+	data, contentType, err := h.cfg.BulkImportService.GetFile(r.Context(), jobID, filename)
+	if err != nil {
+		writeError(w, notFound("import file not found"))
+		return
+	}
+	if contentType == "" {
+		contentType = "application/fhir+ndjson"
+	}
+	w.Header().Set("Content-Type", contentType)
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write(data)
+}
+
+// authorizeImportKickoff requires create and update grants on every input
+// resource type before $import starts. system/*.write or type-level write on
+// each input type is the bar; a Parameters-only grant must not import arbitrary
+// types. Empty inputs fall back to Parameters-level write.
+func (h *handler) authorizeImportKickoff(ctx context.Context, req bulkimport.KickoffRequest) error {
+	seen := make(map[string]struct{}, len(req.Inputs))
+	var types []string
+	for _, input := range req.Inputs {
+		resourceType := strings.TrimSpace(input.Type)
+		if resourceType == "" {
+			continue
+		}
+		if _, ok := seen[resourceType]; ok {
+			continue
+		}
+		seen[resourceType] = struct{}{}
+		types = append(types, resourceType)
+	}
+	if len(types) == 0 {
+		return h.authorizeImportWrite(ctx)
+	}
+	for _, resourceType := range types {
+		if err := h.authorizeWrite(ctx, "create", resourceType, ""); err != nil {
+			return err
+		}
+		if err := h.authorizeWrite(ctx, "update", resourceType, ""); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// authorizeImportWrite requires write access for system-level $import when input
+// types are not available (empty kickoff or cancel). SMART has no dedicated
+// import scope; a read-only export token must not be able to write.
+func (h *handler) authorizeImportWrite(ctx context.Context) error {
+	return h.authorizeWrite(ctx, "operation", "Parameters", "")
+}
diff --git a/pkg/http/bulk_prefer_test.go b/pkg/http/bulk_prefer_test.go
new file mode 100644
index 00000000..ba7870b9
--- /dev/null
+++ b/pkg/http/bulk_prefer_test.go
@@ -0,0 +1,24 @@
+package http
+
+import "testing"
+
+func TestPrefersRespondAsync(t *testing.T) {
+	cases := []struct {
+		header string
+		want   bool
+	}{
+		{"respond-async", true},
+		{"Respond-Async", true},
+		{"respond-async, wait=10", true},
+		{"wait=10, respond-async", true},
+		{"respond-async; wait=10", true},
+		{"", false},
+		{"return=minimal", false},
+		{"respond-async-please", false},
+	}
+	for _, tc := range cases {
+		if got := prefersRespondAsync(tc.header); got != tc.want {
+			t.Errorf("prefersRespondAsync(%q) = %v, want %v", tc.header, got, tc.want)
+		}
+	}
+}
diff --git a/pkg/http/bulk_test.go b/pkg/http/bulk_test.go
index f2240415..4c21ed77 100644
--- a/pkg/http/bulk_test.go
+++ b/pkg/http/bulk_test.go
@@ -5,33 +5,57 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/degoke/health-ai-stack/pkg/auth"
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
 	"github.com/degoke/health-ai-stack/pkg/core"
 	"github.com/degoke/health-ai-stack/pkg/export"
 	hahttp "github.com/degoke/health-ai-stack/pkg/http"
+	"github.com/degoke/health-ai-stack/pkg/registry"
 	"github.com/degoke/health-ai-stack/pkg/types"
 )
 
 type bulkAuthChecker struct {
-	allow bool
-	calls int
+	allowExport    bool
+	allowWrite     bool
+	allowRead      bool
+	allowWriteType map[string]bool
+	exportCalls    int
+	writeCalls     int
+	readCalls      int
+	writeTypes     []string
 }
 
 func (c *bulkAuthChecker) AuthorizeRead(context.Context, auth.Principal, auth.TenantContext, string, string) (auth.Decision, error) {
-	return auth.Allow("ok"), nil
+	c.readCalls++
+	if c.allowRead {
+		return auth.Allow("ok"), nil
+	}
+	return auth.Deny("denied"), nil
 }
-func (c *bulkAuthChecker) AuthorizeWrite(context.Context, auth.Principal, auth.TenantContext, string, string, string) (auth.Decision, error) {
-	return auth.Allow("ok"), nil
+func (c *bulkAuthChecker) AuthorizeWrite(_ context.Context, _ auth.Principal, _ auth.TenantContext, _, resourceType, _ string) (auth.Decision, error) {
+	c.writeCalls++
+	c.writeTypes = append(c.writeTypes, resourceType)
+	if len(c.allowWriteType) > 0 {
+		if c.allowWriteType[resourceType] {
+			return auth.Allow("ok"), nil
+		}
+		return auth.Deny("denied"), nil
+	}
+	if c.allowWrite {
+		return auth.Allow("ok"), nil
+	}
+	return auth.Deny("denied"), nil
 }
 func (c *bulkAuthChecker) AuthorizeSearch(context.Context, auth.Principal, auth.TenantContext, string) (auth.Decision, error) {
 	return auth.Allow("ok"), nil
 }
 func (c *bulkAuthChecker) AuthorizeExport(_ context.Context, _ auth.Principal, _ auth.TenantContext, _ string) (auth.Decision, error) {
-	c.calls++
-	if c.allow {
+	c.exportCalls++
+	if c.allowExport {
 		return auth.Allow("ok"), nil
 	}
 	return auth.Deny("denied"), nil
@@ -78,7 +102,7 @@ func newBulkExportService(t *testing.T) hahttp.BulkExportService {
 }
 
 func TestBulkExportKickoffPollManifest(t *testing.T) {
-	checker := &bulkAuthChecker{allow: true}
+	checker := &bulkAuthChecker{allowExport: true}
 	handler := newTestHandler(t, hahttp.Config{
 		ResourceService:   &bulkResourceService{},
 		BulkExportService: newBulkExportService(t),
@@ -99,7 +123,7 @@ func TestBulkExportKickoffPollManifest(t *testing.T) {
 	if statusURL == "" {
 		t.Fatal("missing Content-Location")
 	}
-	if checker.calls == 0 {
+	if checker.exportCalls == 0 {
 		t.Fatal("expected export authorization")
 	}
 
@@ -127,7 +151,7 @@ func TestBulkExportUnauthorized(t *testing.T) {
 		PrincipalResolver: func(_ context.Context, _ *http.Request) (auth.Principal, auth.TenantContext, error) {
 			return auth.Principal{ID: "user-1", Kind: auth.KindUser}, auth.TenantContext{TenantID: "tenant-a"}, nil
 		},
-		AuthChecker: &bulkAuthChecker{allow: false},
+		AuthChecker: &bulkAuthChecker{allowExport: false},
 	})
 	req := httptest.NewRequest(http.MethodGet, "/fhir/$export", nil)
 	req.Header.Set("Prefer", "respond-async")
@@ -145,3 +169,300 @@ func TestBulkExportReturnsNotImplementedWithoutService(t *testing.T) {
 		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
 	}
 }
+
+type importWriter struct {
+	resources map[string]*types.ResourceEnvelope
+}
+
+func (w *importWriter) key(resourceType, id string) string { return resourceType + "/" + id }
+
+func (w *importWriter) Create(_ context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	if w.resources == nil {
+		w.resources = map[string]*types.ResourceEnvelope{}
+	}
+	cp := *resource
+	w.resources[w.key(resource.ResourceType, resource.ID)] = &cp
+	return &cp, nil
+}
+
+func (w *importWriter) Update(ctx context.Context, resource *types.ResourceEnvelope) (*types.ResourceEnvelope, error) {
+	return w.Create(ctx, resource)
+}
+
+func (w *importWriter) Read(_ context.Context, resourceType, id string) (*types.ResourceEnvelope, error) {
+	if env := w.resources[w.key(resourceType, id)]; env != nil {
+		return env, nil
+	}
+	return nil, &core.ServiceError{Kind: core.ErrorKindNotFound, Message: "not found"}
+}
+
+func newBulkImportService(t *testing.T, writer *importWriter) hahttp.BulkImportService {
+	t.Helper()
+	files := bulkimport.NewInMemoryFileStore()
+	svc, err := bulkimport.NewService(bulkimport.Config{
+		Jobs:     bulkimport.NewInMemoryJobStore(),
+		Files:    files,
+		Executor: &bulkimport.Executor{Resources: writer, Files: files},
+		BasePath: "/fhir",
+		NewID:    func() string { return "import-job-1" },
+	})
+	if err != nil {
+		t.Fatalf("NewService: %v", err)
+	}
+	return svc
+}
+
+func TestBulkImportKickoffPollManifest(t *testing.T) {
+	checker := &bulkAuthChecker{allowWrite: true, allowRead: true}
+	writer := &importWriter{}
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, writer),
+		PrincipalResolver: func(_ context.Context, _ *http.Request) (auth.Principal, auth.TenantContext, error) {
+			return auth.Principal{ID: "svc-1", Kind: auth.KindService}, auth.TenantContext{TenantID: "tenant-a"}, nil
+		},
+		AuthChecker: checker,
+	})
+
+	body := `{"resourceType":"Parameters","parameter":[{"name":"inputFormat","valueCode":"application/fhir+ndjson"},{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}"}]}]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("kickoff status = %d body = %s", rec.Code, rec.Body.String())
+	}
+	statusURL := rec.Header().Get("Content-Location")
+	if statusURL == "" {
+		t.Fatal("missing Content-Location")
+	}
+	if checker.writeCalls == 0 {
+		t.Fatal("expected import write authorization")
+	}
+	if checker.exportCalls != 0 {
+		t.Fatalf("import kickoff used export auth (%d calls)", checker.exportCalls)
+	}
+
+	pollReq := httptest.NewRequest(http.MethodGet, statusURL, nil)
+	pollReq.Header.Set("Accept", "application/json")
+	pollRec := httptest.NewRecorder()
+	handler.ServeHTTP(pollRec, pollReq)
+	if pollRec.Code != http.StatusOK {
+		t.Fatalf("poll status = %d body = %s", pollRec.Code, pollRec.Body.String())
+	}
+	var manifest map[string]any
+	if err := json.Unmarshal(pollRec.Body.Bytes(), &manifest); err != nil {
+		t.Fatalf("manifest decode: %v", err)
+	}
+	output, ok := manifest["output"].([]any)
+	if !ok || len(output) != 1 {
+		t.Fatalf("manifest output = %#v", manifest["output"])
+	}
+	if _, err := writer.Read(context.Background(), "Patient", "p1"); err != nil {
+		t.Fatalf("imported patient: %v", err)
+	}
+}
+
+func TestBulkImportRequiresPreferAsync(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, &importWriter{}),
+	})
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(`{"resourceType":"Parameters","parameter":[{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{}"}]}]}`))
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkImportReturnsNotImplementedWithoutService(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{ResourceService: &fakeResourceService{}})
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(`{}`))
+	req.Header.Set("Prefer", "respond-async")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusNotImplemented {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestCapabilityStatementAdvertisesImportWhenConfigured(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &fakeResourceService{},
+		BulkImportService: newBulkImportService(t, &importWriter{}),
+		CapabilitySource: fakeCapabilitySource{snapshot: registry.CapabilitySnapshot{
+			FHIRVersion: "4.0.1",
+			Resources:   []registry.ResourceCapability{{ResourceType: "Patient"}},
+		}},
+	})
+	rec := doRequest(t, handler, http.MethodGet, "/fhir/metadata", nil)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), `"name":"import"`) {
+		t.Fatalf("expected $import advertised, got %s", rec.Body.String())
+	}
+}
+
+func TestBulkImportDeniedWhenAnyInputTypeUnauthorized(t *testing.T) {
+	writer := &importWriter{}
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, writer),
+		PrincipalResolver: func(_ context.Context, _ *http.Request) (auth.Principal, auth.TenantContext, error) {
+			return auth.Principal{ID: "svc-1", Kind: auth.KindService}, auth.TenantContext{TenantID: "tenant-a"}, nil
+		},
+		AuthChecker: &bulkAuthChecker{
+			allowWrite:     true,
+			allowRead:      true,
+			allowWriteType: map[string]bool{"Patient": true},
+		},
+	})
+	body := `{"resourceType":"Parameters","parameter":[` +
+		`{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}"}]},` +
+		`{"name":"input","part":[{"name":"type","valueCode":"Observation"},{"name":"valueString","valueString":"{\"resourceType\":\"Observation\",\"id\":\"o1\",\"status\":\"final\",\"code\":{\"text\":\"x\"}}"}]}` +
+		`]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusForbidden {
+		t.Fatalf("status = %d body = %s", rec.Code, rec.Body.String())
+	}
+	if _, err := writer.Read(context.Background(), "Patient", "p1"); err == nil {
+		t.Fatal("kickoff must not import Patient when Observation write is denied")
+	}
+}
+
+func TestBulkImportUnauthorizedWithoutWrite(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, &importWriter{}),
+		PrincipalResolver: func(_ context.Context, _ *http.Request) (auth.Principal, auth.TenantContext, error) {
+			return auth.Principal{ID: "svc-1", Kind: auth.KindService}, auth.TenantContext{TenantID: "tenant-a"}, nil
+		},
+		AuthChecker: &bulkAuthChecker{allowExport: true, allowWrite: false, allowRead: true},
+	})
+	body := `{"resourceType":"Parameters","parameter":[{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}"}]}]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusForbidden {
+		t.Fatalf("status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkImportInvalidParameters(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, &importWriter{}),
+	})
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(`{"resourceType":"Patient","id":"p1"}`))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkImportURLWithoutLoaderIsBadRequest(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, &importWriter{}),
+	})
+	body := `{"resourceType":"Parameters","parameter":[{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"url","valueUri":"https://example.test/Patient.ndjson"}]}]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkImportAcceptsPreferCommaForm(t *testing.T) {
+	writer := &importWriter{}
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, writer),
+	})
+	body := `{"resourceType":"Parameters","parameter":[{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}"}]}]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async, wait=10")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("kickoff status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkExportAcceptsPreferCommaForm(t *testing.T) {
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkExportService: newBulkExportService(t),
+	})
+	req := httptest.NewRequest(http.MethodGet, "/fhir/$export?_type=Patient", nil)
+	req.Header.Set("Prefer", "respond-async, wait=10")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("kickoff status = %d body = %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBulkImportErrorFileDownload(t *testing.T) {
+	writer := &importWriter{}
+	handler := newTestHandler(t, hahttp.Config{
+		ResourceService:   &bulkResourceService{},
+		BulkImportService: newBulkImportService(t, writer),
+	})
+	body := `{"resourceType":"Parameters","parameter":[{"name":"input","part":[{"name":"type","valueCode":"Patient"},{"name":"valueString","valueString":"{\"resourceType\":\"Patient\",\"id\":\"p1\"}\nnot-json"}]}]}`
+	req := httptest.NewRequest(http.MethodPost, "/fhir/$import", strings.NewReader(body))
+	req.Header.Set("Prefer", "respond-async")
+	req.Header.Set("Content-Type", "application/fhir+json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("kickoff status = %d body = %s", rec.Code, rec.Body.String())
+	}
+	statusURL := rec.Header().Get("Content-Location")
+	pollReq := httptest.NewRequest(http.MethodGet, statusURL, nil)
+	pollReq.Header.Set("Accept", "application/json")
+	pollRec := httptest.NewRecorder()
+	handler.ServeHTTP(pollRec, pollReq)
+	if pollRec.Code != http.StatusOK {
+		t.Fatalf("poll status = %d body = %s", pollRec.Code, pollRec.Body.String())
+	}
+	var manifest map[string]any
+	if err := json.Unmarshal(pollRec.Body.Bytes(), &manifest); err != nil {
+		t.Fatalf("manifest decode: %v", err)
+	}
+	errors, ok := manifest["error"].([]any)
+	if !ok || len(errors) != 1 {
+		t.Fatalf("manifest error = %#v", manifest["error"])
+	}
+	errObj, _ := errors[0].(map[string]any)
+	fileURL, _ := errObj["url"].(string)
+	if fileURL == "" {
+		t.Fatal("missing error file url")
+	}
+	fileReq := httptest.NewRequest(http.MethodGet, fileURL, nil)
+	fileRec := httptest.NewRecorder()
+	handler.ServeHTTP(fileRec, fileReq)
+	if fileRec.Code != http.StatusOK {
+		t.Fatalf("file status = %d body = %s", fileRec.Code, fileRec.Body.String())
+	}
+	if !strings.Contains(fileRec.Body.String(), "OperationOutcome") {
+		t.Fatalf("error artifact = %s", fileRec.Body.String())
+	}
+}
diff --git a/pkg/http/bundle.go b/pkg/http/bundle.go
index b56c1374..92d98f05 100644
--- a/pkg/http/bundle.go
+++ b/pkg/http/bundle.go
@@ -135,6 +135,7 @@ type capabilityFlags struct {
 	TerminologyInstall bool
 	TerminologyEnable  bool
 	ConformanceRefresh bool
+	BulkImport         bool
 }
 
 func capabilityFromConfig(cfg Config) capabilityFlags {
@@ -156,6 +157,7 @@ func capabilityFromConfig(cfg Config) capabilityFlags {
 		TerminologyInstall: cfg.TerminologyInstallService != nil,
 		TerminologyEnable:  cfg.TerminologyEnableService != nil,
 		ConformanceRefresh: cfg.ConformanceRefresher != nil,
+		BulkImport:         cfg.BulkImportService != nil,
 	}
 }
 
@@ -441,6 +443,12 @@ func systemOperations(flags capabilityFlags) []map[string]string {
 			"definition": "http://hl7.org/fhir/uv/bulkdata/OperationDefinition/export",
 		})
 	}
+	if flags.BulkImport {
+		ops = append(ops, map[string]string{
+			"name":       "import",
+			"definition": "http://hl7.org/fhir/uv/bulkdata/OperationDefinition/import",
+		})
+	}
 	if flags.ViewRun {
 		ops = append(ops, map[string]string{
 			"name": "viewdefinition-run", "definition": "http://hl7.org/fhir/uv/sql-on-fhir/OperationDefinition/ViewDefinition-run",
diff --git a/pkg/http/config.go b/pkg/http/config.go
index f651dc45..d89e8bae 100644
--- a/pkg/http/config.go
+++ b/pkg/http/config.go
@@ -133,6 +133,9 @@ type Config struct {
 	// BulkExportService handles FHIR Bulk Data export when configured.
 	BulkExportService BulkExportService
 
+	// BulkImportService handles FHIR Bulk Data import when configured.
+	BulkImportService BulkImportService
+
 	// ViewMaterializeService handles ViewDefinition/$materialize when configured.
 	ViewMaterializeService ViewMaterializeService
 
diff --git a/pkg/http/doc.go b/pkg/http/doc.go
index 91806b53..27ed7bc0 100644
--- a/pkg/http/doc.go
+++ b/pkg/http/doc.go
@@ -82,6 +82,10 @@
 //   - GET    /fhir/Patient/{id}/$everything  — patient compartment searchset
 //   - GET    /fhir/$export                     — system bulk export kickoff (async; requires BulkExportService)
 //   - GET    /fhir/Patient/$export and /fhir/Patient/{id}/$export — patient bulk export
+//   - POST   /fhir/$import                     — system bulk import kickoff (async; requires BulkImportService)
+//   - GET    /fhir/$import/status/{jobId}      — import status polling / manifest
+//   - DELETE /fhir/$import/status/{jobId}      — cancel import
+//   - GET    /fhir/$import/files/{jobId}/{file} — download import error NDJSON
 //   - GET    /fhir/Group/{id}/$export          — group bulk export kickoff (async)
 //   - GET    /fhir/$export/status/{jobId}      — export status polling / manifest
 //   - DELETE /fhir/$export/status/{jobId}      — cancel export
@@ -145,6 +149,7 @@
 // # Out of scope (MVP)
 //
 //   - BulkExportService — when nil, $export routes return 501
+//   - BulkImportService — when nil, $import routes return 501
 //   - Full CapabilityStatement conformance testing
 //   - SMART metadata, built-in OAuth2/SMART token runtime
 //   - gRPC or non-FHIR content types
diff --git a/pkg/http/handler.go b/pkg/http/handler.go
index ba73f8df..3ecb360a 100644
--- a/pkg/http/handler.go
+++ b/pkg/http/handler.go
@@ -150,6 +150,10 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			h.handleBulkExport(w, r, route)
 			return
 		}
+		if route.operation == "$import" {
+			h.handleBulkImport(w, r)
+			return
+		}
 		if route.operation == "$viewdefinition-run" {
 			h.handleViewDefinitionRun(w, r, parsedRoute{operation: route.operation})
 			return
@@ -171,6 +175,10 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		h.handleBulkExportStatus(w, r, route.jobID)
 	case routeBulkExportFile:
 		h.handleBulkExportFile(w, r, route.jobID, route.filename)
+	case routeBulkImportStatus:
+		h.handleBulkImportStatus(w, r, route.jobID)
+	case routeBulkImportFile:
+		h.handleBulkImportFile(w, r, route.jobID, route.filename)
 	case routeMaterializeStatus:
 		h.handleViewMaterializeStatus(w, r, route.jobID)
 	case routeViewExportStatus:
diff --git a/pkg/http/router.go b/pkg/http/router.go
index 327aac88..a1932b69 100644
--- a/pkg/http/router.go
+++ b/pkg/http/router.go
@@ -29,6 +29,8 @@ const (
 	routeSystemOperation
 	routeBulkExportStatus
 	routeBulkExportFile
+	routeBulkImportStatus
+	routeBulkImportFile
 	routeMaterializeStatus
 	routeViewExportStatus
 	routeViewExportFile
@@ -57,6 +59,29 @@ func parseRoute(basePath, requestPath string) (parsedRoute, error) {
 	}
 
 	parts := strings.Split(rel, "/")
+	if len(parts) >= 2 && parts[0] == "$import" {
+		switch parts[1] {
+		case "status":
+			if len(parts) != 3 {
+				return parsedRoute{}, fmt.Errorf("unsupported path %q", rel)
+			}
+			if err := validateID(parts[2]); err != nil {
+				return parsedRoute{}, err
+			}
+			return parsedRoute{kind: routeBulkImportStatus, jobID: parts[2]}, nil
+		case "files":
+			if len(parts) != 4 {
+				return parsedRoute{}, fmt.Errorf("unsupported path %q", rel)
+			}
+			if err := validateID(parts[2]); err != nil {
+				return parsedRoute{}, err
+			}
+			if err := validateExportFilename(parts[3]); err != nil {
+				return parsedRoute{}, err
+			}
+			return parsedRoute{kind: routeBulkImportFile, jobID: parts[2], filename: parts[3]}, nil
+		}
+	}
 	if len(parts) >= 2 && parts[0] == "$export" {
 		switch parts[1] {
 		case "status":
diff --git a/pkg/http/router_test.go b/pkg/http/router_test.go
index c43b0572..c09b4145 100644
--- a/pkg/http/router_test.go
+++ b/pkg/http/router_test.go
@@ -15,6 +15,32 @@ func TestParseRouteViewExportFileAcceptsArtifactFilename(t *testing.T) {
 	}
 }
 
+func TestParseRouteBulkImportStatus(t *testing.T) {
+	route, err := parseRoute("/fhir", "/fhir/$import/status/import-job-1")
+	if err != nil {
+		t.Fatalf("parseRoute: %v", err)
+	}
+	if route.kind != routeBulkImportStatus {
+		t.Fatalf("kind=%v", route.kind)
+	}
+	if route.jobID != "import-job-1" {
+		t.Fatalf("jobID=%q", route.jobID)
+	}
+}
+
+func TestParseRouteBulkImportFileAcceptsArtifactFilename(t *testing.T) {
+	route, err := parseRoute("/fhir", "/fhir/$import/files/import-job-1/error-0-Patient.ndjson")
+	if err != nil {
+		t.Fatalf("parseRoute: %v", err)
+	}
+	if route.kind != routeBulkImportFile {
+		t.Fatalf("kind=%v", route.kind)
+	}
+	if route.filename != "error-0-Patient.ndjson" {
+		t.Fatalf("filename=%q", route.filename)
+	}
+}
+
 func TestParseRouteBulkExportFileAcceptsArtifactFilename(t *testing.T) {
 	route, err := parseRoute("/fhir", "/fhir/$export/files/export-job-1/Patient.ndjson")
 	if err != nil {
diff --git a/pkg/jobs/types.go b/pkg/jobs/types.go
index eefe9d14..c1051a66 100644
--- a/pkg/jobs/types.go
+++ b/pkg/jobs/types.go
@@ -43,6 +43,9 @@ const (
 	TypeExportCSV = TypePrefixExport + "csv"
 	// TypeExportBulk schedules a FHIR Bulk Data NDJSON export run.
 	TypeExportBulk = TypePrefixExport + "bulk"
+	// TypeImportBulk is the import job type (FHIR Bulk Data NDJSON $import).
+	// The value keeps the export. prefix so existing job-queue wiring stays stable.
+	TypeImportBulk = TypePrefixExport + "import"
 	// TypeViewMaterialize schedules a ViewDefinition materialize run.
 	TypeViewMaterialize = TypePrefixView + "materialize"
 	// TypeViewExport schedules a ViewDefinition export run.
diff --git a/pkg/runtime/services.go b/pkg/runtime/services.go
index 6e7bb810..0ac77a24 100644
--- a/pkg/runtime/services.go
+++ b/pkg/runtime/services.go
@@ -2,6 +2,7 @@ package runtime
 
 import (
 	"github.com/degoke/health-ai-stack/pkg/analytics"
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
 	"github.com/degoke/health-ai-stack/pkg/core"
 	"github.com/degoke/health-ai-stack/pkg/export"
 	"github.com/degoke/health-ai-stack/pkg/fhirpath"
@@ -37,6 +38,8 @@ type ServiceContainer struct {
 
 	// BulkExportService handles FHIR Bulk Data export when job infrastructure is wired.
 	BulkExportService *export.Service
+	// BulkImportService handles FHIR Bulk Data import when job infrastructure is wired.
+	BulkImportService *bulkimport.Service
 
 	// AnalyticsRunner executes ViewDefinitions when analytics is enabled.
 	AnalyticsRunner *analytics.Runner
diff --git a/pkg/runtime/wire.go b/pkg/runtime/wire.go
index 01afd027..7672ad4b 100644
--- a/pkg/runtime/wire.go
+++ b/pkg/runtime/wire.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/degoke/health-ai-stack/pkg/analytics"
+	"github.com/degoke/health-ai-stack/pkg/bulkimport"
 	"github.com/degoke/health-ai-stack/pkg/conceptmap"
 	"github.com/degoke/health-ai-stack/pkg/core"
 	"github.com/degoke/health-ai-stack/pkg/export"
@@ -616,6 +617,27 @@ func (b *Builder) wireCommon(ctx context.Context, state *wireState, pc persisten
 			return fmt.Errorf("runtime: register bulk export handler: %w", err)
 		}
 		state.services.BulkExportService = exportSvc
+		importFiles := bulkimport.NewInMemoryFileStore()
+		importJobs := bulkimport.NewInMemoryJobStore()
+		importExecutor := &bulkimport.Executor{
+			Resources: state.services.ResourceService,
+			Files:     importFiles,
+		}
+		importSvc, err := bulkimport.NewService(bulkimport.Config{
+			Jobs:     importJobs,
+			Files:    importFiles,
+			Executor: importExecutor,
+			JobQueue: pc.jobStore,
+			Loader:   bulkimport.NewHTTPLoader(),
+			BasePath: "/fhir",
+		})
+		if err != nil {
+			return fmt.Errorf("runtime: bulk import service: %w", err)
+		}
+		if err := runner.Register(jobs.TypeImportBulk, importSvc.JobHandler()); err != nil {
+			return fmt.Errorf("runtime: register bulk import handler: %w", err)
+		}
+		state.services.BulkImportService = importSvc
 		moduleWorker := &modules.InstallWorker{
 			Manager:                    modManager,
 			Store:                      pc.jobStore,
@@ -754,6 +776,7 @@ func (b *Builder) wireCommon(ctx context.Context, state *wireState, pc persisten
 		PrincipalResolver:        b.httpPrincipalResolver,
 		AuthChecker:              b.httpAuthChecker,
 		BulkExportService:        state.services.BulkExportService,
+		BulkImportService:        state.services.BulkImportService,
 		ViewMaterializeService:   state.services.MaterializeService,
 		ViewRunService:           state.services.ViewRunService,
 		SQLQueryService:          state.services.SQLQueryService,