diff --git a/go-sdk/bundle/bundlev1/bundlev1server/server.go b/go-sdk/bundle/bundlev1/bundlev1server/server.go
index 2b39392db9c2c..010267b493a33 100644
--- a/go-sdk/bundle/bundlev1/bundlev1server/server.go
+++ b/go-sdk/bundle/bundlev1/bundlev1server/server.go
@@ -22,7 +22,6 @@ import (
 	"log/slog"
 	"os"
 
-	"github.com/evanphx/go-hclog-slog/hclogslog"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
 	flag "github.com/spf13/pflag"
@@ -32,6 +31,7 @@ import (
 	"github.com/apache/airflow/go-sdk/pkg/bundles/shared"
 	"github.com/apache/airflow/go-sdk/pkg/config"
 	"github.com/apache/airflow/go-sdk/pkg/execution"
+	"github.com/apache/airflow/go-sdk/pkg/logging/hclogslog"
 )
 
 // ErrCoordinatorFlagsIncomplete is returned by [Serve] when exactly one of
diff --git a/go-sdk/go.mod b/go-sdk/go.mod
index bd3942939a1c1..bf0a6d7faf8af 100644
--- a/go-sdk/go.mod
+++ b/go-sdk/go.mod
@@ -50,7 +50,6 @@ require (
 
 require (
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
-	github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/google/uuid v1.6.0
 	github.com/jarcoal/httpmock v1.4.0
diff --git a/go-sdk/go.sum b/go-sdk/go.sum
index 38512f16109ad..2b734d0f3d689 100644
--- a/go-sdk/go.sum
+++ b/go-sdk/go.sum
@@ -13,8 +13,6 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6N
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5 h1:Im4NdCnqw9SyBuU8dmXmvTPNukNah3++CTE2X6geTdw=
-github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5/go.mod h1:30+1dTR5EdDQGmcjkgOj4i6iVD3wnI0XymSOTPMlUeA=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
 github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
diff --git a/go-sdk/pkg/logging/hclogslog/hclogslog.go b/go-sdk/pkg/logging/hclogslog/hclogslog.go
new file mode 100644
index 0000000000000..b696d1d030ed5
--- /dev/null
+++ b/go-sdk/pkg/logging/hclogslog/hclogslog.go
@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package hclogslog adapts a log/slog Logger onto an hclog.Logger: it provides
+// an slog.Handler that forwards records to an underlying hclog.Logger. It is the
+// inverse of pkg/logging/shclog (which presents an slog Logger as an hclog).
+//
+// This lets code that logs through slog emit records in hclog's format, which is
+// what the go-plugin host expects to parse from a plugin's stderr.
+package hclogslog
+
+import (
+	"context"
+	"log/slog"
+	"strconv"
+
+	"github.com/hashicorp/go-hclog"
+)
+
+// Handler is an slog.Handler that forwards records to an hclog.Logger.
+type Handler struct {
+	l      hclog.Logger
+	prefix string
+	// basePos is how many positional slots the bound (WithAttrs) attributes at
+	// the current group scope already consumed. Record attributes continue from
+	// it so an empty-keyed bound attr and an empty-keyed record attr never
+	// synthesize the same numeric key (which hclog would emit as a duplicate JSON
+	// key, silently dropping one value). Reset to zero by WithGroup, since a new
+	// group prefix already namespaces its keys.
+	basePos int
+}
+
+// Adapt returns an slog.Handler that emits records through the given hclog.Logger.
+func Adapt(l hclog.Logger) slog.Handler {
+	return &Handler{l: l}
+}
+
+var _ slog.Handler = (*Handler)(nil)
+
+// Enabled reports whether a record at the given level would be logged by the
+// underlying hclog.Logger.
+func (h *Handler) Enabled(_ context.Context, level slog.Level) bool {
+	switch {
+	case level < slog.LevelDebug:
+		return h.l.IsTrace()
+	case level < slog.LevelInfo:
+		return h.l.IsDebug()
+	case level < slog.LevelWarn:
+		return h.l.IsInfo()
+	case level < slog.LevelError:
+		return h.l.IsWarn()
+	default:
+		return h.l.IsError()
+	}
+}
+
+// Handle forwards the record's message, level and attributes to the hclog.Logger.
+func (h *Handler) Handle(_ context.Context, rec slog.Record) error {
+	args := make([]any, 0, rec.NumAttrs()*2)
+	position := h.basePos
+	rec.Attrs(func(a slog.Attr) bool {
+		args = append(args, h.flatten(h.prefix, position, a)...)
+		position++
+		return true
+	})
+	h.l.Log(translateLevel(rec.Level), rec.Message, args...)
+	return nil
+}
+
+// WithAttrs returns a new handler whose records also carry the given attributes.
+func (h *Handler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	args := make([]any, 0, len(attrs)*2)
+	for i, a := range attrs {
+		args = append(args, h.flatten(h.prefix, h.basePos+i, a)...)
+	}
+	return &Handler{l: h.l.With(args...), prefix: h.prefix, basePos: h.basePos + len(attrs)}
+}
+
+// WithGroup returns a new handler that qualifies subsequent attribute keys with
+// the group name.
+func (h *Handler) WithGroup(name string) slog.Handler {
+	if name == "" {
+		return h
+	}
+	return &Handler{l: h.l, prefix: h.prefix + name + "."}
+}
+
+// flatten converts a single slog.Attr into a flat list of hclog key/value pairs.
+// Groups are expanded recursively, their keys joined with a dot; an unnamed group
+// is inlined; an empty attribute is dropped.
+func (h *Handler) flatten(prefix string, position int, a slog.Attr) []any {
+	value := a.Value.Resolve()
+	if a.Equal(slog.Attr{}) {
+		return nil
+	}
+	if value.Kind() != slog.KindGroup {
+		key := a.Key
+		if key == "" {
+			key = strconv.Itoa(position)
+		}
+		return []any{prefix + key, value.Any()}
+	}
+
+	group := value.Group()
+	if len(group) == 0 {
+		return nil
+	}
+	childPrefix := prefix
+	if a.Key != "" {
+		childPrefix = prefix + a.Key + "."
+	}
+	args := make([]any, 0, len(group)*2)
+	for position, sub := range group {
+		args = append(args, h.flatten(childPrefix, position, sub)...)
+	}
+	return args
+}
+
+func translateLevel(level slog.Level) hclog.Level {
+	switch {
+	case level < slog.LevelDebug:
+		return hclog.Trace
+	case level < slog.LevelInfo:
+		return hclog.Debug
+	case level < slog.LevelWarn:
+		return hclog.Info
+	case level < slog.LevelError:
+		return hclog.Warn
+	default:
+		return hclog.Error
+	}
+}
diff --git a/go-sdk/pkg/logging/hclogslog/hclogslog_test.go b/go-sdk/pkg/logging/hclogslog/hclogslog_test.go
new file mode 100644
index 0000000000000..f6c6e74658123
--- /dev/null
+++ b/go-sdk/pkg/logging/hclogslog/hclogslog_test.go
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package hclogslog_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/apache/airflow/go-sdk/pkg/logging"
+	"github.com/apache/airflow/go-sdk/pkg/logging/hclogslog"
+)
+
+// newLogger returns an slog.Logger writing hclog JSON into buf, plus a decoder
+// for the single record produced by one log call.
+func newLogger(buf *bytes.Buffer, level hclog.Level) *slog.Logger {
+	hc := hclog.New(&hclog.LoggerOptions{Level: level, Output: buf, JSONFormat: true})
+	return slog.New(hclogslog.Adapt(hc))
+}
+
+func decode(t *testing.T, buf *bytes.Buffer) map[string]any {
+	t.Helper()
+	var rec map[string]any
+	require.NoError(t, json.Unmarshal(buf.Bytes(), &rec))
+	return rec
+}
+
+func TestForwardsMessageAndAttrs(t *testing.T) {
+	var buf bytes.Buffer
+	newLogger(&buf, hclog.Trace).Info("hello", "user", "alice", "count", 3)
+
+	rec := decode(t, &buf)
+	assert.Equal(t, "hello", rec["@message"])
+	assert.Equal(t, "info", rec["@level"])
+	assert.Equal(t, "alice", rec["user"])
+	assert.EqualValues(t, 3, rec["count"])
+}
+
+func TestTranslatesLevels(t *testing.T) {
+	cases := map[string]struct {
+		log      func(l *slog.Logger)
+		expected string
+	}{
+		"trace": {func(l *slog.Logger) { l.Log(t.Context(), logging.LevelTrace, "m") }, "trace"},
+		"debug": {func(l *slog.Logger) { l.Debug("m") }, "debug"},
+		"info":  {func(l *slog.Logger) { l.Info("m") }, "info"},
+		"warn":  {func(l *slog.Logger) { l.Warn("m") }, "warn"},
+		"error": {func(l *slog.Logger) { l.Error("m") }, "error"},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			var buf bytes.Buffer
+			tc.log(newLogger(&buf, hclog.Trace))
+			assert.Equal(t, tc.expected, decode(t, &buf)["@level"])
+		})
+	}
+}
+
+func TestEnabledRespectsUnderlyingLevel(t *testing.T) {
+	var buf bytes.Buffer
+	log := newLogger(&buf, hclog.Warn)
+
+	log.Info("dropped")
+	assert.Empty(t, buf.String(), "records below the hclog level must be discarded")
+
+	log.Error("kept")
+	assert.Contains(t, buf.String(), "kept")
+}
+
+func TestFlattensGroupsWithDottedKeys(t *testing.T) {
+	var buf bytes.Buffer
+	log := newLogger(&buf, hclog.Trace).WithGroup("req").With("id", "abc")
+	log.Info("done", slog.Group("meta", "n", 1))
+
+	rec := decode(t, &buf)
+	assert.Equal(t, "abc", rec["req.id"])
+	assert.EqualValues(t, 1, rec["req.meta.n"])
+}
+
+func TestEmptyKeyAttrsUsePosition(t *testing.T) {
+	var buf bytes.Buffer
+	log := newLogger(&buf, hclog.Trace).
+		With(slog.String("named", "bound"), slog.String("", "bound-empty")).
+		WithGroup("req")
+	log.LogAttrs(t.Context(), slog.LevelInfo, "done",
+		slog.String("named", "record"),
+		slog.String("", "record-empty"),
+		slog.Group("meta",
+			slog.String("", "group-first"),
+			slog.String("named", "group-named"),
+			slog.String("", "group-last"),
+		),
+	)
+
+	rec := decode(t, &buf)
+	assert.Equal(t, "bound-empty", rec["1"])
+	assert.Equal(t, "record-empty", rec["req.1"])
+	assert.Equal(t, "group-first", rec["req.meta.0"])
+	assert.Equal(t, "group-last", rec["req.meta.2"])
+}
+
+func TestEmptyKeyBoundAndRecordDoNotCollide(t *testing.T) {
+	var buf bytes.Buffer
+	// No WithGroup between the bound and record attrs, so both share the empty
+	// prefix; record positions must continue past the bound ones rather than
+	// restart at 0, otherwise both empty-keyed attrs synthesize key "0" and hclog
+	// emits a duplicate JSON key that drops the bound value.
+	log := newLogger(&buf, hclog.Trace).With(slog.String("", "bound"))
+	log.Info("m", slog.String("", "record"))
+
+	rec := decode(t, &buf)
+	assert.Equal(t, "bound", rec["0"])
+	assert.Equal(t, "record", rec["1"])
+}
+
+func TestUnnamedGroupIsInlinedAndEmptyDropped(t *testing.T) {
+	var buf bytes.Buffer
+	log := newLogger(&buf, hclog.Trace)
+	log.Info("m",
+		slog.Group("", "inlined", "yes"),
+		slog.Group("empty"),
+		slog.Attr{},
+	)
+
+	rec := decode(t, &buf)
+	assert.Equal(t, "yes", rec["inlined"])
+	_, hasEmpty := rec["empty"]
+	assert.False(t, hasEmpty, "empty group must not be emitted")
+}
+
+func TestWithGroupEmptyNameIsNoop(t *testing.T) {
+	var buf bytes.Buffer
+	log := newLogger(&buf, hclog.Trace).WithGroup("")
+	log.Info("m", "k", "v")
+
+	assert.Equal(t, "v", decode(t, &buf)["k"])
+}
+
+// slog.Logger short-circuits WithGroup("") and prunes childless named groups
+// before they reach the handler, so drive the slog.Handler directly to cover
+// those branches.
+func TestHandlerContractEdgeCases(t *testing.T) {
+	var buf bytes.Buffer
+	hc := hclog.New(&hclog.LoggerOptions{Level: hclog.Trace, Output: &buf, JSONFormat: true})
+	h := hclogslog.Adapt(hc)
+
+	assert.Same(t, h, h.WithGroup(""), "empty group name must return the same handler")
+
+	// WithAttrs receives []slog.Attr verbatim (no slog.Logger pruning), so a
+	// childless named group here exercises flatten's empty-group branch.
+	h = h.WithAttrs([]slog.Attr{slog.Group("empty"), slog.String("bound", "b")})
+	rec := slog.NewRecord(time.Time{}, slog.LevelInfo, "m", 0)
+	rec.AddAttrs(slog.String("kept", "v"))
+	require.NoError(t, h.Handle(t.Context(), rec))
+
+	decoded := decode(t, &buf)
+	assert.Equal(t, "v", decoded["kept"])
+	assert.Equal(t, "b", decoded["bound"])
+	_, hasEmpty := decoded["empty"]
+	assert.False(t, hasEmpty, "childless named group must not be emitted")
+}
diff --git a/kubernetes-tests/lang_sdk/go_example/go.mod b/kubernetes-tests/lang_sdk/go_example/go.mod
index 5b06693630e01..a7afa9b04f988 100644
--- a/kubernetes-tests/lang_sdk/go_example/go.mod
+++ b/kubernetes-tests/lang_sdk/go_example/go.mod
@@ -6,7 +6,6 @@ require github.com/apache/airflow/go-sdk v0.0.0
 
 require (
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
-	github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5 // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/fsnotify/fsnotify v1.8.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
diff --git a/kubernetes-tests/lang_sdk/go_example/go.sum b/kubernetes-tests/lang_sdk/go_example/go.sum
index f76761578e649..8caf942fec5eb 100644
--- a/kubernetes-tests/lang_sdk/go_example/go.sum
+++ b/kubernetes-tests/lang_sdk/go_example/go.sum
@@ -13,8 +13,6 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6N
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5 h1:Im4NdCnqw9SyBuU8dmXmvTPNukNah3++CTE2X6geTdw=
-github.com/evanphx/go-hclog-slog v0.0.0-20240717231540-be48fc4c4df5/go.mod h1:30+1dTR5EdDQGmcjkgOj4i6iVD3wnI0XymSOTPMlUeA=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
 github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=