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

@@ -99,13 +99,18 @@ push:
   signingPrivateKeyFile: secrets/push-private-key.pem
 sessions:
   idleTtl: 604800s
+  automaticRecovery: false
 runtimeDefaults:
   controlPlaneUrl: http://kagent-controller.kagent.svc:8083
   gatewayUrl: http://kagent-controller.kagent.svc:8083
   logging:
     level: info
 ```
 
+`sessions.automaticRecovery` enables recovery of crashed runtimes during sends.
+It defaults to false. 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,47 @@ 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`.
+
+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 active, and active work refuses every
+send, so a send refused for active work 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. A send while recovery is pending
+joins it. Reverting needs Session update access, as `RevertSession` does; a
+caller who may only send gets the original refusal. The revert continues if the
+caller disconnects. No Session read calls Substrate.
+
+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

@@ -55,6 +55,8 @@ type gatewayTestStore struct {
 	initialID        string
 	listedIDs        []string
 	revoked          bool
+	reverts          int
+	onRevert         func()
 	session          *apiv1.Session
 	revision         *database.RuntimeRevision
 	err              error
@@ -808,6 +810,74 @@ 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 active, so a busy send probes the runtime once.
+func TestGatewayBusySendRecoversACrashedTurn(t *testing.T) {
+	for _, crashed := range []bool{false, true} {
+		store := &gatewayTestStore{session: gatewayTestSession(), reserveErr: database.ErrConflict, taskErr: database.ErrNotFound}
+		if crashed {
+			store.onRevert = func() { store.reserveErr = nil }
+		}
+		runtime := &gatewayTestRuntime{}
+		gateway := newTestGateway(store, &gatewayTestAuthorizer{}, &gatewayTestDialer{client: gatewayTestClient(t, runtime)}, gatewayTestURL)
+		_, err := gateway.SendMessage(gatewayTestContext(), gatewayTestRequest())
+		if store.reverts != 1 || store.reserveCalls != 2 {
+			t.Fatalf("crashed = %t: reverts = %d, reservations = %d", crashed, store.reverts, store.reserveCalls)
+		}
+		if crashed {
+			if runtime.sendCalls != 1 {
+				t.Fatalf("a restored runtime must receive the send: %d sends, err = %v", runtime.sendCalls, err)
+			}
+			continue
+		}
+		var protocolError *a2atype.Error
+		if !errors.As(err, &protocolError) || !errors.Is(err, a2atype.ErrUnsupportedOperation) || runtime.sendCalls != 0 {
+			t.Fatalf("a busy healthy runtime must refuse the send as before: %v, %d sends", err, runtime.sendCalls)
+		}
+		if meta, _ := protocolError.ErrorInfo().Value["metadata"].(map[string]string); meta["reason"] == "KAGENT_SEND_NOT_ACCEPTED" {
+			t.Fatalf("a busy healthy runtime must not invite a retry loop: %v", err)
+		}
+	}
+}
+
+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)
 	}
 }
 

go/core/internal/a2agateway/interactions.go

@@ -14,7 +14,6 @@ import (
 
 	apia2a "github.com/kagent-dev/kagent/go/api/a2a"
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
-	sessionsvc "github.com/kagent-dev/kagent/go/core/internal/service/session"
 	"github.com/kagent-dev/kagent/go/pkg/logging"
 )
 
@@ -187,7 +186,7 @@ func (g *Gateway) finishSend(ctx context.Context, agent types.NamespacedName, me
 	if err != nil {
 		logging.FromContext(ctx).ErrorContext(ctx, "release runtime dispatch", "session_id", message.ContextID, "error", err)
 	} else if revoked && sendErr != nil && a2atype.ErrorReason(sendErr) == a2atype.ErrorReason(a2atype.ErrInternalError) {
-		return sessionsvc.ErrSendNotAccepted
+		return g.interactions.RecoverRejectedSend(ctx, agent, message)
 	}
 	return sendErr
 }