runtime/session: recover interrupted actor resumes

docs/architecture/runtime-and-lifecycle.md

@@ -208,21 +208,63 @@ 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. RUNNING and RESUMING actors can be claimed for cleanup. The
+claim atomically rechecks the dispatch generation, absence of active tasks, and
+other lifecycle work. A newer message supersedes unclaimed cleanup; a claimed
+cleanup 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.
+
+After claiming, the worker reads the actor again. If it is RESUMING, the worker
+retries resume before using the existing suspension path. Once resume succeeds
+or suspension starts, that owner only attempts suspension and snapshot recording.
+If another suspension completed before this owner began, it releases its claim
+but retains the expired dispatch so a later ingress resume is still discovered.
+Transient failures use fresh bounded calls and backoff capped at 15 seconds. Claimed retries
+and recovery scans share a queue with four workers, so repeatedly failing actors
+yield to other Sessions. The same owner and completed phases survive requeueing;
+database retries never repeat a completed runtime operation.
+
+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
 claims still block new work, but completed results remain readable. The recorded
 actor UID is checked before lifecycle calls; a same-name replacement cannot be
 adopted implicitly.
 
+Explicit force deletion may supersede dispatch cleanup, including an uncertain
+claim. It atomically enters DELETING and revokes the dispatch, permanently closing
+execution admission. Queued cleanup stops when it observes deletion; an already
+issued resume or suspend cannot recreate the deleted Actor. Ordinary deletion
+waits for cleanup. Checkpoint and claimed task-boundary work still block force
+deletion.
+
 `SessionService.RevertSession` lets an authorized caller restore a confirmed
 crashed runtime. It accepts active and idle Sessions, but verifies Substrate's
 CRASHED state for the recorded Actor UID and pinned template before admitting
@@ -231,8 +273,9 @@ replacements cannot authorize a revert. A runtime that is not crashed, including
 a Session at rest outside READY, has nothing to recover, so RevertSession
 returns the Session unchanged. That is also what a retry sees after a revert
 whose response was lost. A missing or replaced Actor returns
-`FAILED_PRECONDITION`, because no retry restores it. A transport failure returns `UNAVAILABLE`. Only an active
-lifecycle operation returns `ABORTED`.
+`FAILED_PRECONDITION`, because no retry restores it. A transport failure returns
+`UNAVAILABLE`; conflicting lifecycle work or claimed dispatch cleanup returns
+`ABORTED`.
 
 Admission records FAILED/REVERT with a RuntimeCrashed failure under the Session
 lock. REVERT is its own `RuntimeOperation`: only a confirmed crash starts one, and
@@ -258,10 +301,17 @@ Actor UID and returns the Session to READY, so only the generation exposes a
 stale observation. The recorded UID changes only through a lifecycle operation,
 which also replaces the generation.
 
-An issued idle suspension does not exclude recovery. It cannot finish against the
+An issued task-boundary suspension does not exclude recovery. It cannot finish against the
 crashed Actor, and its worker keeps the claim when suspension fails. Admission
 retires the claim in its transaction, so a late finish records nothing.
 
+Claimed dispatch cleanup does exclude recovery because its owner may still retry
+resume and suspension against the same Actor UID. Its owner can release the claim
+after a fresh CRASHED observation only if all mutation outcomes were conclusive
+and it stops issuing calls. A later successful resume does not erase uncertainty
+from an earlier timeout or lost response. Such claims remain fenced; explicit
+force deletion is available when the Session cannot safely be recovered.
+
 Execution reverts the same Actor to its last durable snapshot without replaying
 input. A runtime without a durable snapshot returns to its initial state.
 An interrupted revert continues from REVERTING; a lost response can be reconciled

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

@@ -43,12 +43,14 @@ func beginSessionDeletion(ctx context.Context, tx pgx.Tx, row sessionRow, reason
 			return nil, err
 		}
 	}
-	// Record deletion and revoke dispatch together so late input cannot start a turn.
+	// DELETING closes admission permanently, so force can retire dispatch cleanup
+	// without exposing a newer turn to its in-flight calls. Actor deletion fences
+	// those calls; the Session tombstone prevents this runtime from being reused.
 	operation, err := startSessionOperation(ctx, tx, row, kind)
 	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 {
@@ -58,8 +60,8 @@ func beginSessionDeletion(ctx context.Context, tx pgx.Tx, row sessionRow, reason
 }
 
 // admitForcedSessionDeletion checks work that force must not interrupt:
-// checkpoint snapshot copying and idle suspension already claimed by a worker.
-// It skips turn, dispatch, and runtime-cleanup checks. These are kagent work
+// checkpoint snapshot copying and claimed task-boundary suspension.
+// It skips turn, dispatch, and dispatch-cleanup checks. These are kagent work
 // records, not the Substrate actor state; RESUMING alone does not block deletion.
 // The caller holds the session lock. runtimeOperationNeeded checks other
 // lifecycle conflicts before this helper runs.

go/core/internal/database/session_deletion_test.go

@@ -178,6 +178,64 @@ func TestForceDeletionRetainsIdleLifecycleOwnership(t *testing.T) {
 	require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, current.State)
 }
 
+func TestForceDeletionSupersedesDispatchCleanup(t *testing.T) {
+	// Given a waiting task whose abandoned dispatch has an issued cleanup claim.
+	client, session, task, version := deletionTaskFixture(t, a2a.TaskStateInputRequired)
+	ctx := t.Context()
+	require.NoError(t, client.SettleSessionTask(ctx, session.Id, string(task.ID), version))
+	candidate := abandonedDispatchFixture(t, client, session)
+	work, err := client.ClaimSessionDispatchCleanup(ctx, candidate)
+	require.NoError(t, err)
+	_, err = client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE, false)
+	require.ErrorIs(t, err, ErrFailedPrecondition, "ordinary deletion must still wait for issued cleanup")
+
+	// When force deletion is admitted, it retires cleanup while permanently
+	// closing admission. A stalled cleanup owner cannot keep deletion blocked.
+	operation, err := client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE, true)
+	require.NoError(t, err)
+	require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_DELETING, operation.Instance.State)
+	_, err = client.GetPendingSessionDispatchCleanup(ctx, "")
+	require.ErrorIs(t, err, ErrNotFound)
+	_, err = client.ClaimSessionDispatchCleanup(ctx, candidate)
+	require.ErrorIs(t, err, ErrNotFound)
+	require.ErrorIs(t, client.ReleaseSessionDispatchCleanup(ctx, work), ErrConflict)
+	require.NoError(t, client.FinishSessionDispatchCleanup(ctx, work, dispatchCleanupSnapshot()), "retired cleanup completion must do nothing")
+	require.ErrorIs(t, client.ReserveSessionDispatch(ctx, session.Id, uuid.New(), "late-input"), ErrConflict)
+	for _, token := range []string{work.DispatchID.String(), ""} {
+		_, err = client.UpdateSessionTask(ctx, session.Id, version, taskMutationHash("late update"), task, task, token)
+		require.ErrorIs(t, err, ErrConflict)
+	}
+	row, err := readSession(ctx, client.db, session.Id)
+	require.NoError(t, err)
+	stored, err := readSessionTask(ctx, client.db, row.HistoryID, string(task.ID))
+	require.NoError(t, err)
+	require.Nil(t, stored.SnapshotURI, "late cleanup must not record a snapshot after deletion took ownership")
+	current, err := client.GetSessionOperation(ctx, session.Id, operation.ID)
+	require.NoError(t, err)
+	require.Equal(t, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE, current.Instance.Operation)
+
+	// A failed preparation keeps DELETING and a new deletion generation. Late
+	// cleanup callbacks cannot release its fence or replace that operation.
+	_, err = client.FinishSessionOperation(ctx, session.Id, operation.ID, uuid.Nil, "", "", "preparation failed")
+	require.NoError(t, err)
+	retry, err := client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE, false)
+	require.NoError(t, err)
+	require.NotEqual(t, operation.ID, retry.ID)
+	require.ErrorIs(t, client.ReleaseSessionDispatchCleanup(ctx, work), ErrConflict)
+	require.NoError(t, client.FinishSessionDispatchCleanup(ctx, work, dispatchCleanupSnapshot()))
+	current, err = client.GetSessionOperation(ctx, session.Id, retry.ID)
+	require.NoError(t, err)
+	require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_DELETING, current.Instance.State)
+	require.ErrorIs(t, client.ReserveSessionDispatch(ctx, session.Id, uuid.New(), "late-input"), ErrConflict)
+	finishAdmittedDeletion(t, client, retry)
+	require.ErrorIs(t, client.ReleaseSessionDispatchCleanup(ctx, work), ErrConflict)
+	require.NoError(t, client.FinishSessionDispatchCleanup(ctx, work, dispatchCleanupSnapshot()))
+	deleted, err := client.BeginSessionOperation(ctx, session.Id, apiv1.RuntimeOperation_RUNTIME_OPERATION_DELETE, false)
+	require.NoError(t, err)
+	require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_DELETED, deleted.Instance.State)
+	require.ErrorIs(t, client.ReserveSessionDispatch(ctx, session.Id, uuid.New(), "late-input"), ErrConflict)
+}
+
 func TestDeletionSerializesWithMessageAdmission(t *testing.T) {
 	client, session := deletionSessionFixture(t)
 	start := make(chan struct{})