diff --git a/cmd/wfctl/type_registry.go b/cmd/wfctl/type_registry.go
index c3dfe207..31ae0e99 100644
--- a/cmd/wfctl/type_registry.go
+++ b/cmd/wfctl/type_registry.go
@@ -68,6 +68,14 @@ func KnownModuleTypes() map[string]ModuleTypeInfo {
 			ConfigKeys: []string{"address", "password", "db", "prefix", "defaultTTL"},
 		},
 
+		// configprovider plugin
+		"config.provider": {
+			Type:       "config.provider",
+			Plugin:     "configprovider",
+			Stateful:   false,
+			ConfigKeys: []string{"sources", "schema"},
+		},
+
 		// http plugin
 		"http.server": {
 			Type:       "http.server",
diff --git a/module/config_provider.go b/module/config_provider.go
new file mode 100644
index 00000000..a6e7faba
--- /dev/null
+++ b/module/config_provider.go
@@ -0,0 +1,287 @@
+// This file implements the config.provider module type and config registry.
+package module
+
+import (
+	"fmt"
+	"os"
+	"regexp"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/CrisisTextLine/modular"
+)
+
+// configKeyRegexp matches {{config "key"}}, {{ config "key" }}, {{config 'key'}},
+// or {{ config 'key' }} patterns. It handles single or double quotes and optional whitespace.
+var configKeyRegexp = regexp.MustCompile(`\{\{\s*config\s+["']([^"']+)["']\s*\}\}`)
+
+// SchemaEntry defines a single configuration key's metadata.
+type SchemaEntry struct {
+	Env       string `json:"env"`
+	Required  bool   `json:"required"`
+	Default   string `json:"default"`
+	Sensitive bool   `json:"sensitive"`
+	Desc      string `json:"desc"`
+}
+
+// ConfigRegistry is a thread-safe, immutable store of resolved configuration values.
+type ConfigRegistry struct {
+	mu        sync.RWMutex
+	values    map[string]string
+	sensitive map[string]bool
+	frozen    bool
+}
+
+// globalConfigRegistry is the singleton config registry used by the engine.
+var globalConfigRegistry = &ConfigRegistry{
+	values:    make(map[string]string),
+	sensitive: make(map[string]bool),
+}
+
+// GetConfigRegistry returns the global config registry singleton.
+func GetConfigRegistry() *ConfigRegistry {
+	return globalConfigRegistry
+}
+
+// NewConfigRegistry creates a fresh ConfigRegistry. Primarily used for testing.
+func NewConfigRegistry() *ConfigRegistry {
+	return &ConfigRegistry{
+		values:    make(map[string]string),
+		sensitive: make(map[string]bool),
+	}
+}
+
+// Set stores a value in the registry. Returns an error if the registry is frozen.
+func (r *ConfigRegistry) Set(key, value string, sensitive bool) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.frozen {
+		return fmt.Errorf("config registry is frozen; cannot set key %q", key)
+	}
+	r.values[key] = value
+	r.sensitive[key] = sensitive
+	return nil
+}
+
+// Get retrieves a value from the registry.
+func (r *ConfigRegistry) Get(key string) (string, bool) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	v, ok := r.values[key]
+	return v, ok
+}
+
+// IsSensitive returns whether a key is marked as sensitive.
+func (r *ConfigRegistry) IsSensitive(key string) bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.sensitive[key]
+}
+
+// Freeze makes the registry immutable. After calling Freeze, Set will return an error.
+func (r *ConfigRegistry) Freeze() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.frozen = true
+}
+
+// Reset clears all values and unfreezes the registry. Intended for testing.
+func (r *ConfigRegistry) Reset() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.values = make(map[string]string)
+	r.sensitive = make(map[string]bool)
+	r.frozen = false
+}
+
+// Keys returns all registered configuration key names.
+func (r *ConfigRegistry) Keys() []string {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	keys := make([]string, 0, len(r.values))
+	for k := range r.values {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// RedactedValue returns the value for display purposes. Sensitive values are
+// replaced with "********".
+func (r *ConfigRegistry) RedactedValue(key string) string {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	if r.sensitive[key] {
+		return "********"
+	}
+	return r.values[key]
+}
+
+// ExpandConfigTemplate replaces all {{config "key"}} references in a string
+// with their resolved values from the registry. Unresolved keys are left as-is.
+func (r *ConfigRegistry) ExpandConfigTemplate(s string) string {
+	return configKeyRegexp.ReplaceAllStringFunc(s, func(match string) string {
+		sub := configKeyRegexp.FindStringSubmatch(match)
+		if len(sub) < 2 {
+			return match
+		}
+		if v, ok := r.Get(sub[1]); ok {
+			return v
+		}
+		return match
+	})
+}
+
+// ParseSchema parses a schema definition from a config map.
+func ParseSchema(raw map[string]any) (map[string]SchemaEntry, error) {
+	schema := make(map[string]SchemaEntry)
+	for key, val := range raw {
+		entryMap, ok := val.(map[string]any)
+		if !ok {
+			return nil, fmt.Errorf("schema entry %q must be a map", key)
+		}
+		entry := SchemaEntry{}
+		if v, ok := entryMap["env"].(string); ok {
+			entry.Env = v
+		}
+		if v, ok := entryMap["required"].(bool); ok {
+			entry.Required = v
+		}
+		if v, ok := entryMap["default"].(string); ok {
+			entry.Default = v
+		}
+		if v, ok := entryMap["sensitive"].(bool); ok {
+			entry.Sensitive = v
+		}
+		if v, ok := entryMap["desc"].(string); ok {
+			entry.Desc = v
+		}
+		schema[key] = entry
+	}
+	return schema, nil
+}
+
+// LoadConfigSources loads configuration values into the registry from the
+// declared sources in order. Later sources override earlier ones.
+// Supported source types: "defaults" (from schema defaults) and "env" (from
+// environment variables, with optional prefix).
+func LoadConfigSources(registry *ConfigRegistry, sources []map[string]any, schemaEntries map[string]SchemaEntry) error {
+	for _, src := range sources {
+		srcType, _ := src["type"].(string)
+		switch srcType {
+		case "defaults":
+			for key, entry := range schemaEntries {
+				if entry.Default != "" {
+					if err := registry.Set(key, entry.Default, entry.Sensitive); err != nil {
+						return err
+					}
+				}
+			}
+		case "env":
+			prefix, _ := src["prefix"].(string)
+			for key, entry := range schemaEntries {
+				envKey := entry.Env
+				if envKey == "" {
+					continue
+				}
+				if prefix != "" {
+					envKey = prefix + envKey
+				}
+				if val, ok := os.LookupEnv(envKey); ok {
+					if err := registry.Set(key, val, entry.Sensitive); err != nil {
+						return err
+					}
+				}
+			}
+		default:
+			return fmt.Errorf("unsupported config source type: %q", srcType)
+		}
+	}
+	return nil
+}
+
+// ValidateRequired checks that all required schema keys have values in the
+// registry. Returns an error listing all missing keys.
+func ValidateRequired(registry *ConfigRegistry, schemaEntries map[string]SchemaEntry) error {
+	var missing []string
+	for key, entry := range schemaEntries {
+		if entry.Required {
+			if _, ok := registry.Get(key); !ok {
+				missing = append(missing, key)
+			}
+		}
+	}
+	if len(missing) > 0 {
+		sort.Strings(missing)
+		return fmt.Errorf("missing required config keys: %s", strings.Join(missing, ", "))
+	}
+	return nil
+}
+
+// ExpandConfigRefsMap recursively walks a config map and expands all
+// {{config "key"}} references in string values using the given registry.
+func ExpandConfigRefsMap(registry *ConfigRegistry, cfg map[string]any) {
+	if registry == nil || cfg == nil {
+		return
+	}
+	for k, v := range cfg {
+		switch val := v.(type) {
+		case string:
+			cfg[k] = registry.ExpandConfigTemplate(val)
+		case map[string]any:
+			ExpandConfigRefsMap(registry, val)
+		case []any:
+			expandConfigRefsSlice(registry, val)
+		}
+	}
+}
+
+// expandConfigRefsSlice recursively walks a slice and expands all
+// {{config "key"}} references in string values.
+func expandConfigRefsSlice(registry *ConfigRegistry, items []any) {
+	for i, item := range items {
+		switch v := item.(type) {
+		case string:
+			items[i] = registry.ExpandConfigTemplate(v)
+		case map[string]any:
+			ExpandConfigRefsMap(registry, v)
+		case []any:
+			expandConfigRefsSlice(registry, v)
+		}
+	}
+}
+
+// ConfigProviderModule implements modular.Module for the config.provider type.
+// It acts as a no-op module at runtime since all config resolution happens at
+// build time via the ConfigTransformHook. The module exists to hold the config
+// registry reference for service discovery.
+type ConfigProviderModule struct {
+	name     string
+	config   map[string]any
+	registry *ConfigRegistry
+}
+
+// NewConfigProviderModule creates a new ConfigProviderModule.
+func NewConfigProviderModule(name string, cfg map[string]any) *ConfigProviderModule {
+	return &ConfigProviderModule{
+		name:     name,
+		config:   cfg,
+		registry: globalConfigRegistry,
+	}
+}
+
+// Name returns the module name.
+func (m *ConfigProviderModule) Name() string { return m.name }
+
+// Dependencies returns an empty slice — config.provider has no dependencies.
+func (m *ConfigProviderModule) Dependencies() []string { return nil }
+
+// Init registers the config registry as a service in the application.
+func (m *ConfigProviderModule) Init(app modular.Application) error {
+	return app.RegisterService("config.registry", m.registry)
+}
+
+// Registry returns the underlying ConfigRegistry.
+func (m *ConfigProviderModule) Registry() *ConfigRegistry {
+	return m.registry
+}
diff --git a/module/config_provider_test.go b/module/config_provider_test.go
new file mode 100644
index 00000000..00f7b431
--- /dev/null
+++ b/module/config_provider_test.go
@@ -0,0 +1,497 @@
+package module
+
+import (
+	"os"
+	"sort"
+	"testing"
+)
+
+// --- ConfigRegistry tests ---
+
+func TestConfigRegistrySetAndGet(t *testing.T) {
+	r := NewConfigRegistry()
+	if err := r.Set("key1", "value1", false); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	v, ok := r.Get("key1")
+	if !ok || v != "value1" {
+		t.Fatalf("expected value1, got %q (ok=%v)", v, ok)
+	}
+}
+
+func TestConfigRegistryGetMissing(t *testing.T) {
+	r := NewConfigRegistry()
+	_, ok := r.Get("nonexistent")
+	if ok {
+		t.Fatal("expected key not found")
+	}
+}
+
+func TestConfigRegistrySensitive(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("secret", "s3cr3t", true)
+	_ = r.Set("plain", "hello", false)
+
+	if !r.IsSensitive("secret") {
+		t.Fatal("expected secret to be sensitive")
+	}
+	if r.IsSensitive("plain") {
+		t.Fatal("expected plain to not be sensitive")
+	}
+}
+
+func TestConfigRegistryRedactedValue(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("secret", "s3cr3t", true)
+	_ = r.Set("plain", "hello", false)
+
+	if r.RedactedValue("secret") != "********" {
+		t.Fatalf("expected redacted, got %q", r.RedactedValue("secret"))
+	}
+	if r.RedactedValue("plain") != "hello" {
+		t.Fatalf("expected hello, got %q", r.RedactedValue("plain"))
+	}
+}
+
+func TestConfigRegistryFreeze(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("key1", "value1", false)
+	r.Freeze()
+
+	if err := r.Set("key2", "value2", false); err == nil {
+		t.Fatal("expected error setting after freeze")
+	}
+	// Can still read
+	v, ok := r.Get("key1")
+	if !ok || v != "value1" {
+		t.Fatal("expected to read frozen value")
+	}
+}
+
+func TestConfigRegistryReset(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("key1", "value1", false)
+	r.Freeze()
+	r.Reset()
+
+	// Should be writable again
+	if err := r.Set("key2", "value2", false); err != nil {
+		t.Fatalf("unexpected error after reset: %v", err)
+	}
+	_, ok := r.Get("key1")
+	if ok {
+		t.Fatal("expected key1 to be cleared after reset")
+	}
+}
+
+func TestConfigRegistryKeys(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("b", "2", false)
+	_ = r.Set("a", "1", false)
+	_ = r.Set("c", "3", false)
+	keys := r.Keys()
+	sort.Strings(keys)
+	if len(keys) != 3 || keys[0] != "a" || keys[1] != "b" || keys[2] != "c" {
+		t.Fatalf("unexpected keys: %v", keys)
+	}
+}
+
+// --- ExpandConfigTemplate tests ---
+
+func TestExpandConfigTemplate(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("db_dsn", "postgres://localhost/db", false)
+	_ = r.Set("api_port", "8080", false)
+
+	tests := []struct {
+		input    string
+		expected string
+	}{
+		{`{{config "db_dsn"}}`, "postgres://localhost/db"},
+		{`{{ config "api_port" }}`, "8080"},
+		{`host:{{config "api_port"}}`, "host:8080"},
+		{`no-template-here`, "no-template-here"},
+		{`{{config "missing"}}`, `{{config "missing"}}`}, // unresolved stays
+		{`dsn={{config "db_dsn"}}&port={{config "api_port"}}`, "dsn=postgres://localhost/db&port=8080"},
+		// single-quoted variants
+		{`{{config 'db_dsn'}}`, "postgres://localhost/db"},
+		{`{{ config 'api_port' }}`, "8080"},
+	}
+	for _, tt := range tests {
+		result := r.ExpandConfigTemplate(tt.input)
+		if result != tt.expected {
+			t.Errorf("ExpandConfigTemplate(%q) = %q, want %q", tt.input, result, tt.expected)
+		}
+	}
+}
+
+// --- ParseSchema tests ---
+
+func TestParseSchema(t *testing.T) {
+	raw := map[string]any{
+		"db_dsn": map[string]any{
+			"env":       "DB_DSN",
+			"required":  true,
+			"sensitive": true,
+			"desc":      "Database connection string",
+		},
+		"api_port": map[string]any{
+			"env":     "API_PORT",
+			"default": "8080",
+			"desc":    "HTTP listen port",
+		},
+	}
+	schema, err := ParseSchema(raw)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(schema) != 2 {
+		t.Fatalf("expected 2 entries, got %d", len(schema))
+	}
+	db := schema["db_dsn"]
+	if db.Env != "DB_DSN" || !db.Required || !db.Sensitive || db.Desc != "Database connection string" {
+		t.Fatalf("unexpected db_dsn entry: %+v", db)
+	}
+	port := schema["api_port"]
+	if port.Env != "API_PORT" || port.Default != "8080" || port.Required {
+		t.Fatalf("unexpected api_port entry: %+v", port)
+	}
+}
+
+func TestParseSchemaInvalid(t *testing.T) {
+	raw := map[string]any{
+		"bad_entry": "not-a-map",
+	}
+	_, err := ParseSchema(raw)
+	if err == nil {
+		t.Fatal("expected error for non-map entry")
+	}
+}
+
+// --- LoadConfigSources tests ---
+
+func TestLoadConfigSourcesDefaults(t *testing.T) {
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"port":   {Default: "8080"},
+		"region": {Default: "us-east-1"},
+		"secret": {Default: "default-secret", Sensitive: true},
+	}
+	sources := []map[string]any{
+		{"type": "defaults"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	v, ok := r.Get("port")
+	if !ok || v != "8080" {
+		t.Fatalf("expected 8080, got %q", v)
+	}
+	if !r.IsSensitive("secret") {
+		t.Fatal("expected secret to be sensitive")
+	}
+}
+
+func TestLoadConfigSourcesEnv(t *testing.T) {
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"port": {Env: "TEST_CFG_PORT", Default: "8080"},
+	}
+
+	t.Setenv("TEST_CFG_PORT", "9090")
+
+	sources := []map[string]any{
+		{"type": "defaults"},
+		{"type": "env"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	v, _ := r.Get("port")
+	if v != "9090" {
+		t.Fatalf("expected env override 9090, got %q", v)
+	}
+}
+
+func TestLoadConfigSourcesEnvPrefix(t *testing.T) {
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"port": {Env: "PORT"},
+	}
+
+	t.Setenv("APP_PORT", "3000")
+
+	sources := []map[string]any{
+		{"type": "env", "prefix": "APP_"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	v, _ := r.Get("port")
+	if v != "3000" {
+		t.Fatalf("expected 3000, got %q", v)
+	}
+}
+
+func TestLoadConfigSourcesUnknownType(t *testing.T) {
+	r := NewConfigRegistry()
+	sources := []map[string]any{
+		{"type": "unknown"},
+	}
+	err := LoadConfigSources(r, sources, nil)
+	if err == nil {
+		t.Fatal("expected error for unknown source type")
+	}
+}
+
+// --- ValidateRequired tests ---
+
+func TestValidateRequiredAllPresent(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("db_dsn", "postgres://...", false)
+	_ = r.Set("token", "abc", false)
+	schema := map[string]SchemaEntry{
+		"db_dsn": {Required: true},
+		"token":  {Required: true},
+		"port":   {Required: false},
+	}
+	if err := ValidateRequired(r, schema); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestValidateRequiredMissing(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("token", "abc", false)
+	schema := map[string]SchemaEntry{
+		"db_dsn": {Required: true},
+		"token":  {Required: true},
+	}
+	err := ValidateRequired(r, schema)
+	if err == nil {
+		t.Fatal("expected error for missing required key")
+	}
+	if got := err.Error(); got != "missing required config keys: db_dsn" {
+		t.Fatalf("unexpected error: %q", got)
+	}
+}
+
+func TestValidateRequiredMissingSorted(t *testing.T) {
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"z_key": {Required: true},
+		"a_key": {Required: true},
+		"m_key": {Required: true},
+	}
+	err := ValidateRequired(r, schema)
+	if err == nil {
+		t.Fatal("expected error for missing required keys")
+	}
+	// Keys must appear in sorted order regardless of map iteration order
+	const want = "missing required config keys: a_key, m_key, z_key"
+	if got := err.Error(); got != want {
+		t.Fatalf("expected sorted error:\nwant: %q\ngot:  %q", want, got)
+	}
+}
+
+// --- ExpandConfigRefsMap tests ---
+
+func TestExpandConfigRefsMap(t *testing.T) {
+	r := NewConfigRegistry()
+	_ = r.Set("db_dsn", "postgres://localhost/db", false)
+	_ = r.Set("port", "8080", false)
+
+	cfg := map[string]any{
+		"dsn":     `{{config "db_dsn"}}`,
+		"address": `0.0.0.0:{{config "port"}}`,
+		"nested": map[string]any{
+			"value": `{{config "db_dsn"}}`,
+		},
+		"list": []any{
+			`item-{{config "port"}}`,
+			map[string]any{
+				"inner": `{{config "db_dsn"}}`,
+			},
+		},
+		"unchanged": "no-template",
+	}
+
+	ExpandConfigRefsMap(r, cfg)
+
+	if cfg["dsn"] != "postgres://localhost/db" {
+		t.Fatalf("dsn: expected expanded, got %q", cfg["dsn"])
+	}
+	if cfg["address"] != "0.0.0.0:8080" {
+		t.Fatalf("address: expected expanded, got %q", cfg["address"])
+	}
+	nested := cfg["nested"].(map[string]any)
+	if nested["value"] != "postgres://localhost/db" {
+		t.Fatalf("nested.value: expected expanded, got %q", nested["value"])
+	}
+	list := cfg["list"].([]any)
+	if list[0] != "item-8080" {
+		t.Fatalf("list[0]: expected expanded, got %q", list[0])
+	}
+	innerMap := list[1].(map[string]any)
+	if innerMap["inner"] != "postgres://localhost/db" {
+		t.Fatalf("list[1].inner: expected expanded, got %q", innerMap["inner"])
+	}
+	if cfg["unchanged"] != "no-template" {
+		t.Fatalf("unchanged: expected no change, got %q", cfg["unchanged"])
+	}
+}
+
+func TestExpandConfigRefsMapNilSafe(t *testing.T) {
+	// Should not panic
+	ExpandConfigRefsMap(nil, nil)
+	ExpandConfigRefsMap(NewConfigRegistry(), nil)
+	ExpandConfigRefsMap(nil, map[string]any{"a": "b"})
+}
+
+// --- ConfigProviderModule tests ---
+
+func TestConfigProviderModuleName(t *testing.T) {
+	m := NewConfigProviderModule("my-config", nil)
+	if m.Name() != "my-config" {
+		t.Fatalf("expected my-config, got %q", m.Name())
+	}
+}
+
+func TestConfigProviderModuleDependencies(t *testing.T) {
+	m := NewConfigProviderModule("my-config", nil)
+	if deps := m.Dependencies(); deps != nil {
+		t.Fatalf("expected nil dependencies, got %v", deps)
+	}
+}
+
+func TestConfigProviderModuleRegistry(t *testing.T) {
+	m := NewConfigProviderModule("my-config", nil)
+	if m.Registry() == nil {
+		t.Fatal("expected non-nil registry")
+	}
+}
+
+// --- Integration: end-to-end schema + sources + expand ---
+
+func TestEndToEndConfigProvider(t *testing.T) {
+	// Simulate what the ConfigTransformHook does
+
+	t.Setenv("E2E_DB_DSN", "postgres://prod/mydb")
+	// Don't set E2E_API_PORT — should use default
+
+	r := NewConfigRegistry()
+
+	schemaRaw := map[string]any{
+		"db_dsn": map[string]any{
+			"env":       "E2E_DB_DSN",
+			"required":  true,
+			"sensitive": true,
+			"desc":      "Database connection string",
+		},
+		"api_port": map[string]any{
+			"env":     "E2E_API_PORT",
+			"default": "8080",
+			"desc":    "HTTP listen port",
+		},
+		"region": map[string]any{
+			"env":     "E2E_REGION",
+			"default": "us-east-1",
+		},
+	}
+
+	schema, err := ParseSchema(schemaRaw)
+	if err != nil {
+		t.Fatalf("ParseSchema: %v", err)
+	}
+
+	sources := []map[string]any{
+		{"type": "defaults"},
+		{"type": "env"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("LoadConfigSources: %v", err)
+	}
+	if err := ValidateRequired(r, schema); err != nil {
+		t.Fatalf("ValidateRequired: %v", err)
+	}
+	r.Freeze()
+
+	// Expand references in module config
+	moduleCfg := map[string]any{
+		"driver":  "postgres",
+		"dsn":     `{{config "db_dsn"}}`,
+		"address": `0.0.0.0:{{config "api_port"}}`,
+		"region":  `{{config "region"}}`,
+	}
+	ExpandConfigRefsMap(r, moduleCfg)
+
+	if moduleCfg["dsn"] != "postgres://prod/mydb" {
+		t.Fatalf("dsn: expected env value, got %q", moduleCfg["dsn"])
+	}
+	if moduleCfg["address"] != "0.0.0.0:8080" {
+		t.Fatalf("address: expected default, got %q", moduleCfg["address"])
+	}
+	if moduleCfg["region"] != "us-east-1" {
+		t.Fatalf("region: expected default, got %q", moduleCfg["region"])
+	}
+
+	// Verify sensitive redaction
+	if r.RedactedValue("db_dsn") != "********" {
+		t.Fatal("expected db_dsn to be redacted")
+	}
+	if r.RedactedValue("api_port") != "8080" {
+		t.Fatal("expected api_port to not be redacted")
+	}
+}
+
+func TestEndToEndRequiredMissing(t *testing.T) {
+	// Unset env to trigger missing required
+	os.Unsetenv("E2E_MISSING_KEY")
+
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"missing_key": {Env: "E2E_MISSING_KEY", Required: true},
+	}
+	sources := []map[string]any{
+		{"type": "defaults"},
+		{"type": "env"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("LoadConfigSources: %v", err)
+	}
+	err := ValidateRequired(r, schema)
+	if err == nil {
+		t.Fatal("expected validation error for missing required key")
+	}
+}
+
+func TestEnvOverridesDefault(t *testing.T) {
+	r := NewConfigRegistry()
+	schema := map[string]SchemaEntry{
+		"port": {Env: "TEST_OVERRIDE_PORT", Default: "8080"},
+	}
+	t.Setenv("TEST_OVERRIDE_PORT", "9999")
+
+	sources := []map[string]any{
+		{"type": "defaults"},
+		{"type": "env"},
+	}
+	if err := LoadConfigSources(r, sources, schema); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	v, _ := r.Get("port")
+	if v != "9999" {
+		t.Fatalf("expected env override 9999, got %q", v)
+	}
+}
+
+// --- Global registry tests ---
+
+func TestGlobalConfigRegistry(t *testing.T) {
+	reg := GetConfigRegistry()
+	if reg == nil {
+		t.Fatal("expected non-nil global registry")
+	}
+	// Reset to clean state for other tests
+	reg.Reset()
+}
diff --git a/module/pipeline_template.go b/module/pipeline_template.go
index 51870a61..a8110555 100644
--- a/module/pipeline_template.go
+++ b/module/pipeline_template.go
@@ -354,5 +354,13 @@ func templateFuncMap() template.FuncMap {
 			}
 			return string(b)
 		},
+		// config looks up a value from the global config registry (populated by
+		// a config.provider module). Returns an empty string if the key is not found.
+		"config": func(key string) string {
+			if v, ok := GetConfigRegistry().Get(key); ok {
+				return v
+			}
+			return ""
+		},
 	}
 }
diff --git a/plugins/all/all.go b/plugins/all/all.go
index 4ca17ce0..ef5bbe76 100644
--- a/plugins/all/all.go
+++ b/plugins/all/all.go
@@ -27,6 +27,7 @@ import (
 	pluginauth "github.com/GoCodeAlone/workflow/plugins/auth"
 	plugincicd "github.com/GoCodeAlone/workflow/plugins/cicd"
 	plugincloud "github.com/GoCodeAlone/workflow/plugins/cloud"
+	pluginconfigprovider "github.com/GoCodeAlone/workflow/plugins/configprovider"
 	plugindatastores "github.com/GoCodeAlone/workflow/plugins/datastores"
 	plugindlq "github.com/GoCodeAlone/workflow/plugins/dlq"
 	pluginevstore "github.com/GoCodeAlone/workflow/plugins/eventstore"
@@ -61,6 +62,7 @@ type PluginLoader interface {
 func DefaultPlugins() []plugin.EnginePlugin {
 	return []plugin.EnginePlugin{
 		pluginlicense.New(),
+		pluginconfigprovider.New(),
 		pluginhttp.New(),
 		pluginobs.New(),
 		pluginmessaging.New(),
diff --git a/plugins/configprovider/plugin.go b/plugins/configprovider/plugin.go
new file mode 100644
index 00000000..3136c7dc
--- /dev/null
+++ b/plugins/configprovider/plugin.go
@@ -0,0 +1,172 @@
+// Package configprovider registers the config.provider module type and its
+// ConfigTransformHook. The hook runs before module registration to parse the
+// config.provider schema, load values from declared sources, validate required
+// keys, and expand {{config "key"}} references throughout the rest of the
+// configuration.
+package configprovider
+
+import (
+	"fmt"
+
+	"github.com/CrisisTextLine/modular"
+	"github.com/GoCodeAlone/workflow/config"
+	"github.com/GoCodeAlone/workflow/module"
+	"github.com/GoCodeAlone/workflow/plugin"
+)
+
+// Plugin registers the config.provider module factory and a ConfigTransformHook
+// that resolves {{config "key"}} references at config load time.
+type Plugin struct {
+	plugin.BaseEnginePlugin
+}
+
+// New creates a new config provider plugin.
+func New() *Plugin {
+	return &Plugin{
+		BaseEnginePlugin: plugin.BaseEnginePlugin{
+			BaseNativePlugin: plugin.BaseNativePlugin{
+				PluginName:        "configprovider",
+				PluginVersion:     "1.0.0",
+				PluginDescription: "Application configuration registry with schema validation, defaults, and source layering",
+			},
+			Manifest: plugin.PluginManifest{
+				Name:        "configprovider",
+				Version:     "1.0.0",
+				Author:      "GoCodeAlone",
+				Description: "Application configuration registry with schema validation, defaults, and source layering",
+				Tier:        plugin.TierCore,
+				ModuleTypes: []string{"config.provider"},
+			},
+		},
+	}
+}
+
+// ModuleFactories returns the config.provider module factory.
+func (p *Plugin) ModuleFactories() map[string]plugin.ModuleFactory {
+	return map[string]plugin.ModuleFactory{
+		"config.provider": func(name string, cfg map[string]any) modular.Module {
+			return module.NewConfigProviderModule(name, cfg)
+		},
+	}
+}
+
+// ConfigTransformHooks returns a high-priority hook that processes config.provider
+// modules before any other modules are registered. It:
+//  1. Finds config.provider modules in the config
+//  2. Parses their schema definitions
+//  3. Loads values from declared sources (defaults, env)
+//  4. Validates all required keys are present
+//  5. Expands {{config "key"}} references in all other module, workflow, trigger, and pipeline configs
+func (p *Plugin) ConfigTransformHooks() []plugin.ConfigTransformHook {
+	return []plugin.ConfigTransformHook{
+		{
+			Name:     "config-provider-expansion",
+			Priority: 1000, // Run before other transform hooks
+			Hook:     configTransformHook,
+		},
+	}
+}
+
+// configTransformHook processes all config.provider modules in the configuration.
+func configTransformHook(cfg *config.WorkflowConfig) error {
+	registry := module.GetConfigRegistry()
+	registry.Reset()
+
+	found := false
+	for _, modCfg := range cfg.Modules {
+		if modCfg.Type != "config.provider" {
+			continue
+		}
+		found = true
+
+		if err := processConfigProvider(registry, modCfg.Config); err != nil {
+			return fmt.Errorf("config.provider module %q: %w", modCfg.Name, err)
+		}
+	}
+
+	if !found {
+		return nil // No config.provider modules — nothing to do
+	}
+
+	registry.Freeze()
+
+	// Expand {{config "key"}} in all module configs (except config.provider itself)
+	for i := range cfg.Modules {
+		if cfg.Modules[i].Type == "config.provider" {
+			continue
+		}
+		module.ExpandConfigRefsMap(registry, cfg.Modules[i].Config)
+	}
+
+	// Expand in workflow configs
+	for key, wf := range cfg.Workflows {
+		if m, ok := wf.(map[string]any); ok {
+			module.ExpandConfigRefsMap(registry, m)
+			cfg.Workflows[key] = m
+		}
+	}
+
+	// Expand in trigger configs
+	for key, tr := range cfg.Triggers {
+		if m, ok := tr.(map[string]any); ok {
+			module.ExpandConfigRefsMap(registry, m)
+			cfg.Triggers[key] = m
+		}
+	}
+
+	// Expand in pipeline configs
+	for key, pl := range cfg.Pipelines {
+		if m, ok := pl.(map[string]any); ok {
+			module.ExpandConfigRefsMap(registry, m)
+			cfg.Pipelines[key] = m
+		}
+	}
+
+	// Expand in platform configs
+	module.ExpandConfigRefsMap(registry, cfg.Platform)
+
+	return nil
+}
+
+// processConfigProvider parses a single config.provider module's config,
+// loads sources, and validates required keys.
+func processConfigProvider(registry *module.ConfigRegistry, cfg map[string]any) error {
+	// Parse schema
+	schemaRaw, ok := cfg["schema"].(map[string]any)
+	if !ok {
+		return fmt.Errorf("config.provider requires a 'schema' section")
+	}
+	schemaEntries, err := module.ParseSchema(schemaRaw)
+	if err != nil {
+		return fmt.Errorf("invalid schema: %w", err)
+	}
+
+	// Parse and load sources
+	sourcesRaw, ok := cfg["sources"]
+	if !ok {
+		return fmt.Errorf("config.provider requires a 'sources' section")
+	}
+	sourcesSlice, ok := sourcesRaw.([]any)
+	if !ok {
+		return fmt.Errorf("'sources' must be a list")
+	}
+	sources := make([]map[string]any, 0, len(sourcesSlice))
+	for _, s := range sourcesSlice {
+		sm, ok := s.(map[string]any)
+		if !ok {
+			return fmt.Errorf("each source must be a map")
+		}
+		sources = append(sources, sm)
+	}
+
+	if err := module.LoadConfigSources(registry, sources, schemaEntries); err != nil {
+		return fmt.Errorf("loading config sources: %w", err)
+	}
+
+	// Validate required keys
+	if err := module.ValidateRequired(registry, schemaEntries); err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/plugins/configprovider/plugin_test.go b/plugins/configprovider/plugin_test.go
new file mode 100644
index 00000000..bdc544b9
--- /dev/null
+++ b/plugins/configprovider/plugin_test.go
@@ -0,0 +1,265 @@
+package configprovider
+
+import (
+	"testing"
+
+	"github.com/GoCodeAlone/workflow/config"
+	"github.com/GoCodeAlone/workflow/module"
+)
+
+func TestPluginMetadata(t *testing.T) {
+	p := New()
+	if p.Name() != "configprovider" {
+		t.Fatalf("expected name 'configprovider', got %q", p.Name())
+	}
+	if p.Version() != "1.0.0" {
+		t.Fatalf("expected version '1.0.0', got %q", p.Version())
+	}
+	manifest := p.EngineManifest()
+	if len(manifest.ModuleTypes) != 1 || manifest.ModuleTypes[0] != "config.provider" {
+		t.Fatalf("unexpected module types: %v", manifest.ModuleTypes)
+	}
+}
+
+func TestPluginModuleFactories(t *testing.T) {
+	p := New()
+	factories := p.ModuleFactories()
+	if _, ok := factories["config.provider"]; !ok {
+		t.Fatal("expected config.provider factory")
+	}
+	mod := factories["config.provider"]("test-config", map[string]any{})
+	if mod == nil {
+		t.Fatal("expected non-nil module")
+	}
+	if mod.Name() != "test-config" {
+		t.Fatalf("expected module name 'test-config', got %q", mod.Name())
+	}
+}
+
+func TestPluginConfigTransformHooks(t *testing.T) {
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	if len(hooks) != 1 {
+		t.Fatalf("expected 1 hook, got %d", len(hooks))
+	}
+	if hooks[0].Name != "config-provider-expansion" {
+		t.Fatalf("unexpected hook name: %q", hooks[0].Name)
+	}
+	if hooks[0].Priority != 1000 {
+		t.Fatalf("expected priority 1000, got %d", hooks[0].Priority)
+	}
+}
+
+func TestConfigTransformHookNoProvider(t *testing.T) {
+	// When there's no config.provider module, hook is a no-op
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{Name: "server", Type: "http.server", Config: map[string]any{"port": "8080"}},
+		},
+	}
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	if err := hooks[0].Hook(cfg); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestConfigTransformHookBasic(t *testing.T) {
+	module.GetConfigRegistry().Reset()
+
+	t.Setenv("HOOK_TEST_DB_DSN", "postgres://test/db")
+
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{
+				Name: "app-config",
+				Type: "config.provider",
+				Config: map[string]any{
+					"sources": []any{
+						map[string]any{"type": "defaults"},
+						map[string]any{"type": "env"},
+					},
+					"schema": map[string]any{
+						"db_dsn": map[string]any{
+							"env":       "HOOK_TEST_DB_DSN",
+							"required":  true,
+							"sensitive": true,
+						},
+						"port": map[string]any{
+							"env":     "HOOK_TEST_PORT",
+							"default": "8080",
+						},
+					},
+				},
+			},
+			{
+				Name: "db",
+				Type: "database.workflow",
+				Config: map[string]any{
+					"driver":  "postgres",
+					"dsn":     `{{config "db_dsn"}}`,
+					"address": `0.0.0.0:{{config "port"}}`,
+				},
+			},
+		},
+	}
+
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	if err := hooks[0].Hook(cfg); err != nil {
+		t.Fatalf("hook error: %v", err)
+	}
+
+	// Verify the db module's config was expanded
+	dbCfg := cfg.Modules[1].Config
+	if dbCfg["dsn"] != "postgres://test/db" {
+		t.Fatalf("dsn not expanded: %q", dbCfg["dsn"])
+	}
+	if dbCfg["address"] != "0.0.0.0:8080" {
+		t.Fatalf("address not expanded: %q", dbCfg["address"])
+	}
+	if dbCfg["driver"] != "postgres" {
+		t.Fatalf("driver changed unexpectedly: %q", dbCfg["driver"])
+	}
+}
+
+func TestConfigTransformHookMissingRequired(t *testing.T) {
+	module.GetConfigRegistry().Reset()
+
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{
+				Name: "app-config",
+				Type: "config.provider",
+				Config: map[string]any{
+					"sources": []any{
+						map[string]any{"type": "defaults"},
+						map[string]any{"type": "env"},
+					},
+					"schema": map[string]any{
+						"required_key": map[string]any{
+							"env":      "NEVER_SET_THIS_KEY_12345",
+							"required": true,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	err := hooks[0].Hook(cfg)
+	if err == nil {
+		t.Fatal("expected error for missing required key")
+	}
+}
+
+func TestConfigTransformHookWorkflowExpansion(t *testing.T) {
+	module.GetConfigRegistry().Reset()
+
+	t.Setenv("HOOK_WF_REGION", "eu-west-1")
+
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{
+				Name: "app-config",
+				Type: "config.provider",
+				Config: map[string]any{
+					"sources": []any{
+						map[string]any{"type": "defaults"},
+						map[string]any{"type": "env"},
+					},
+					"schema": map[string]any{
+						"region": map[string]any{
+							"env":     "HOOK_WF_REGION",
+							"default": "us-east-1",
+						},
+					},
+				},
+			},
+		},
+		Workflows: map[string]any{
+			"http": map[string]any{
+				"region": `{{config "region"}}`,
+			},
+		},
+		Triggers: map[string]any{
+			"main": map[string]any{
+				"region": `{{config "region"}}`,
+			},
+		},
+		Pipelines: map[string]any{
+			"pipeline1": map[string]any{
+				"region": `{{config "region"}}`,
+			},
+		},
+	}
+
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	if err := hooks[0].Hook(cfg); err != nil {
+		t.Fatalf("hook error: %v", err)
+	}
+
+	wf := cfg.Workflows["http"].(map[string]any)
+	if wf["region"] != "eu-west-1" {
+		t.Fatalf("workflow region not expanded: %q", wf["region"])
+	}
+	tr := cfg.Triggers["main"].(map[string]any)
+	if tr["region"] != "eu-west-1" {
+		t.Fatalf("trigger region not expanded: %q", tr["region"])
+	}
+	pl := cfg.Pipelines["pipeline1"].(map[string]any)
+	if pl["region"] != "eu-west-1" {
+		t.Fatalf("pipeline region not expanded: %q", pl["region"])
+	}
+}
+
+func TestConfigTransformHookMissingSchema(t *testing.T) {
+	module.GetConfigRegistry().Reset()
+
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{
+				Name: "app-config",
+				Type: "config.provider",
+				Config: map[string]any{
+					"sources": []any{map[string]any{"type": "defaults"}},
+				},
+			},
+		},
+	}
+
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	err := hooks[0].Hook(cfg)
+	if err == nil {
+		t.Fatal("expected error for missing schema")
+	}
+}
+
+func TestConfigTransformHookMissingSources(t *testing.T) {
+	module.GetConfigRegistry().Reset()
+
+	cfg := &config.WorkflowConfig{
+		Modules: []config.ModuleConfig{
+			{
+				Name: "app-config",
+				Type: "config.provider",
+				Config: map[string]any{
+					"schema": map[string]any{
+						"key": map[string]any{"default": "val"},
+					},
+				},
+			},
+		},
+	}
+
+	p := New()
+	hooks := p.ConfigTransformHooks()
+	err := hooks[0].Hook(cfg)
+	if err == nil {
+		t.Fatal("expected error for missing sources")
+	}
+}
diff --git a/schema/module_schema.go b/schema/module_schema.go
index a0f3939a..5fc9b715 100644
--- a/schema/module_schema.go
+++ b/schema/module_schema.go
@@ -472,6 +472,18 @@ func (r *ModuleSchemaRegistry) registerBuiltins() {
 		ConfigFields: []ConfigFieldDef{},
 	})
 
+	r.Register(&ModuleSchema{
+		Type:        "config.provider",
+		Label:       "Config Provider",
+		Category:    "infrastructure",
+		Description: "Application configuration registry with schema validation, defaults, and source layering. Provides {{config \"key\"}} template references.",
+		ConfigFields: []ConfigFieldDef{
+			{Key: "sources", Label: "Sources", Type: FieldTypeArray, Required: true, Description: "Ordered list of config sources (later overrides earlier). Supported types: defaults, env"},
+			{Key: "schema", Label: "Schema", Type: FieldTypeMap, Required: true, Description: "Map of config key definitions with env, required, default, sensitive, desc fields"},
+		},
+		MaxIncoming: intPtr(0),
+	})
+
 	r.Register(&ModuleSchema{
 		Type:         "jsonschema.modular",
 		Label:        "JSON Schema Validator",
diff --git a/schema/schema.go b/schema/schema.go
index 4f8ea0f6..80a25830 100644
--- a/schema/schema.go
+++ b/schema/schema.go
@@ -150,6 +150,7 @@ var coreModuleTypes = []string{
 	"auth.jwt",
 	"auth.user-store",
 	"cache.modular",
+	"config.provider",
 	"data.transformer",
 	"database.workflow",
 	"dlq.service",