fix(session): allow explicit deletion after task settlement failure

docs/architecture/a2a-gateway.md

@@ -80,6 +80,20 @@ and history in one transaction. Clients can read completion even when runtime
 pause/suspend is slow, unavailable, or has not started. TaskStore performs no
 runtime lifecycle calls and runs no background workers.
 
+Explicit Session deletion can proceed even if a runtime died before acknowledging
+its saved boundary ([#3010](https://github.com/kagent-dev/kagent/issues/3010)).
+It closes task admission and retains the deletion intent until Substrate confirms
+compute deletion. It does not publish the pending boundary or recover the same
+conversation. Unexpired dispatches, claimed idle work, and checkpoint creation
+still block deletion; a failed backend call leaves deletion pending for retry.
+Deletion admission and settlement take the same session lock. If deletion wins,
+a late acknowledgement cannot publish the pending state/history or enqueue a
+push notification, including after deletion preparation fails. If settlement
+committed first, acknowledging that published version remains idempotent until
+final deletion.
+Idle expiration retains the native-cleanup requirement and never applies this
+explicit-deletion exception.
+
 An independent Session lifecycle worker pauses waiting actors or suspends
 terminal ones and records their matching snapshot. A new turn can supersede idle
 work that has not been claimed. Once claimed, pause/suspend blocks new execution,

go/core/internal/database/active_task_test.go

@@ -32,6 +32,8 @@ func TestCheckpointCreationBlocksSessionTaskWrites(t *testing.T) {
 	next.ContextID = session.GetContextId()
 	_, err = client.CreateRuntimeTask(ctx, session.GetId(), taskMutationHash("next-request"), next, "")
 	require.ErrorIs(t, err, ErrConflict)
+	_, err = client.BeginSessionOperation(ctx, session.GetId(), apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE)
+	require.ErrorIs(t, err, ErrConflict, "explicit deletion must not supersede checkpoint creation")
 	require.ErrorIs(t, saveRuntimeTask(t, client, session.GetId(), task, task, nil), ErrFailedPrecondition)
 	_, _, err = client.BeginDeleteSessionCheckpoint(ctx, checkpoint.GetId(), "alice")
 	require.ErrorIs(t, err, ErrNotFound)

go/core/internal/database/checkpoints.go

@@ -252,7 +252,7 @@ func (c *Client) ReserveSessionCheckpoint(ctx context.Context, checkpoint *apiv1
 		if pending {
 			return fmt.Errorf("checkpoint requires all tasks to be terminal: %w", ErrFailedPrecondition)
 		}
-		if err := requireSettledRuntime(ctx, tx, session.HistoryID, ""); err != nil {
+		if err := requireRuntimeAdmission(ctx, tx, session.HistoryID, "", false); err != nil {
 			if errors.Is(err, ErrFailedPrecondition) {
 				return ErrSnapshotPending
 			}

go/core/internal/database/session_expiration.go

@@ -81,15 +81,17 @@ func (c *Client) BeginIdleSessionDeletion(ctx context.Context, id string, before
 	return deletion, nil
 }
 
-// requireIdleSession runs under the session lock, before lifecycle admission
-// checks dispatch, native cleanup, quiescence, and checkpoint creation.
+// requireIdleSession runs under the session lock. Unlike explicit deletion,
+// expiration must wait for native cleanup. Lifecycle admission additionally
+// checks dispatch, claimed idle work, and checkpoint creation.
 func requireIdleSession(ctx context.Context, tx pgx.Tx, row sessionRow, idleSince, before time.Time) error {
 	if idleSince.After(before) || row.State == "RUNTIME_STATE_DELETED" || row.State == "RUNTIME_STATE_DELETING" || row.Operation != "RUNTIME_OPERATION_NONE" {
 		return ErrConflict
 	}
 	busy, err := queryOne(ctx, tx, `
 		SELECT EXISTS (SELECT 1 FROM session_task WHERE history_id = $1
 		    AND state NOT IN ('TASK_STATE_COMPLETED', 'TASK_STATE_CANCELED', 'TASK_STATE_FAILED', 'TASK_STATE_REJECTED'))
+		    OR EXISTS (SELECT 1 FROM session_task_event WHERE history_id = $1 AND NOT published)
 	`, pgx.RowTo[bool], row.HistoryID)
 	if err != nil {
 		return err

go/core/internal/database/session_expiration_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/require"
 
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
@@ -55,7 +56,15 @@ func TestSessionExpirationTaskAdmission(t *testing.T) {
 func TestSessionExpirationRechecksActivity(t *testing.T) {
 	client, session := expirationFixture(t)
 	ctx := t.Context()
-	before := time.Now()
+	// Event timestamps come from PostgreSQL. Use its clock for the precise
+	// before/after cutoffs too: a container's clock can lag the test host.
+	databaseTime := func() time.Time {
+		t.Helper()
+		now, err := queryOne(ctx, client.db, `SELECT clock_timestamp()`, pgx.RowTo[time.Time])
+		require.NoError(t, err)
+		return now
+	}
+	before := databaseTime()
 	ids, err := client.ListIdleSessions(ctx, before, "", 100)
 	require.NoError(t, err)
 	require.Equal(t, []string{session.Id}, ids)
@@ -64,13 +73,13 @@ func TestSessionExpirationRechecksActivity(t *testing.T) {
 	task.ContextID, task.Status.State = session.ContextId, a2a.TaskStateCompleted
 	version, err := client.CreateRuntimeTask(ctx, session.Id, taskMutationHash("completed"), task, "")
 	require.NoError(t, err)
-	savedBefore := time.Now()
-	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, time.Now())
+	savedBefore := databaseTime()
+	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, databaseTime())
 	require.ErrorIs(t, err, ErrConflict, "unpublished runtime cleanup blocks expiration")
 	require.NoError(t, client.SettleSessionTask(ctx, session.Id, string(task.ID), version))
 	work, err := client.ClaimSessionQuiescence(ctx)
 	require.NoError(t, err)
-	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, time.Now())
+	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, databaseTime())
 	require.ErrorIs(t, err, ErrFailedPrecondition, "a claimed suspension blocks expiration")
 	require.NoError(t, client.FinishSessionQuiescence(ctx, work, &SessionTaskSnapshot{Atespace: "team-a", URI: "s3://snapshot", ContentScope: "FULL"}))
 	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, before)
@@ -93,6 +102,8 @@ func TestSessionExpirationFencesDispatchAndRetries(t *testing.T) {
 	require.NoError(t, client.ReserveSessionDispatch(ctx, session.Id, dispatch, "message"))
 	_, err := client.BeginIdleSessionDeletion(ctx, session.Id, time.Now())
 	require.ErrorIs(t, err, ErrFailedPrecondition)
+	_, err = client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE)
+	require.ErrorIs(t, err, ErrFailedPrecondition, "explicit deletion must not supersede a live dispatch")
 	_, err = client.RevokeSessionDispatch(ctx, session.Id, dispatch, "message")
 	require.NoError(t, err)
 	_, err = client.BeginIdleSessionDeletion(ctx, session.Id, time.Time{})

go/core/internal/database/session_operations.go

@@ -87,7 +87,11 @@ func beginSessionOperation(ctx context.Context, tx pgx.Tx, row sessionRow, kind
 // admitAgentLifecycle runs under the same row lock as task and checkpoint
 // admission. Common transitions have no knowledge of these agent-only records.
 func admitAgentLifecycle(ctx context.Context, tx pgx.Tx, row sessionRow, kind apiv1.RuntimeOperation) error {
-	if err := requireSettledRuntime(ctx, tx, row.HistoryID, ""); err != nil {
+	// Explicit deletion can supersede an unpublished boundary (#3010, case B).
+	// It closes admission and retains its intent until compute deletion succeeds;
+	// it does not publish the boundary or permit reuse of the conversation.
+	// Claimed lifecycle work remains protected until recovery can fence its owner.
+	if err := requireRuntimeAdmission(ctx, tx, row.HistoryID, "", kind == apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE); err != nil {
 		return err
 	}
 	busy, err := queryOne(ctx, tx, `

go/core/internal/database/session_tasks.go

@@ -61,7 +61,7 @@ func (c *Client) ReserveSessionDispatch(ctx context.Context, sessionID string, d
 		if blocked {
 			return ErrConflict
 		}
-		if err := requireSettledRuntime(ctx, tx, session.HistoryID, ""); err != nil {
+		if err := requireRuntimeAdmission(ctx, tx, session.HistoryID, "", false); err != nil {
 			if errors.Is(err, ErrFailedPrecondition) {
 				return ErrDispatchBusy
 			}
@@ -201,7 +201,7 @@ func (c *Client) writeRuntimeTask(ctx context.Context, sessionID string, expecte
 				return fmt.Errorf("dispatch expired or was revoked before acceptance: %w", ErrFailedPrecondition)
 			}
 		}
-		if err := requireSettledRuntime(ctx, tx, session.HistoryID, dispatchID); err != nil {
+		if err := requireRuntimeAdmission(ctx, tx, session.HistoryID, dispatchID, false); err != nil {
 			return err
 		}
 		// New execution supersedes idle work that no worker has claimed yet.
@@ -809,16 +809,17 @@ func readTaskMessages(ctx context.Context, db dbExecutor, historyID uuid.UUID, t
 	`, pgx.RowToStructByName[taskHistoryRow], historyID, taskIDs, historyLength, includeUnpublished)
 }
 
-// requireSettledRuntime rejects unfinished native cleanup, claimed idle work, or
-// another unexpired dispatch. A writer may pass its own dispatch ID.
+// requireRuntimeAdmission rejects claimed idle work and another unexpired
+// dispatch. A writer may pass its own dispatch ID. Unpublished events also block
+// admission unless allowUnpublished is set for explicit deletion.
 // Callers hold the session lock so new execution and lifecycle claims cannot race.
-func requireSettledRuntime(ctx context.Context, db dbExecutor, historyID uuid.UUID, dispatchID string) error {
+func requireRuntimeAdmission(ctx context.Context, db dbExecutor, historyID uuid.UUID, dispatchID string, allowUnpublished bool) error {
 	pending, err := queryOne(ctx, db, `
 		SELECT EXISTS (SELECT 1 FROM session_task_event WHERE history_id = $1
-		    AND (NOT published OR (quiescence_pending AND quiescence_executor_id IS NOT NULL)))
+		    AND ((NOT published AND NOT $3::boolean) OR (quiescence_pending AND quiescence_executor_id IS NOT NULL)))
 		    OR EXISTS (SELECT 1 FROM session WHERE history_id = $1 AND dispatch_expires_at > clock_timestamp()
 		        AND ($2::text = '' OR dispatch_id::text <> $2))
-	`, pgx.RowTo[bool], historyID, dispatchID)
+	`, pgx.RowTo[bool], historyID, dispatchID, allowUnpublished)
 	if err != nil {
 		return err
 	}

go/core/internal/database/task_finalization.go

@@ -13,7 +13,8 @@ import (
 // SettleSessionTask publishes a saved terminal/waiting task and its history
 // atomically after native cleanup. It does not wait for runtime pause or snapshots.
 // Retries cannot republish an old state over a later turn. Callers authenticate
-// runtime authority; a missing or deleted session returns ErrNotFound.
+// runtime authority; a missing or deleted session returns ErrNotFound. Deletion
+// supersedes unpublished boundaries; their late acknowledgements return ErrConflict.
 func (c *Client) SettleSessionTask(ctx context.Context, sessionID, taskID string, version int64) error {
 	return c.withTx(ctx, func(tx pgx.Tx) error {
 		session, err := lockSession(ctx, tx, sessionID)
@@ -42,6 +43,12 @@ func (c *Client) SettleSessionTask(ctx context.Context, sessionID, taskID string
 		if row.Published {
 			return nil
 		}
+		// Deletion keeps admission closed even if preparation clears the
+		// operation. Reject unpublished acknowledgements in that state so
+		// discarded results cannot become public or enqueue callbacks.
+		if session.State == "RUNTIME_STATE_DELETING" {
+			return ErrConflict
+		}
 		stored, err := readSessionTask(ctx, tx, session.HistoryID, taskID)
 		if err != nil {
 			return err

go/core/internal/database/task_store_test.go

@@ -182,7 +182,8 @@ func TestRuntimeCompletionDoesNotWaitForSnapshot(t *testing.T) {
 	require.Equal(t, a2a.TaskStateWorking, public.Status.State)
 	_, err = client.GetSettledSessionTask(t.Context(), session.Id, string(waiting.ID), nil)
 	require.ErrorIs(t, err, ErrConflict)
-	require.ErrorIs(t, deleteSession(t.Context(), client, session.Id), ErrFailedPrecondition)
+	_, err = client.BeginSessionOperation(t.Context(), session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_SUSPEND)
+	require.ErrorIs(t, err, ErrFailedPrecondition)
 	_, err = client.ClaimSessionQuiescence(t.Context())
 	require.ErrorIs(t, err, ErrNotFound) // The native cleanup callback has not finished.
 	require.NoError(t, client.SettleSessionTask(t.Context(), session.Id, string(waiting.ID), version))
@@ -218,6 +219,69 @@ func TestRuntimeCompletionDoesNotWaitForSnapshot(t *testing.T) {
 	require.ErrorIs(t, err, ErrNotFound)
 }
 
+func TestSettlementSerializesWithSessionDeletion(t *testing.T) {
+	client := NewClient(setupTestDB(t), "public")
+	ctx := t.Context()
+	sessionFixture(t, client, ctx, "team-a", "revision", "assistant", "kagent")
+	for _, ordering := range []string{"delete_first", "settle_first", "concurrent"} {
+		t.Run(ordering, func(t *testing.T) {
+			session, _, err := client.CreateSession(ctx, newSessionRequest(uuid.NewString(), "assistant", "kagent", ""), uuid.NewString())
+			require.NoError(t, err)
+			session, err = markSessionReady(ctx, client, session.Id, "agent.example")
+			require.NoError(t, err)
+			task := newSessionTask(uuid.NewString(), uuid.NewString())
+			task.ContextID, task.Status.State = session.ContextId, a2a.TaskStateCompleted
+			require.NoError(t, client.RegisterSessionPushNotification(ctx, session.Id, task.History[0].ID, "", &a2a.PushConfig{ID: "callback", URL: "http://receiver"}))
+			version, err := client.CreateRuntimeTask(ctx, session.Id, taskMutationHash("completed"), task, "")
+			require.NoError(t, err)
+			settle := func() error { return client.SettleSessionTask(ctx, session.Id, string(task.ID), version) }
+			remove := func() error {
+				_, err := client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE)
+				return err
+			}
+			var settleErr error
+			switch ordering {
+			case "delete_first":
+				require.NoError(t, remove())
+				settleErr = settle()
+				require.ErrorIs(t, settleErr, ErrConflict)
+			case "settle_first":
+				settleErr = settle()
+				require.NoError(t, settleErr)
+				require.NoError(t, remove())
+			case "concurrent":
+				start := make(chan struct{})
+				settled, deleted := make(chan error, 1), make(chan error, 1)
+				go func() { <-start; settled <- settle() }()
+				go func() { <-start; deleted <- remove() }()
+				close(start)
+				require.NoError(t, <-deleted)
+				settleErr = <-settled
+				if settleErr != nil {
+					require.ErrorIs(t, settleErr, ErrConflict)
+				}
+			}
+			public, err := client.GetSessionTask(ctx, session.Id, string(task.ID), nil)
+			require.NoError(t, err)
+			delivery, err := client.ClaimDuePushDelivery(ctx)
+			require.NoError(t, err)
+			if settleErr == nil {
+				require.Equal(t, a2a.TaskStateCompleted, public.Status.State)
+				require.NotNil(t, delivery)
+				require.Equal(t, string(task.ID), delivery.TaskID)
+				require.NoError(t, client.FinishPushDelivery(ctx, *delivery, true))
+				// Acknowledging a publication that won the race remains idempotent.
+				require.NoError(t, settle())
+			} else {
+				require.Equal(t, a2a.TaskStateSubmitted, public.Status.State)
+				require.Nil(t, delivery)
+			}
+			_, err = client.ClaimSessionQuiescence(ctx)
+			require.ErrorIs(t, err, ErrNotFound, "deletion excludes background runtime work")
+		})
+	}
+}
+
 func TestRuntimeForkRetainsOnlyTheCheckpointBoundary(t *testing.T) {
 	client := NewClient(setupTestDB(t), "public")
 	sessionFixture(t, client, t.Context(), "team-a", "revision", "assistant", "kagent")