feat(session): add opt-in crash recovery on send

.github/workflows/ci.yaml

@@ -154,6 +154,8 @@ jobs:
                 port: http
               periodSeconds: 1
           config:
+            sessions:
+              automaticRecovery: true
             push:
               allowHttp: true
               allowPrivateNetworks: true

docs/architecture/process-configuration.md

@@ -114,6 +114,12 @@ runtimeDefaults:
     level: info
 ```
 
+`sessions.automaticRecovery` enables recovery of crashed runtimes during sends.
+It defaults to false. The chart renders it only when set, so a rollback to a
+release that predates the setting can still read the default config. Explicit
+`RevertSession` calls remain available with either value. The setting does not
+change migrations or stored data.
+
 The controller connects with two logins. `database.owner` applies migrations at
 startup; `database.readWrite` serves every runtime query. Each assumes its `role`
 after connecting. Set `database.skipMigrations: true` when migrations are applied

docs/architecture/runtime-and-lifecycle.md

@@ -265,25 +265,82 @@ retires the claim in its transaction, so a late finish records nothing.
 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
-from SUSPENDED. Nothing retries a recovery in the background. An admitted
-recovery survives API restarts as a durable REVERT operation, and the client
-continues it by retrying RevertSession, as with Create and Delete. While it is
-pending, `GetSession` reports FAILED/REVERT with failure reason `RuntimeCrashed`,
-and interactions are refused because a lifecycle operation is in progress.
+from SUSPENDED. An admitted recovery survives API restarts as a durable REVERT
+operation, and the client continues it by retrying RevertSession, as with Create
+and Delete. Nothing retries it in the background. While it is
+pending, `GetSession` reports FAILED/REVERT with failure reason `RuntimeCrashed`.
+
+With `sessions.automaticRecovery` enabled in controller config, an A2A send also
+starts recovery, so chat and API clients recover with the retry they already
+make. Substrate cannot resume a crashed Actor, so a send to one fails before the
+runtime accepts it. When the gateway proves the input was not accepted, it runs
+the same Revert on that Session before answering `KAGENT_SEND_NOT_ACCEPTED`, and
+the client's retry reaches the restored runtime. A crash during a turn leaves
+that turn open, and an open turn refuses every send, so a send refused for an
+open turn also runs Revert once and then tries its reservation again: a crashed
+runtime is restored and the send proceeds, and a busy healthy runtime refuses it
+as before. Other conflicts, such as a checkpoint in progress, do not start
+recovery. A send while recovery is pending joins it.
+
+A worker can also be lost while the Actor suspends after a turn, for example
+during a node drain or scale-down. The Actor is then CRASHED, the Session still
+reads READY, and the idle worker keeps the suspension's claim because the outcome
+is unknown; nothing takes the claim over on a timeout. Every later send waits out
+the 10-second busy period and is told to retry, so the conversation would never
+recover on its own. Reservation reports a held suspension claim as
+`ErrDispatchSuspending`, a kind of `ErrDispatchBusy`. When that claim outlasts
+the whole wait, which a healthy suspension does not, the send attempts the same
+bounded recovery once. Admitting the recovery retires the claim. Revert leaves a
+suspension that is still running alone, and one more reservation, without
+another wait, proceeds or answers retry as before.
+
+A send waits at most 10 seconds on the recovery it starts. A recovery still
+running then stays admitted: the attempt releases its claim, the send is
+answered `KAGENT_SEND_NOT_ACCEPTED` with `state: RECOVERING` and a one-second
+`retryAfterMs`, and the client's retry continues it. The reason is unchanged so
+existing clients keep retrying. Reverting needs Session update access, as
+`RevertSession` does. When the revert cannot run, whether for missing access,
+contention, or an outage, the send keeps the answer it would give without
+recovery, unless a recovery is already admitted, which answers RECOVERING. No
+Session read calls Substrate.
+
+The setting is off by default. Sends then keep the refusals they had before
+recovery existed, and only an explicit `RevertSession` recovers. Chat retries a
+send refused as `KAGENT_SEND_NOT_ACCEPTED` with the same message ID, doubling
+the wait from the gateway's `retryAfterMs` up to two seconds for 30 seconds,
+and then offers Recover, which calls `RevertSession`. A crash during a turn
+leaves that turn open, so with the setting off the next send is refused
+outright while the Session still reads READY. The gateway names that refusal
+`KAGENT_TURN_ACTIVE` in its ErrorInfo metadata, and chat offers Recover for it
+too, because nothing else observes the crash. Other refusals keep a plain error.
+Chat also offers Recover while a recovery is pending, because a recovery whose
+attempt failed stays FAILED/REVERT until a client retries it. Recover returns a
+healthy runtime unchanged and records a lost one, so it is safe to offer on a
+guess.
+
+A missing, replaced, or retemplated Actor is not this Session's runtime, and
+reverting it would restore the wrong one. Whichever Revert observes that,
+directly or while executing an admitted recovery, records it in one transaction
+guarded by the generation it observed: the Session becomes FAILED/NONE with
+failure reason `RuntimeLost`, every open task fails, and a pending dispatch is
+cleared. A newer generation makes the observation stale and leaves the record
+alone. Later sends and reverts answer from PostgreSQL with a non-retryable error
+and do not ask Substrate again. Delete still works.
 
 Delete supersedes an admitted recovery even after an attempt issued the revert.
 Other issued lifecycle work blocks Delete because it may still create something
 Delete would miss; a revert restores the recorded Actor in place and creates
 nothing. Substrate serializes the two calls, and a late revert fails against the
-deleted Actor. A recovery whose Actor is missing or replaced stays pending, and
-each retry returns
-`FAILED_PRECONDITION` so the client deletes the Session.
+deleted Actor. Recording a lost runtime supersedes an admitted recovery the
+same way.
 
 A successful revert returns the Session to READY with the same Actor suspended.
 The next authorized interaction resumes it through Substrate ingress. Runtime
 changes after the retained snapshot are lost; the discarded turn receives no
 checkpoint at that older snapshot. Another revert returns the Session.
 
+Chat shows a pending recovery and a lost runtime from the stored Session state.
+
 ```mermaid
 sequenceDiagram
     participant Client

docs/lifecycle-retries.md

@@ -41,7 +41,10 @@ should likewise retain the `request_id` passed to `create_sandbox`.
 
 ## Recovering a crashed Session
 
-Call `SessionService.RevertSession` with the Session ID to restore its recorded runtime to the last durable snapshot.
+A send recovers a crashed runtime. It returns `KAGENT_SEND_NOT_ACCEPTED` after reverting, so retry the send as usual.
+A send refused because a crashed turn is still active recovers the runtime first and then proceeds.
+A send while recovery is pending joins it. Recovering through a send needs Session update access.
+Call `SessionService.RevertSession` with the Session ID to restore the recorded runtime to its last durable snapshot without sending.
 The caller needs Session update access. The operation verifies Substrate's CRASHED state and the recorded Actor identity.
 Recovery fails interrupted turns and waiting tasks absent from the retained snapshot.
 It preserves durable terminal results and returns the Session to READY with its Actor suspended.
@@ -51,15 +54,17 @@ A runtime without a snapshot restarts from its initial state.
 Retry an interrupted revert against the same Session ID; nothing retries it in the background.
 Pending recovery retains its operation across API restarts, and a retry joins it.
 While recovery is pending, `GetSession` reports state `FAILED`, operation `REVERT`, and failure reason `RuntimeCrashed`.
-Interactions fail because a lifecycle operation is in progress. Call `RevertSession` again to finish the recovery.
+Retry the send, or call `RevertSession` again, to finish the recovery.
 A retry after a revert whose response was lost returns the recovered Session.
 Checkpoint creation returns `ABORTED` until it finishes.
 A revert that observed the crash before another lifecycle operation began also returns `ABORTED`.
 So does a revert while another lifecycle operation is active.
 
 A revert of a runtime that is not crashed returns the Session unchanged.
-A revert that can never succeed returns `FAILED_PRECONDITION`: the recorded Actor is missing, or a same-name Actor replaced it.
-Stop retrying, delete the Session, then start another or fork one from a checkpoint. Delete succeeds even while the recovery is pending.
+When the recorded Actor is missing, or a same-name Actor replaced it, the runtime cannot be recovered.
+The Session is recorded as state `FAILED`, operation `NONE`, failure reason `RuntimeLost`, and its open tasks fail.
+From then on, `RevertSession` returns `FAILED_PRECONDITION` and sends fail with an error that is not retryable.
+Stop retrying, delete the Session, then start another or fork one from a checkpoint.
 Read the reason rather than the message text, which is not stable.
 
 ## Errors

go/core/internal/a2agateway/gateway.go

@@ -24,6 +24,7 @@ type interactionService interface {
 	PrepareCancelTask(context.Context, types.NamespacedName, *a2a.CancelTaskRequest) (*apiv1.Session, *a2a.Task, error)
 	PrepareTaskSubscription(context.Context, types.NamespacedName, *a2a.SubscribeToTaskRequest) (*apiv1.Session, *a2a.Task, error)
 	RevokeSend(context.Context, types.NamespacedName, *a2a.Message, uuid.UUID) (bool, error)
+	RecoverRejectedSend(context.Context, types.NamespacedName, *a2a.Message) error
 	GetTaskByMessage(context.Context, types.NamespacedName, *a2a.Message) (*a2a.Task, error)
 	GetSendResult(context.Context, types.NamespacedName, *a2a.Message, a2a.TaskID, *int) (*a2a.Task, error)
 	GetCancelResult(context.Context, types.NamespacedName, a2a.TaskID) (*a2a.Task, error)

go/core/internal/a2agateway/gateway_test.go

@@ -19,8 +19,10 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/test/bufconn"
 	"google.golang.org/protobuf/proto"
 
@@ -55,6 +57,9 @@ type gatewayTestStore struct {
 	initialID        string
 	listedIDs        []string
 	revoked          bool
+	reverts          int
+	onRevert         func()
+	revertErr        error
 	session          *apiv1.Session
 	revision         *database.RuntimeRevision
 	err              error
@@ -808,6 +813,81 @@ func TestGatewayReportsOnlyRevokedAttemptsAsNotAccepted(t *testing.T) {
 		if runtime.sendCalls != 1 {
 			t.Fatalf("retried a transport failure: %d sends", runtime.sendCalls)
 		}
+		if want := map[bool]int{false: 0, true: 1}[revoked]; store.reverts != want {
+			t.Fatalf("reverts = %d, want %d: only a refused send checks for a crashed runtime", store.reverts, want)
+		}
+	}
+}
+
+func TestGatewaySendJoinsAnAdmittedCrashRecovery(t *testing.T) {
+	session := gatewayTestSession()
+	session.State, session.Operation = apiv1.RuntimeState_RUNTIME_STATE_FAILED, apiv1.RuntimeOperation_RUNTIME_OPERATION_REVERT
+	store := &gatewayTestStore{session: session}
+	runtime := &gatewayTestRuntime{}
+	gateway := newTestGateway(store, &gatewayTestAuthorizer{}, &gatewayTestDialer{client: gatewayTestClient(t, runtime)}, gatewayTestURL)
+	_, err := gateway.SendMessage(gatewayTestContext(), gatewayTestRequest())
+	var protocolError *a2atype.Error
+	if !errors.As(err, &protocolError) || protocolError.ErrorInfo().Value["metadata"].(map[string]string)["reason"] != "KAGENT_SEND_NOT_ACCEPTED" {
+		t.Fatalf("error = %v, want a retryable send", err)
+	}
+	if store.reverts != 1 || runtime.sendCalls != 0 {
+		t.Fatalf("reverts = %d, sends = %d: the send must join the recovery, not reach the runtime", store.reverts, runtime.sendCalls)
+	}
+}
+
+// A crash during a turn leaves it open, so a send refused for an open turn probes
+// the runtime once. Other conflicts, and a probe that cannot run, keep the refusal.
+func TestGatewayBusySendRecoversACrashedTurn(t *testing.T) {
+	for _, test := range []struct {
+		name          string
+		reserveErr    error
+		crashed       bool
+		revertErr     error
+		reverts, sent int
+		reason        string
+	}{
+		{name: "a busy healthy runtime", reserveErr: database.ErrActiveTurn, reverts: 1, reason: "KAGENT_TURN_ACTIVE"},
+		{name: "a crashed turn", reserveErr: database.ErrActiveTurn, crashed: true, reverts: 1, sent: 1},
+		{name: "a revert that Substrate cannot run", reserveErr: database.ErrActiveTurn,
+			revertErr: status.Error(codes.Unavailable, "control plane unavailable"), reverts: 1, reason: "KAGENT_TURN_ACTIVE"},
+		{name: "a checkpoint in progress", reserveErr: database.ErrConflict},
+	} {
+		t.Run("Given "+test.name+", when a send is refused, assert that recovery runs only for an open turn", func(t *testing.T) {
+			store := &gatewayTestStore{session: gatewayTestSession(), reserveErr: test.reserveErr, taskErr: database.ErrNotFound, revertErr: test.revertErr}
+			if test.crashed {
+				store.onRevert = func() { store.reserveErr = nil }
+			}
+			runtime := &gatewayTestRuntime{}
+			gateway := newTestGateway(store, &gatewayTestAuthorizer{}, &gatewayTestDialer{client: gatewayTestClient(t, runtime)}, gatewayTestURL)
+			_, err := gateway.SendMessage(gatewayTestContext(), gatewayTestRequest())
+			require.Equal(t, test.reverts, store.reverts)
+			require.Equal(t, test.sent, runtime.sendCalls)
+			if test.sent == 1 {
+				return
+			}
+			require.ErrorIs(t, err, a2atype.ErrUnsupportedOperation)
+			require.Equal(t, test.reason, errorInfoReason(err), "a refusal that is not a crash must not invite a retry loop")
+		})
+	}
+}
+
+func TestGatewaySendToLostRuntimeFailsFromStoredState(t *testing.T) {
+	session := gatewayTestSession()
+	session.State, session.Operation = apiv1.RuntimeState_RUNTIME_STATE_FAILED, apiv1.RuntimeOperation_RUNTIME_OPERATION_NONE
+	session.Failure = &apiv1.Failure{Reason: database.SessionRuntimeLost, Message: "Runtime cannot be recovered"}
+	store := &gatewayTestStore{session: session}
+	runtime := &gatewayTestRuntime{}
+	gateway := newTestGateway(store, &gatewayTestAuthorizer{}, &gatewayTestDialer{client: gatewayTestClient(t, runtime)}, gatewayTestURL)
+	_, err := gateway.SendMessage(gatewayTestContext(), gatewayTestRequest())
+	var protocolError *a2atype.Error
+	if !errors.As(err, &protocolError) || !errors.Is(err, a2atype.ErrUnsupportedOperation) {
+		t.Fatalf("error = %v, want a non-retryable refusal", err)
+	}
+	if meta, _ := protocolError.ErrorInfo().Value["metadata"].(map[string]string); meta["reason"] == "KAGENT_SEND_NOT_ACCEPTED" {
+		t.Fatalf("a lost runtime must not invite a retry: %v", err)
+	}
+	if store.reverts != 0 || runtime.sendCalls != 0 {
+		t.Fatalf("reverts = %d, sends = %d: a recorded loss is answered from PostgreSQL", store.reverts, runtime.sendCalls)
 	}
 }