runtime/session: suspend actors after abandoned delivery

docs/architecture/runtime-and-lifecycle.md

@@ -208,14 +208,39 @@ leave an actor running until a surviving replica's next recovery scan; increasin
 the interval increases that delay and the time until its snapshot is available.
 Ordinary committed settlements do not wait for the recovery interval.
 
+The existing Session dispatch ID and expiry also track abandoned delivery.
+Revocation expires the reservation immediately, retains its ID, and wakes the
+same workers. An unrevoked reservation becomes eligible at its existing expiry.
+This covers cancellation after actor restore but before a turn starts, when no
+task boundary exists to trigger suspension. Accepted execution clears the
+reservation; saving continuation input without leaving a waiting state retains
+it. An expired reservation cannot authorize a late first task write.
+
+For abandoned dispatches, the worker first reads and verifies the actor without
+blocking new messages. Only a RUNNING actor can be claimed for suspension. The
+claim atomically rechecks the dispatch generation, absence of active tasks, and
+other lifecycle work. A newer message supersedes unclaimed cleanup; a claimed
+suspension blocks new execution until its snapshot is recorded and the dispatch
+reservation and claim are cleared. Cleanup preserves
+the existing history and logical READY state. When history exists, its captured
+head receives the snapshot without adding a task event.
+
+RESUMING or SUSPENDED actors and failed preparation reads remain unclaimed. Each
+worker visits candidates in Session ID order once per scan, then revisits them on
+the existing worker wakeups and recovery scans. A SUSPENDED observation alone
+cannot prove that canceled ingress will not finish restoring later. Such cleanup
+can remain pending until an actor runs, a new message supersedes it, or an explicit
+lifecycle operation retires it. This is cancellation cleanup, not general idle
+actor detection.
+
 An empty claim is followed by a check for unclaimed work on ready sessions. If
 dispatch, checkpoint creation, a lifecycle operation, or a row lock blocks that
 work, workers retry within one second until it becomes eligible or is superseded.
 Database errors also use this short retry. Sessions that are not READY and claimed
 work do not keep the short retry active. Idle workers make one claim attempt and
 one pending-work check per recovery interval, rather than querying every second.
 Other background workers and connection-pool idle settings still affect whether
-PostgreSQL can become idle.
+PostgreSQL can become idle. Each scan also checks for abandoned dispatch cleanup.
 
 Unclaimed idle work survives API restarts. A claim for possibly issued runtime work
 never expires: losing the worker does not prove that the suspend stopped. Uncertain

go/core/internal/a2agateway/routing_test.go

@@ -54,7 +54,7 @@ func newTestSessions(store *gatewayTestStore, authorizer auth.Authorizer) *sessi
 }
 
 func newTestGateway(store *gatewayTestStore, authorizer auth.Authorizer, dialer *gatewayTestDialer, url string) a2asrv.RequestHandler {
-	interactions := sessionsvc.NewInteractionService(store, gatewayTestAgents{store, authorizer}, newTestSessions(store, authorizer), false)
+	interactions := sessionsvc.NewInteractionService(store, gatewayTestAgents{store, authorizer}, newTestSessions(store, authorizer), false, nil)
 	return New(interactions, dialer, url)
 }
 

go/core/internal/controller/scheduledrun/controller_test.go

@@ -338,7 +338,7 @@ func TestControllerDispatchUsesBoundedInProcessAuthority(t *testing.T) {
 	store := &scopedControllerStore{session: &apiv1.Session{Id: id, ContextId: id, Creator: "owner", Agent: &apiv1.AgentProvenance{Ref: &apiv1.ResourceReference{Namespace: "team", Name: "assistant"}}, State: apiv1.RuntimeState_RUNTIME_STATE_READY, Operation: apiv1.RuntimeOperation_RUNTIME_OPERATION_NONE}}
 	policy := &controllerSessionPolicy{}
 	sessions := sessionsvc.NewService(store, policy, nil)
-	interactions := sessionsvc.NewInteractionService(store, nil, sessions, false)
+	interactions := sessionsvc.NewInteractionService(store, nil, sessions, false, nil)
 	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
 	defer cancel()
 	gateway := scopedControllerGateway{interactions: interactions, cancel: cancel, agent: types.NamespacedName{Namespace: "team", Name: "assistant"}}

go/core/internal/database/session_deletion.go

@@ -48,7 +48,7 @@ func beginSessionDeletion(ctx context.Context, tx pgx.Tx, row sessionRow, reason
 	if err != nil {
 		return nil, err
 	}
-	if err := execSQL(ctx, tx, `UPDATE session SET dispatch_id = NULL, dispatch_expires_at = NULL WHERE id = $1`, row.ID); err != nil {
+	if err := retireSessionDispatchCleanup(ctx, tx, row.ID); err != nil {
 		return nil, err
 	}
 	if err := execSQL(ctx, tx, `UPDATE session SET deletion_reason = COALESCE(deletion_reason, $2) WHERE id = $1`, row.ID, string(reason)); err != nil {
@@ -68,6 +68,7 @@ func admitForcedSessionDeletion(ctx context.Context, tx pgx.Tx, row sessionRow)
 		SELECT EXISTS (SELECT 1 FROM session_checkpoint WHERE source_session_id = $1 AND state = 'CREATING')
 		    OR EXISTS (SELECT 1 FROM session_task_event WHERE history_id = $2
 		        AND quiescence_pending AND quiescence_executor_id IS NOT NULL)
+		    OR EXISTS (SELECT 1 FROM session WHERE id = $1 AND dispatch_cleanup_executor_id IS NOT NULL)
 	`, pgx.RowTo[bool], row.ID, row.HistoryID)
 	if err != nil {
 		return err

go/core/internal/database/session_dispatch_cleanup.go

@@ -0,0 +1,202 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+
+	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
+)
+
+// SessionDispatchCleanup identifies an abandoned delivery's runtime cleanup.
+// An unclaimed candidate grants no execution authority. Claim returns a new
+// value whose executor fences runtime work until its known outcome is stored.
+type SessionDispatchCleanup struct {
+	Session    *apiv1.Session
+	DispatchID uuid.UUID
+	ActorUID   string
+	ExecutorID uuid.UUID
+	boundary   *dispatchCleanupBoundary
+}
+
+type dispatchCleanupBoundary struct {
+	TaskID  string
+	Version int64
+	State   string
+}
+
+// GetPendingSessionDispatchCleanup observes the next revoked or expired delivery
+// after the Session cursor without holding locks during runtime preparation.
+// Missing or blocked work returns ErrNotFound. Claim revalidates the generation;
+// advancing the cursor lets a worker skip an actor that is not ready to suspend.
+func (c *Client) GetPendingSessionDispatchCleanup(ctx context.Context, afterSessionID string) (*SessionDispatchCleanup, error) {
+	type candidate struct {
+		SessionID  string
+		DispatchID uuid.UUID
+		ActorUID   string
+	}
+	row, err := queryOne(ctx, c.db, `
+		SELECT s.id::text AS session_id, s.dispatch_id, s.actor_uid
+		FROM session s JOIN runtime_instance r USING (id)
+		WHERE s.dispatch_id IS NOT NULL
+		  AND s.dispatch_cleanup_executor_id IS NULL AND s.actor_uid IS NOT NULL
+		  AND ($1::text = '' OR s.id > NULLIF($1::text, '')::uuid)
+		  AND r.state = 'RUNTIME_STATE_READY' AND r.operation = 'RUNTIME_OPERATION_NONE'
+		  AND s.dispatch_expires_at <= clock_timestamp()
+		  AND NOT EXISTS (SELECT 1 FROM session_task t WHERE t.history_id = s.history_id
+		      AND t.state NOT IN ('TASK_STATE_COMPLETED', 'TASK_STATE_CANCELED', 'TASK_STATE_FAILED',
+		          'TASK_STATE_REJECTED', 'TASK_STATE_INPUT_REQUIRED', 'TASK_STATE_AUTH_REQUIRED'))
+		  AND NOT EXISTS (SELECT 1 FROM session_task_event e WHERE e.history_id = s.history_id
+		      AND (NOT e.published OR (e.quiescence_pending AND e.quiescence_executor_id IS NOT NULL)))
+		  AND NOT EXISTS (SELECT 1 FROM session_checkpoint WHERE source_session_id = s.id AND state = 'CREATING')
+		ORDER BY s.id LIMIT 1
+	`, pgx.RowToStructByName[candidate], afterSessionID)
+	if err != nil {
+		return nil, notFoundOr(err)
+	}
+	session, err := readSession(ctx, c.db, row.SessionID)
+	if err != nil {
+		return nil, notFoundOr(err)
+	}
+	value, err := toSession(session)
+	if err != nil {
+		return nil, err
+	}
+	return &SessionDispatchCleanup{Session: value, DispatchID: row.DispatchID, ActorUID: row.ActorUID}, nil
+}
+
+// ClaimSessionDispatchCleanup grants exclusive runtime authority for this exact
+// delivery and actor. Admission and history capture commit together. A stale or
+// blocked candidate returns ErrNotFound; issued work is never reclaimed by time.
+func (c *Client) ClaimSessionDispatchCleanup(ctx context.Context, candidate *SessionDispatchCleanup) (*SessionDispatchCleanup, error) {
+	if candidate == nil || candidate.Session == nil || candidate.DispatchID == uuid.Nil || candidate.ActorUID == "" {
+		return nil, ErrFailedPrecondition
+	}
+	var result *SessionDispatchCleanup
+	err := c.withTx(ctx, func(tx pgx.Tx) error {
+		session, err := lockSession(ctx, tx, candidate.Session.Id)
+		if err != nil {
+			return notFoundOr(err)
+		}
+		executor := uuid.New()
+		tag, err := tx.Exec(ctx, `
+			UPDATE session s SET dispatch_cleanup_executor_id = $4
+			FROM runtime_instance r
+			WHERE s.id = $1 AND r.id = s.id AND s.dispatch_id = $2 AND s.actor_uid = $3
+			  AND s.dispatch_cleanup_executor_id IS NULL
+			  AND r.state = 'RUNTIME_STATE_READY' AND r.operation = 'RUNTIME_OPERATION_NONE'
+			  AND s.dispatch_expires_at <= clock_timestamp()
+			  AND NOT EXISTS (SELECT 1 FROM session_task t WHERE t.history_id = s.history_id
+			      AND t.state NOT IN ('TASK_STATE_COMPLETED', 'TASK_STATE_CANCELED', 'TASK_STATE_FAILED',
+			          'TASK_STATE_REJECTED', 'TASK_STATE_INPUT_REQUIRED', 'TASK_STATE_AUTH_REQUIRED'))
+			  AND NOT EXISTS (SELECT 1 FROM session_task_event e WHERE e.history_id = s.history_id
+			      AND (NOT e.published OR (e.quiescence_pending AND e.quiescence_executor_id IS NOT NULL)))
+			  AND NOT EXISTS (SELECT 1 FROM session_checkpoint WHERE source_session_id = s.id AND state = 'CREATING')
+		`, session.ID, candidate.DispatchID, candidate.ActorUID, executor)
+		if err != nil {
+			return err
+		}
+		if tag.RowsAffected() != 1 {
+			return ErrNotFound
+		}
+		value, err := toSession(session)
+		if err != nil {
+			return err
+		}
+		result = &SessionDispatchCleanup{Session: value, DispatchID: candidate.DispatchID, ActorUID: candidate.ActorUID, ExecutorID: executor}
+		boundary, err := queryOne(ctx, tx, `
+			SELECT e.task_id, e.sequence AS version, t.state FROM session_task_event e
+			JOIN session_task t ON t.history_id = e.history_id AND t.id = e.task_id
+			WHERE e.history_id = $1 ORDER BY e.sequence DESC LIMIT 1
+		`, pgx.RowToStructByName[dispatchCleanupBoundary], session.HistoryID)
+		if err == nil {
+			result.boundary = &boundary
+			return nil
+		}
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return err
+		}
+		hasTasks, err := queryOne(ctx, tx, `SELECT EXISTS (SELECT 1 FROM session_task WHERE history_id = $1)`, pgx.RowTo[bool], session.HistoryID)
+		if err != nil {
+			return err
+		}
+		if hasTasks {
+			return fmt.Errorf("session task history is missing: %w", ErrFailedPrecondition)
+		}
+		return nil
+	})
+	return result, err
+}
+
+// FinishSessionDispatchCleanup records the claimed snapshot without inventing
+// history or changing logical Session state. It releases admission atomically
+// with snapshot metadata and obsolete task-quiescence work. Completed retries
+// do nothing; a newer cleanup generation or owner returns ErrConflict.
+func (c *Client) FinishSessionDispatchCleanup(ctx context.Context, work *SessionDispatchCleanup, snapshot *SessionTaskSnapshot) error {
+	if work == nil || work.Session == nil || work.DispatchID == uuid.Nil || work.ExecutorID == uuid.Nil {
+		return ErrFailedPrecondition
+	}
+	if snapshot == nil || snapshot.Atespace == "" || snapshot.URI == "" || snapshot.Fidelity == "" {
+		return fmt.Errorf("dispatch cleanup requires a runtime snapshot: %w", ErrFailedPrecondition)
+	}
+	return c.withTx(ctx, func(tx pgx.Tx) error {
+		session, err := lockSession(ctx, tx, work.Session.Id)
+		if err != nil {
+			return notFoundOr(err)
+		}
+		type claim struct {
+			ID       *uuid.UUID
+			Executor *uuid.UUID
+			ActorUID *string
+		}
+		current, err := queryOne(ctx, tx, `
+			SELECT dispatch_id AS id, dispatch_cleanup_executor_id AS executor, actor_uid
+			FROM session WHERE id = $1
+		`, pgx.RowToStructByName[claim], session.ID)
+		if err != nil {
+			return err
+		}
+		if current.ID == nil {
+			return nil
+		}
+		if *current.ID != work.DispatchID || current.Executor == nil || *current.Executor != work.ExecutorID || current.ActorUID == nil || *current.ActorUID != work.ActorUID {
+			return ErrConflict
+		}
+		if target := work.boundary; target != nil {
+			latest, err := queryOne(ctx, tx, `SELECT MAX(sequence) FROM session_task_event WHERE history_id = $1`, pgx.RowTo[int64], session.HistoryID)
+			if err != nil {
+				return err
+			}
+			stored, err := readSessionTask(ctx, tx, session.HistoryID, target.TaskID)
+			if err != nil {
+				return err
+			}
+			if latest != target.Version || stored.State != target.State {
+				return ErrConflict
+			}
+			if err := saveSessionTaskSnapshot(ctx, tx, session.HistoryID, target.TaskID, target.Version, snapshot); err != nil {
+				return err
+			}
+		}
+		if err := execSQL(ctx, tx, `
+			UPDATE session_task_event SET quiescence_pending = FALSE
+			WHERE history_id = $1 AND quiescence_pending AND quiescence_executor_id IS NULL
+		`, session.HistoryID); err != nil {
+			return err
+		}
+		return retireSessionDispatchCleanup(ctx, tx, session.ID)
+	})
+}
+
+// retireSessionDispatchCleanup removes compensation after execution or an
+// authoritative lifecycle outcome takes responsibility. Callers hold the Session
+// lock and must have checked admission or established runtime termination.
+func retireSessionDispatchCleanup(ctx context.Context, tx pgx.Tx, sessionID uuid.UUID) error {
+	return execSQL(ctx, tx, `
+		UPDATE session SET dispatch_id = NULL, dispatch_expires_at = NULL, dispatch_cleanup_executor_id = NULL
+		WHERE id = $1
+	`, sessionID)
+}

go/core/internal/database/session_operations.go

@@ -150,6 +150,15 @@ func (c *Client) FinishSessionOperation(ctx context.Context, sessionID string, i
 			return err
 		}
 		if failure == "" {
+			// Successful lifecycle work supersedes abandoned delivery from the
+			// prior runtime state. Retire its dispatch and suspension claim in
+			// this transaction so stale cleanup cannot suspend a resumed actor.
+			// Creation has no prior dispatch; failures leave cleanup intact.
+			if result.Operation != apiv1.RuntimeOperation_RUNTIME_OPERATION_CREATE {
+				if err := retireSessionDispatchCleanup(ctx, tx, row.ID); err != nil {
+					return err
+				}
+			}
 			switch result.Operation {
 			case apiv1.RuntimeOperation_RUNTIME_OPERATION_CREATE:
 				if authority == "" || actorUID == "" {

go/core/internal/database/session_quiescence.go

@@ -56,6 +56,7 @@ func (c *Client) ClaimSessionQuiescence(ctx context.Context) (*SessionQuiescence
 			  AND i.state = 'RUNTIME_STATE_READY'
 			  AND i.operation = 'RUNTIME_OPERATION_NONE'
 			  AND (i.dispatch_expires_at IS NULL OR i.dispatch_expires_at <= clock_timestamp())
+			  AND NOT EXISTS (SELECT 1 FROM session WHERE id = i.id AND dispatch_cleanup_executor_id IS NOT NULL)
 			  AND NOT EXISTS (SELECT 1 FROM session_checkpoint WHERE source_session_id = i.id AND state = 'CREATING')
 			ORDER BY e.sequence LIMIT 1 FOR UPDATE OF i SKIP LOCKED
 		`, pgx.RowToStructByName[candidate])
@@ -89,6 +90,7 @@ func (c *Client) ClaimSessionQuiescence(ctx context.Context) (*SessionQuiescence
 			WHERE sequence = $1 AND published AND quiescence_pending
 			  AND quiescence_executor_id IS NULL
 			  AND NOT EXISTS (SELECT 1 FROM session WHERE id = $3 AND dispatch_expires_at > clock_timestamp())
+			  AND NOT EXISTS (SELECT 1 FROM session WHERE id = $3 AND dispatch_cleanup_executor_id IS NOT NULL)
 			  AND NOT EXISTS (SELECT 1 FROM session_checkpoint WHERE source_session_id = $3 AND state = 'CREATING')
 		`, result.Version, result.ExecutorID, session.ID)
 		if err != nil {
@@ -138,23 +140,28 @@ func (c *Client) FinishSessionQuiescence(ctx context.Context, work *SessionQuies
 		if stored.State != string(work.State) {
 			return ErrConflict
 		}
-		if snapshot != nil {
-			if err := execSQL(ctx, tx, `
-				UPDATE session_task SET snapshot_atespace = $3, snapshot_uri = $4,
-				    snapshot_fidelity = $5, history_sequence = $6
-				WHERE history_id = $1 AND id = $2
-			`, session.HistoryID, work.TaskID, snapshot.Atespace, snapshot.URI, snapshot.Fidelity, work.Version); err != nil {
-				return err
-			}
-			if err := execSQL(ctx, tx, `
-				UPDATE session_task_event SET snapshot_atespace = $2, snapshot_uri = $3, snapshot_fidelity = $4
-				WHERE sequence = $1
-			`, work.Version, snapshot.Atespace, snapshot.URI, snapshot.Fidelity); err != nil {
-				return err
-			}
+		if err := saveSessionTaskSnapshot(ctx, tx, session.HistoryID, work.TaskID, work.Version, snapshot); err != nil {
+			return err
 		}
 		return execSQL(ctx, tx, `
 			UPDATE session_task_event SET quiescence_pending = FALSE WHERE sequence = $1
 		`, work.Version)
 	})
 }
+
+// saveSessionTaskSnapshot associates a confirmed runtime snapshot with its
+// captured history version and head task. The caller holds the Session lock and
+// has validated the version under exclusive lifecycle authority.
+func saveSessionTaskSnapshot(ctx context.Context, tx pgx.Tx, historyID uuid.UUID, taskID string, version int64, snapshot *SessionTaskSnapshot) error {
+	if err := execSQL(ctx, tx, `
+		UPDATE session_task SET snapshot_atespace = $3, snapshot_uri = $4,
+		    snapshot_fidelity = $5, history_sequence = $6
+		WHERE history_id = $1 AND id = $2
+	`, historyID, taskID, snapshot.Atespace, snapshot.URI, snapshot.Fidelity, version); err != nil {
+		return err
+	}
+	return execSQL(ctx, tx, `
+		UPDATE session_task_event SET snapshot_atespace = $2, snapshot_uri = $3, snapshot_fidelity = $4
+		WHERE sequence = $1
+	`, version, snapshot.Atespace, snapshot.URI, snapshot.Fidelity)
+}

go/core/internal/database/session_recovery.go

@@ -116,7 +116,10 @@ func (c *Client) BeginCrashedSessionRecovery(ctx context.Context, generation *Se
 		if err != nil {
 			return err
 		}
-		if err := execSQL(ctx, tx, `UPDATE session SET data = $2, dispatch_id = NULL, dispatch_expires_at = NULL WHERE id = $1`, row.ID, row.Data); err != nil {
+		if err := execSQL(ctx, tx, `
+			UPDATE session SET data = $2, dispatch_id = NULL, dispatch_expires_at = NULL, dispatch_cleanup_executor_id = NULL
+			WHERE id = $1
+		`, row.ID, row.Data); err != nil {
 			return err
 		}
 		if err := settleCrashedSessionTasks(ctx, tx, row, snapshot); err != nil {