refactor: remove local HITL task store (#2572)

## Summary

- remove the runtime-local SQLite A2A task store added by #2567
- pass canonical waiting task state from the gateway PostgreSQL store to
newly resumed runtimes
- seed the upstream in-memory task store before HITL continuation
- use the typed A2A gRPC transport in the suspension E2E test

The reserved continuation metadata is cleared from public input,
populated from canonical gateway state, consumed before task history is
stored, and supports both input-required and auth-required states.

## Test plan

- go test ./api/a2a ./adk/pkg/app ./adk/pkg/a2a ./core/v2/a2agateway
- go test ./core/test/e2e -run
'^TestAgentInstanceAskUserSurvivesSuspension$'
- git diff --check

Related to #2567.

Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io>

Author: EItanya

go/adk/cmd/main.go

@@ -241,7 +241,6 @@ func main() {
 		ShutdownTimeout: 5 * time.Second,
 		Logger:          logger,
 		Agent:           runnerConfig.Agent,
-		SessionDBURL:    agentConfig.SessionDBURL,
 	}, executor)
 	if err != nil {
 		logger.Error(err, "Failed to create app")

go/adk/pkg/app/app.go

@@ -15,7 +15,7 @@ import (
 	"github.com/go-logr/zapr"
 	"github.com/kagent-dev/kagent/go/adk/pkg/a2a"
 	"github.com/kagent-dev/kagent/go/adk/pkg/a2a/server"
-	localtaskstore "github.com/kagent-dev/kagent/go/adk/pkg/taskstore"
+	apia2a "github.com/kagent-dev/kagent/go/api/a2a"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	adkagent "google.golang.org/adk/v2/agent"
@@ -56,10 +56,6 @@ type AppConfig struct {
 	// Agent is the ADK agent used to enrich the agent card with skills via
 	// adka2a.BuildAgentSkills. Optional; when nil, the card is used as-is.
 	Agent adkagent.Agent
-
-	// SessionDBURL locates actor-local durable storage. When set, A2A tasks are
-	// persisted beside the ADK session database so HITL tasks survive suspension.
-	SessionDBURL string
 }
 
 // KAgentApp wires an AgentExecutor with kagent's A2A server.
@@ -81,12 +77,19 @@ func (i seedTaskInterceptor) Before(ctx context.Context, _ *a2asrv.CallContext,
 	if !ok || send.Message == nil || send.Message.TaskID == "" {
 		return ctx, nil, nil
 	}
+	storedTask, err := apia2a.TakeStoredTask(send.Message)
+	if err != nil {
+		return ctx, nil, err
+	}
 	if _, err := i.store.Get(ctx, send.Message.TaskID); err == nil {
 		return ctx, nil, nil
 	} else if !errors.Is(err, a2atype.ErrTaskNotFound) {
 		return ctx, nil, fmt.Errorf("load actor task: %w", err)
 	}
-	if _, err := i.store.Create(ctx, a2atype.NewSubmittedTask(send.Message, send.Message)); err != nil && !errors.Is(err, a2ataskstore.ErrTaskAlreadyExists) {
+	if storedTask == nil {
+		storedTask = a2atype.NewSubmittedTask(send.Message, send.Message)
+	}
+	if _, err := i.store.Create(ctx, storedTask); err != nil && !errors.Is(err, a2ataskstore.ErrTaskAlreadyExists) {
 		return ctx, nil, fmt.Errorf("seed actor task: %w", err)
 	}
 	return ctx, nil, nil
@@ -104,17 +107,7 @@ func New(cfg AppConfig, executor a2asrv.AgentExecutor) (*KAgentApp, error) {
 	log := cfg.Logger
 
 	app := &KAgentApp{logger: log}
-	authenticator := a2asrv.NewTaskStoreAuthenticator()
-	var tasks a2ataskstore.Store
-	if cfg.SessionDBURL == "" {
-		tasks = a2ataskstore.NewInMemory(&a2ataskstore.InMemoryStoreConfig{Authenticator: authenticator})
-	} else {
-		var err error
-		tasks, err = localtaskstore.New(cfg.SessionDBURL, authenticator)
-		if err != nil {
-			return nil, fmt.Errorf("open local task store: %w", err)
-		}
-	}
+	tasks := a2ataskstore.NewInMemory(&a2ataskstore.InMemoryStoreConfig{Authenticator: a2asrv.NewTaskStoreAuthenticator()})
 	handlerOpts := []a2asrv.RequestHandlerOption{a2asrv.WithTaskStore(tasks)}
 
 	// The private runtime receives a gateway-assigned ID for a new task. Seed it

go/adk/pkg/app/app_test.go

@@ -9,6 +9,8 @@ import (
 	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/a2aproject/a2a-go/v2/a2asrv"
 	a2ataskstore "github.com/a2aproject/a2a-go/v2/a2asrv/taskstore"
+	"github.com/kagent-dev/kagent/go/adk/pkg/a2a"
+	apia2a "github.com/kagent-dev/kagent/go/api/a2a"
 )
 
 // fakeExecutor implements a2asrv.AgentExecutor for testing.
@@ -60,6 +62,48 @@ func TestSeedTaskInterceptor(t *testing.T) {
 	}
 }
 
+func TestSeedTaskInterceptorRestoresWaitingTask(t *testing.T) {
+	store := a2ataskstore.NewInMemory(nil)
+	status := a2a.AttachHitlExtension(a2atype.NewMessage(a2atype.MessageRoleAgent), &a2a.AskUserRequest{
+		Type: a2a.HITLTypeAskUserRequest, ID: "question-1",
+	})
+	waiting := &a2atype.Task{
+		ID: "task-1", ContextID: "instance-1",
+		Status: a2atype.TaskStatus{State: a2atype.TaskStateInputRequired, Message: status},
+	}
+	reply := a2atype.NewMessage(a2atype.MessageRoleUser)
+	reply.TaskID, reply.ContextID = waiting.ID, waiting.ContextID
+	if err := apia2a.AttachStoredTask(reply, waiting); err != nil {
+		t.Fatal(err)
+	}
+	interceptor := seedTaskInterceptor{store: store}
+	if _, _, err := interceptor.Before(t.Context(), nil, &a2asrv.Request{Payload: &a2atype.SendMessageRequest{Message: reply}}); err != nil {
+		t.Fatal(err)
+	}
+	stored, err := store.Get(t.Context(), waiting.ID)
+	if err != nil || stored.Task.Status.State != a2atype.TaskStateInputRequired || a2a.GetAskUserRequest(stored.Task.Status.Message) == nil {
+		t.Fatalf("restored task = %#v, error = %v", stored, err)
+	}
+	if restored, err := apia2a.TakeStoredTask(reply); err != nil || restored != nil {
+		t.Fatalf("private state was not consumed: task = %#v, error = %v", restored, err)
+	}
+}
+
+func TestStoredTaskMetadataSupportsAuthRequired(t *testing.T) {
+	waiting := &a2atype.Task{
+		ID: "task-1", ContextID: "instance-1",
+		Status: a2atype.TaskStatus{State: a2atype.TaskStateAuthRequired},
+	}
+	reply := &a2atype.Message{TaskID: waiting.ID, ContextID: waiting.ContextID}
+	if err := apia2a.AttachStoredTask(reply, waiting); err != nil {
+		t.Fatal(err)
+	}
+	restored, err := apia2a.TakeStoredTask(reply)
+	if err != nil || restored == nil || restored.Status.State != a2atype.TaskStateAuthRequired || restored.Status.Message != nil {
+		t.Fatalf("restored task = %#v, error = %v", restored, err)
+	}
+}
+
 func TestApplyDefaults_Port(t *testing.T) {
 	t.Setenv("PORT", "")
 	cfg := applyDefaults(AppConfig{})