diff --git a/docs/architecture/runtime-and-lifecycle.md b/docs/architecture/runtime-and-lifecycle.md
index 1b7b405d46..62fb0133ad 100644
--- a/docs/architecture/runtime-and-lifecycle.md
+++ b/docs/architecture/runtime-and-lifecycle.md
@@ -252,6 +252,23 @@ changes after the retained snapshot are lost; the discarded turn receives no
 checkpoint at that older snapshot. Another revert against the now-healthy Actor
 returns `FAILED_PRECONDITION` with `KAGENT_SESSION_RUNTIME_NOT_CRASHED`.
 
+`GetSession` reports a read-only recovery observation for the recorded Actor.
+It is the only Session read that calls Substrate; MCP and the other Session
+responses read the stored record and omit `recovery`. An unavailable or
+replaced Actor leaves the observation unknown and does not prevent history
+reads. The observation and `RevertSession` share one crash check, so chat offers
+a revert exactly when the revert would be admitted. Chat offers that revert and
+the existing fork action for a saved READY checkpoint.
+
+Recovery records exact task and message identities for user inputs after the
+retained snapshot boundary, including accepted inputs not yet published. Chat
+marks those inputs "Not remembered after recovery" and retains the public
+transcript; completed A2A tasks and their history are not rewound, so a reply
+to a marked input can still appear. The annotations remain on the recovered
+Session. A revert to a snapshot that no history event records leaves the
+boundary unknown and adds no annotations. Forks copy checkpoint-bounded history
+into a fresh Session and do not inherit its delivery annotations.
+
 ```mermaid
 sequenceDiagram
     participant Client
diff --git a/docs/lifecycle-retries.md b/docs/lifecycle-retries.md
index fb59f20355..c4ea91c50c 100644
--- a/docs/lifecycle-retries.md
+++ b/docs/lifecycle-retries.md
@@ -41,12 +41,18 @@ 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.
+Chat offers a revert or a fork from an existing READY checkpoint after a confirmed
+runtime crash. Call `SessionService.RevertSession` with the Session ID to restore
+its recorded runtime to the last durable snapshot.
 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.
 It discards runtime changes since that snapshot and does not replay input.
 A runtime without a snapshot restarts from its initial state.
+`GetSession.recovery.failed_messages` identifies user inputs absent from the
+restored runtime by both task and message ID. Their public history remains
+available, and chat marks them "Not remembered after recovery". Only
+`GetSession` returns `recovery`.
 
 Retry an interrupted revert against the same Session ID. Pending recovery retains its operation across API restarts.
 The lifecycle worker retries already admitted recoveries; it does not initiate recovery for other crashed Sessions.
diff --git a/go/api/gen/kagent/api/v1/sessions.pb.go b/go/api/gen/kagent/api/v1/sessions.pb.go
index 1926bb5626..cf59330e37 100644
--- a/go/api/gen/kagent/api/v1/sessions.pb.go
+++ b/go/api/gen/kagent/api/v1/sessions.pb.go
@@ -93,7 +93,9 @@ type Session struct {
 	Name string `protobuf:"bytes,13,opt,name=name,proto3" json:"name,omitempty"`
 	// Output only. A2A conversation ID, equal to this session's ID.
 	// Forks receive a new conversation ID and new task IDs.
-	ContextId     string `protobuf:"bytes,14,opt,name=context_id,json=contextId,proto3" json:"context_id,omitempty"`
+	ContextId string `protobuf:"bytes,14,opt,name=context_id,json=contextId,proto3" json:"context_id,omitempty"`
+	// Output only. Set only by GetSession; every other response leaves it unset.
+	Recovery      *SessionRecovery `protobuf:"bytes,16,opt,name=recovery,proto3" json:"recovery,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -212,6 +214,124 @@ func (x *Session) GetContextId() string {
 	return ""
 }
 
+func (x *Session) GetRecovery() *SessionRecovery {
+	if x != nil {
+		return x.Recovery
+	}
+	return nil
+}
+
+type SessionRecovery struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Whether the recorded runtime crashed and needs a revert. Unset when
+	// GetSession could not observe the runtime; clients must not offer a revert then.
+	Needed *bool `protobuf:"varint,1,opt,name=needed,proto3,oneof" json:"needed,omitempty"`
+	// User messages the restored runtime does not remember, from every revert of
+	// this Session. Each appears once. Their A2A history remains available,
+	// including completed task results, so the list lasts as long as the
+	// Session and grows only when a revert discards input.
+	FailedMessages []*SessionMessageReference `protobuf:"bytes,3,rep,name=failed_messages,json=failedMessages,proto3" json:"failed_messages,omitempty"`
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
+}
+
+func (x *SessionRecovery) Reset() {
+	*x = SessionRecovery{}
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *SessionRecovery) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SessionRecovery) ProtoMessage() {}
+
+func (x *SessionRecovery) ProtoReflect() protoreflect.Message {
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[2]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SessionRecovery.ProtoReflect.Descriptor instead.
+func (*SessionRecovery) Descriptor() ([]byte, []int) {
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *SessionRecovery) GetNeeded() bool {
+	if x != nil && x.Needed != nil {
+		return *x.Needed
+	}
+	return false
+}
+
+func (x *SessionRecovery) GetFailedMessages() []*SessionMessageReference {
+	if x != nil {
+		return x.FailedMessages
+	}
+	return nil
+}
+
+// Message IDs are unique within a task, not across a conversation.
+type SessionMessageReference struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	TaskId        string                 `protobuf:"bytes,1,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`
+	MessageId     string                 `protobuf:"bytes,2,opt,name=message_id,json=messageId,proto3" json:"message_id,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *SessionMessageReference) Reset() {
+	*x = SessionMessageReference{}
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[3]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *SessionMessageReference) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SessionMessageReference) ProtoMessage() {}
+
+func (x *SessionMessageReference) ProtoReflect() protoreflect.Message {
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[3]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SessionMessageReference.ProtoReflect.Descriptor instead.
+func (*SessionMessageReference) Descriptor() ([]byte, []int) {
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *SessionMessageReference) GetTaskId() string {
+	if x != nil {
+		return x.TaskId
+	}
+	return ""
+}
+
+func (x *SessionMessageReference) GetMessageId() string {
+	if x != nil {
+		return x.MessageId
+	}
+	return ""
+}
+
 type AgentProvenance struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Ref           *ResourceReference     `protobuf:"bytes,1,opt,name=ref,proto3" json:"ref,omitempty"`
@@ -222,7 +342,7 @@ type AgentProvenance struct {
 
 func (x *AgentProvenance) Reset() {
 	*x = AgentProvenance{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[2]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[4]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -234,7 +354,7 @@ func (x *AgentProvenance) String() string {
 func (*AgentProvenance) ProtoMessage() {}
 
 func (x *AgentProvenance) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[2]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -247,7 +367,7 @@ func (x *AgentProvenance) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use AgentProvenance.ProtoReflect.Descriptor instead.
 func (*AgentProvenance) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{2}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *AgentProvenance) GetRef() *ResourceReference {
@@ -276,7 +396,7 @@ type CreateSessionRequest struct {
 
 func (x *CreateSessionRequest) Reset() {
 	*x = CreateSessionRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[3]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[5]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -288,7 +408,7 @@ func (x *CreateSessionRequest) String() string {
 func (*CreateSessionRequest) ProtoMessage() {}
 
 func (x *CreateSessionRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[3]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[5]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -301,7 +421,7 @@ func (x *CreateSessionRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateSessionRequest.ProtoReflect.Descriptor instead.
 func (*CreateSessionRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{3}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{5}
 }
 
 func (x *CreateSessionRequest) GetAgent() *ResourceReference {
@@ -334,7 +454,7 @@ type CreateSessionResponse struct {
 
 func (x *CreateSessionResponse) Reset() {
 	*x = CreateSessionResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[4]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[6]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -346,7 +466,7 @@ func (x *CreateSessionResponse) String() string {
 func (*CreateSessionResponse) ProtoMessage() {}
 
 func (x *CreateSessionResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[4]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[6]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -359,7 +479,7 @@ func (x *CreateSessionResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateSessionResponse.ProtoReflect.Descriptor instead.
 func (*CreateSessionResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{4}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{6}
 }
 
 func (x *CreateSessionResponse) GetSession() *Session {
@@ -378,7 +498,7 @@ type GetSessionRequest struct {
 
 func (x *GetSessionRequest) Reset() {
 	*x = GetSessionRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[5]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[7]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -390,7 +510,7 @@ func (x *GetSessionRequest) String() string {
 func (*GetSessionRequest) ProtoMessage() {}
 
 func (x *GetSessionRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[5]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[7]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -403,7 +523,7 @@ func (x *GetSessionRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetSessionRequest.ProtoReflect.Descriptor instead.
 func (*GetSessionRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{5}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{7}
 }
 
 func (x *GetSessionRequest) GetSessionId() string {
@@ -422,7 +542,7 @@ type GetSessionResponse struct {
 
 func (x *GetSessionResponse) Reset() {
 	*x = GetSessionResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[6]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[8]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -434,7 +554,7 @@ func (x *GetSessionResponse) String() string {
 func (*GetSessionResponse) ProtoMessage() {}
 
 func (x *GetSessionResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[6]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[8]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -447,7 +567,7 @@ func (x *GetSessionResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetSessionResponse.ProtoReflect.Descriptor instead.
 func (*GetSessionResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{6}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{8}
 }
 
 func (x *GetSessionResponse) GetSession() *Session {
@@ -470,7 +590,7 @@ type ListSessionsRequest struct {
 
 func (x *ListSessionsRequest) Reset() {
 	*x = ListSessionsRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[7]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[9]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -482,7 +602,7 @@ func (x *ListSessionsRequest) String() string {
 func (*ListSessionsRequest) ProtoMessage() {}
 
 func (x *ListSessionsRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[7]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[9]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -495,7 +615,7 @@ func (x *ListSessionsRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListSessionsRequest.ProtoReflect.Descriptor instead.
 func (*ListSessionsRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{7}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{9}
 }
 
 func (x *ListSessionsRequest) GetAllCreators() bool {
@@ -529,7 +649,7 @@ type ListSessionsResponse struct {
 
 func (x *ListSessionsResponse) Reset() {
 	*x = ListSessionsResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[8]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[10]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -541,7 +661,7 @@ func (x *ListSessionsResponse) String() string {
 func (*ListSessionsResponse) ProtoMessage() {}
 
 func (x *ListSessionsResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[8]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[10]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -554,7 +674,7 @@ func (x *ListSessionsResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListSessionsResponse.ProtoReflect.Descriptor instead.
 func (*ListSessionsResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{8}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{10}
 }
 
 func (x *ListSessionsResponse) GetSessions() []*Session {
@@ -583,7 +703,7 @@ type UpdateSessionNameRequest struct {
 
 func (x *UpdateSessionNameRequest) Reset() {
 	*x = UpdateSessionNameRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[9]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[11]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -595,7 +715,7 @@ func (x *UpdateSessionNameRequest) String() string {
 func (*UpdateSessionNameRequest) ProtoMessage() {}
 
 func (x *UpdateSessionNameRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[9]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[11]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -608,7 +728,7 @@ func (x *UpdateSessionNameRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use UpdateSessionNameRequest.ProtoReflect.Descriptor instead.
 func (*UpdateSessionNameRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{9}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{11}
 }
 
 func (x *UpdateSessionNameRequest) GetSessionId() string {
@@ -634,7 +754,7 @@ type UpdateSessionNameResponse struct {
 
 func (x *UpdateSessionNameResponse) Reset() {
 	*x = UpdateSessionNameResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[10]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[12]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -646,7 +766,7 @@ func (x *UpdateSessionNameResponse) String() string {
 func (*UpdateSessionNameResponse) ProtoMessage() {}
 
 func (x *UpdateSessionNameResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[10]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[12]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -659,7 +779,7 @@ func (x *UpdateSessionNameResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use UpdateSessionNameResponse.ProtoReflect.Descriptor instead.
 func (*UpdateSessionNameResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{10}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{12}
 }
 
 func (x *UpdateSessionNameResponse) GetSession() *Session {
@@ -678,7 +798,7 @@ type RevertSessionRequest struct {
 
 func (x *RevertSessionRequest) Reset() {
 	*x = RevertSessionRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[11]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[13]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -690,7 +810,7 @@ func (x *RevertSessionRequest) String() string {
 func (*RevertSessionRequest) ProtoMessage() {}
 
 func (x *RevertSessionRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[11]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[13]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -703,7 +823,7 @@ func (x *RevertSessionRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RevertSessionRequest.ProtoReflect.Descriptor instead.
 func (*RevertSessionRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{11}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{13}
 }
 
 func (x *RevertSessionRequest) GetSessionId() string {
@@ -722,7 +842,7 @@ type RevertSessionResponse struct {
 
 func (x *RevertSessionResponse) Reset() {
 	*x = RevertSessionResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[12]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[14]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -734,7 +854,7 @@ func (x *RevertSessionResponse) String() string {
 func (*RevertSessionResponse) ProtoMessage() {}
 
 func (x *RevertSessionResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[12]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[14]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -747,7 +867,7 @@ func (x *RevertSessionResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RevertSessionResponse.ProtoReflect.Descriptor instead.
 func (*RevertSessionResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{12}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{14}
 }
 
 func (x *RevertSessionResponse) GetSession() *Session {
@@ -766,7 +886,7 @@ type DeleteSessionRequest struct {
 
 func (x *DeleteSessionRequest) Reset() {
 	*x = DeleteSessionRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[13]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[15]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -778,7 +898,7 @@ func (x *DeleteSessionRequest) String() string {
 func (*DeleteSessionRequest) ProtoMessage() {}
 
 func (x *DeleteSessionRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[13]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[15]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -791,7 +911,7 @@ func (x *DeleteSessionRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use DeleteSessionRequest.ProtoReflect.Descriptor instead.
 func (*DeleteSessionRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{13}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{15}
 }
 
 func (x *DeleteSessionRequest) GetSessionId() string {
@@ -810,7 +930,7 @@ type DeleteSessionResponse struct {
 
 func (x *DeleteSessionResponse) Reset() {
 	*x = DeleteSessionResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[14]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[16]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -822,7 +942,7 @@ func (x *DeleteSessionResponse) String() string {
 func (*DeleteSessionResponse) ProtoMessage() {}
 
 func (x *DeleteSessionResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[14]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[16]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -835,7 +955,7 @@ func (x *DeleteSessionResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use DeleteSessionResponse.ProtoReflect.Descriptor instead.
 func (*DeleteSessionResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{14}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{16}
 }
 
 func (x *DeleteSessionResponse) GetSession() *Session {
@@ -859,7 +979,7 @@ type SessionShare struct {
 
 func (x *SessionShare) Reset() {
 	*x = SessionShare{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[15]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[17]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -871,7 +991,7 @@ func (x *SessionShare) String() string {
 func (*SessionShare) ProtoMessage() {}
 
 func (x *SessionShare) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[15]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[17]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -884,7 +1004,7 @@ func (x *SessionShare) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use SessionShare.ProtoReflect.Descriptor instead.
 func (*SessionShare) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{15}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{17}
 }
 
 func (x *SessionShare) GetId() string {
@@ -928,7 +1048,7 @@ type CreateSessionShareRequest struct {
 
 func (x *CreateSessionShareRequest) Reset() {
 	*x = CreateSessionShareRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[16]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[18]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -940,7 +1060,7 @@ func (x *CreateSessionShareRequest) String() string {
 func (*CreateSessionShareRequest) ProtoMessage() {}
 
 func (x *CreateSessionShareRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[16]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[18]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -953,7 +1073,7 @@ func (x *CreateSessionShareRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateSessionShareRequest.ProtoReflect.Descriptor instead.
 func (*CreateSessionShareRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{16}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{18}
 }
 
 func (x *CreateSessionShareRequest) GetSessionId() string {
@@ -981,7 +1101,7 @@ type CreateSessionShareResponse struct {
 
 func (x *CreateSessionShareResponse) Reset() {
 	*x = CreateSessionShareResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[17]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[19]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -993,7 +1113,7 @@ func (x *CreateSessionShareResponse) String() string {
 func (*CreateSessionShareResponse) ProtoMessage() {}
 
 func (x *CreateSessionShareResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[17]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[19]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1006,7 +1126,7 @@ func (x *CreateSessionShareResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use CreateSessionShareResponse.ProtoReflect.Descriptor instead.
 func (*CreateSessionShareResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{17}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{19}
 }
 
 func (x *CreateSessionShareResponse) GetShare() *SessionShare {
@@ -1033,7 +1153,7 @@ type ListSessionSharesRequest struct {
 
 func (x *ListSessionSharesRequest) Reset() {
 	*x = ListSessionSharesRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[18]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[20]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1045,7 +1165,7 @@ func (x *ListSessionSharesRequest) String() string {
 func (*ListSessionSharesRequest) ProtoMessage() {}
 
 func (x *ListSessionSharesRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[18]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[20]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1058,7 +1178,7 @@ func (x *ListSessionSharesRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListSessionSharesRequest.ProtoReflect.Descriptor instead.
 func (*ListSessionSharesRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{18}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{20}
 }
 
 func (x *ListSessionSharesRequest) GetSessionId() string {
@@ -1085,7 +1205,7 @@ type ListSessionSharesResponse struct {
 
 func (x *ListSessionSharesResponse) Reset() {
 	*x = ListSessionSharesResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[19]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[21]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1097,7 +1217,7 @@ func (x *ListSessionSharesResponse) String() string {
 func (*ListSessionSharesResponse) ProtoMessage() {}
 
 func (x *ListSessionSharesResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[19]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[21]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1110,7 +1230,7 @@ func (x *ListSessionSharesResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use ListSessionSharesResponse.ProtoReflect.Descriptor instead.
 func (*ListSessionSharesResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{19}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{21}
 }
 
 func (x *ListSessionSharesResponse) GetShares() []*SessionShare {
@@ -1136,7 +1256,7 @@ type RevokeSessionShareRequest struct {
 
 func (x *RevokeSessionShareRequest) Reset() {
 	*x = RevokeSessionShareRequest{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[20]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[22]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1148,7 +1268,7 @@ func (x *RevokeSessionShareRequest) String() string {
 func (*RevokeSessionShareRequest) ProtoMessage() {}
 
 func (x *RevokeSessionShareRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[20]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[22]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1161,7 +1281,7 @@ func (x *RevokeSessionShareRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RevokeSessionShareRequest.ProtoReflect.Descriptor instead.
 func (*RevokeSessionShareRequest) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{20}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{22}
 }
 
 func (x *RevokeSessionShareRequest) GetShareId() string {
@@ -1179,7 +1299,7 @@ type RevokeSessionShareResponse struct {
 
 func (x *RevokeSessionShareResponse) Reset() {
 	*x = RevokeSessionShareResponse{}
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[21]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[23]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1191,7 +1311,7 @@ func (x *RevokeSessionShareResponse) String() string {
 func (*RevokeSessionShareResponse) ProtoMessage() {}
 
 func (x *RevokeSessionShareResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_kagent_api_v1_sessions_proto_msgTypes[21]
+	mi := &file_kagent_api_v1_sessions_proto_msgTypes[23]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1204,7 +1324,7 @@ func (x *RevokeSessionShareResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RevokeSessionShareResponse.ProtoReflect.Descriptor instead.
 func (*RevokeSessionShareResponse) Descriptor() ([]byte, []int) {
-	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{21}
+	return file_kagent_api_v1_sessions_proto_rawDescGZIP(), []int{23}
 }
 
 var File_kagent_api_v1_sessions_proto protoreflect.FileDescriptor
@@ -1214,7 +1334,7 @@ const file_kagent_api_v1_sessions_proto_rawDesc = "" +
 	"\x1ckagent/api/v1/sessions.proto\x12\rkagent.api.v1\x1a\x1bbuf/validate/validate.proto\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1akagent/api/v1/common.proto\x1a\x1bkagent/api/v1/runtime.proto\";\n" +
 	"\aFailure\x12\x16\n" +
 	"\x06reason\x18\x01 \x01(\tR\x06reason\x12\x18\n" +
-	"\amessage\x18\x02 \x01(\tR\amessage\"\x88\x04\n" +
+	"\amessage\x18\x02 \x01(\tR\amessage\"\xc4\x04\n" +
 	"\aSession\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\tR\x02id\x12\x18\n" +
 	"\acreator\x18\x02 \x01(\tR\acreator\x124\n" +
@@ -1231,7 +1351,16 @@ const file_kagent_api_v1_sessions_proto_rawDesc = "" +
 	"updated_at\x18\v \x01(\v2\x1a.google.protobuf.TimestampR\tupdatedAt\x12\x12\n" +
 	"\x04name\x18\r \x01(\tR\x04name\x12\x1d\n" +
 	"\n" +
-	"context_id\x18\x0e \x01(\tR\tcontextId\"W\n" +
+	"context_id\x18\x0e \x01(\tR\tcontextId\x12:\n" +
+	"\brecovery\x18\x10 \x01(\v2\x1e.kagent.api.v1.SessionRecoveryR\brecovery\"\x8a\x01\n" +
+	"\x0fSessionRecovery\x12\x1b\n" +
+	"\x06needed\x18\x01 \x01(\bH\x00R\x06needed\x88\x01\x01\x12O\n" +
+	"\x0ffailed_messages\x18\x03 \x03(\v2&.kagent.api.v1.SessionMessageReferenceR\x0efailedMessagesB\t\n" +
+	"\a_needed\"Q\n" +
+	"\x17SessionMessageReference\x12\x17\n" +
+	"\atask_id\x18\x01 \x01(\tR\x06taskId\x12\x1d\n" +
+	"\n" +
+	"message_id\x18\x02 \x01(\tR\tmessageId\"W\n" +
 	"\x0fAgentProvenance\x122\n" +
 	"\x03ref\x18\x01 \x01(\v2 .kagent.api.v1.ResourceReferenceR\x03ref\x12\x10\n" +
 	"\x03uid\x18\x02 \x01(\tR\x03uid\"\xd4\x01\n" +
@@ -1320,86 +1449,90 @@ func file_kagent_api_v1_sessions_proto_rawDescGZIP() []byte {
 	return file_kagent_api_v1_sessions_proto_rawDescData
 }
 
-var file_kagent_api_v1_sessions_proto_msgTypes = make([]protoimpl.MessageInfo, 22)
+var file_kagent_api_v1_sessions_proto_msgTypes = make([]protoimpl.MessageInfo, 24)
 var file_kagent_api_v1_sessions_proto_goTypes = []any{
 	(*Failure)(nil),                    // 0: kagent.api.v1.Failure
 	(*Session)(nil),                    // 1: kagent.api.v1.Session
-	(*AgentProvenance)(nil),            // 2: kagent.api.v1.AgentProvenance
-	(*CreateSessionRequest)(nil),       // 3: kagent.api.v1.CreateSessionRequest
-	(*CreateSessionResponse)(nil),      // 4: kagent.api.v1.CreateSessionResponse
-	(*GetSessionRequest)(nil),          // 5: kagent.api.v1.GetSessionRequest
-	(*GetSessionResponse)(nil),         // 6: kagent.api.v1.GetSessionResponse
-	(*ListSessionsRequest)(nil),        // 7: kagent.api.v1.ListSessionsRequest
-	(*ListSessionsResponse)(nil),       // 8: kagent.api.v1.ListSessionsResponse
-	(*UpdateSessionNameRequest)(nil),   // 9: kagent.api.v1.UpdateSessionNameRequest
-	(*UpdateSessionNameResponse)(nil),  // 10: kagent.api.v1.UpdateSessionNameResponse
-	(*RevertSessionRequest)(nil),       // 11: kagent.api.v1.RevertSessionRequest
-	(*RevertSessionResponse)(nil),      // 12: kagent.api.v1.RevertSessionResponse
-	(*DeleteSessionRequest)(nil),       // 13: kagent.api.v1.DeleteSessionRequest
-	(*DeleteSessionResponse)(nil),      // 14: kagent.api.v1.DeleteSessionResponse
-	(*SessionShare)(nil),               // 15: kagent.api.v1.SessionShare
-	(*CreateSessionShareRequest)(nil),  // 16: kagent.api.v1.CreateSessionShareRequest
-	(*CreateSessionShareResponse)(nil), // 17: kagent.api.v1.CreateSessionShareResponse
-	(*ListSessionSharesRequest)(nil),   // 18: kagent.api.v1.ListSessionSharesRequest
-	(*ListSessionSharesResponse)(nil),  // 19: kagent.api.v1.ListSessionSharesResponse
-	(*RevokeSessionShareRequest)(nil),  // 20: kagent.api.v1.RevokeSessionShareRequest
-	(*RevokeSessionShareResponse)(nil), // 21: kagent.api.v1.RevokeSessionShareResponse
-	(RuntimeState)(0),                  // 22: kagent.api.v1.RuntimeState
-	(RuntimeOperation)(0),              // 23: kagent.api.v1.RuntimeOperation
-	(*timestamppb.Timestamp)(nil),      // 24: google.protobuf.Timestamp
-	(*ResourceReference)(nil),          // 25: kagent.api.v1.ResourceReference
-	(*PageRequest)(nil),                // 26: kagent.api.v1.PageRequest
-	(*PageResponse)(nil),               // 27: kagent.api.v1.PageResponse
-	(*durationpb.Duration)(nil),        // 28: google.protobuf.Duration
+	(*SessionRecovery)(nil),            // 2: kagent.api.v1.SessionRecovery
+	(*SessionMessageReference)(nil),    // 3: kagent.api.v1.SessionMessageReference
+	(*AgentProvenance)(nil),            // 4: kagent.api.v1.AgentProvenance
+	(*CreateSessionRequest)(nil),       // 5: kagent.api.v1.CreateSessionRequest
+	(*CreateSessionResponse)(nil),      // 6: kagent.api.v1.CreateSessionResponse
+	(*GetSessionRequest)(nil),          // 7: kagent.api.v1.GetSessionRequest
+	(*GetSessionResponse)(nil),         // 8: kagent.api.v1.GetSessionResponse
+	(*ListSessionsRequest)(nil),        // 9: kagent.api.v1.ListSessionsRequest
+	(*ListSessionsResponse)(nil),       // 10: kagent.api.v1.ListSessionsResponse
+	(*UpdateSessionNameRequest)(nil),   // 11: kagent.api.v1.UpdateSessionNameRequest
+	(*UpdateSessionNameResponse)(nil),  // 12: kagent.api.v1.UpdateSessionNameResponse
+	(*RevertSessionRequest)(nil),       // 13: kagent.api.v1.RevertSessionRequest
+	(*RevertSessionResponse)(nil),      // 14: kagent.api.v1.RevertSessionResponse
+	(*DeleteSessionRequest)(nil),       // 15: kagent.api.v1.DeleteSessionRequest
+	(*DeleteSessionResponse)(nil),      // 16: kagent.api.v1.DeleteSessionResponse
+	(*SessionShare)(nil),               // 17: kagent.api.v1.SessionShare
+	(*CreateSessionShareRequest)(nil),  // 18: kagent.api.v1.CreateSessionShareRequest
+	(*CreateSessionShareResponse)(nil), // 19: kagent.api.v1.CreateSessionShareResponse
+	(*ListSessionSharesRequest)(nil),   // 20: kagent.api.v1.ListSessionSharesRequest
+	(*ListSessionSharesResponse)(nil),  // 21: kagent.api.v1.ListSessionSharesResponse
+	(*RevokeSessionShareRequest)(nil),  // 22: kagent.api.v1.RevokeSessionShareRequest
+	(*RevokeSessionShareResponse)(nil), // 23: kagent.api.v1.RevokeSessionShareResponse
+	(RuntimeState)(0),                  // 24: kagent.api.v1.RuntimeState
+	(RuntimeOperation)(0),              // 25: kagent.api.v1.RuntimeOperation
+	(*timestamppb.Timestamp)(nil),      // 26: google.protobuf.Timestamp
+	(*ResourceReference)(nil),          // 27: kagent.api.v1.ResourceReference
+	(*PageRequest)(nil),                // 28: kagent.api.v1.PageRequest
+	(*PageResponse)(nil),               // 29: kagent.api.v1.PageResponse
+	(*durationpb.Duration)(nil),        // 30: google.protobuf.Duration
 }
 var file_kagent_api_v1_sessions_proto_depIdxs = []int32{
-	2,  // 0: kagent.api.v1.Session.agent:type_name -> kagent.api.v1.AgentProvenance
-	22, // 1: kagent.api.v1.Session.state:type_name -> kagent.api.v1.RuntimeState
-	23, // 2: kagent.api.v1.Session.operation:type_name -> kagent.api.v1.RuntimeOperation
+	4,  // 0: kagent.api.v1.Session.agent:type_name -> kagent.api.v1.AgentProvenance
+	24, // 1: kagent.api.v1.Session.state:type_name -> kagent.api.v1.RuntimeState
+	25, // 2: kagent.api.v1.Session.operation:type_name -> kagent.api.v1.RuntimeOperation
 	0,  // 3: kagent.api.v1.Session.failure:type_name -> kagent.api.v1.Failure
-	24, // 4: kagent.api.v1.Session.created_at:type_name -> google.protobuf.Timestamp
-	24, // 5: kagent.api.v1.Session.updated_at:type_name -> google.protobuf.Timestamp
-	25, // 6: kagent.api.v1.AgentProvenance.ref:type_name -> kagent.api.v1.ResourceReference
-	25, // 7: kagent.api.v1.CreateSessionRequest.agent:type_name -> kagent.api.v1.ResourceReference
-	1,  // 8: kagent.api.v1.CreateSessionResponse.session:type_name -> kagent.api.v1.Session
-	1,  // 9: kagent.api.v1.GetSessionResponse.session:type_name -> kagent.api.v1.Session
-	26, // 10: kagent.api.v1.ListSessionsRequest.page:type_name -> kagent.api.v1.PageRequest
-	25, // 11: kagent.api.v1.ListSessionsRequest.agent:type_name -> kagent.api.v1.ResourceReference
-	1,  // 12: kagent.api.v1.ListSessionsResponse.sessions:type_name -> kagent.api.v1.Session
-	27, // 13: kagent.api.v1.ListSessionsResponse.page:type_name -> kagent.api.v1.PageResponse
-	1,  // 14: kagent.api.v1.UpdateSessionNameResponse.session:type_name -> kagent.api.v1.Session
-	1,  // 15: kagent.api.v1.RevertSessionResponse.session:type_name -> kagent.api.v1.Session
-	1,  // 16: kagent.api.v1.DeleteSessionResponse.session:type_name -> kagent.api.v1.Session
-	24, // 17: kagent.api.v1.SessionShare.created_at:type_name -> google.protobuf.Timestamp
-	24, // 18: kagent.api.v1.SessionShare.expires_at:type_name -> google.protobuf.Timestamp
-	28, // 19: kagent.api.v1.CreateSessionShareRequest.ttl:type_name -> google.protobuf.Duration
-	15, // 20: kagent.api.v1.CreateSessionShareResponse.share:type_name -> kagent.api.v1.SessionShare
-	26, // 21: kagent.api.v1.ListSessionSharesRequest.page:type_name -> kagent.api.v1.PageRequest
-	15, // 22: kagent.api.v1.ListSessionSharesResponse.shares:type_name -> kagent.api.v1.SessionShare
-	27, // 23: kagent.api.v1.ListSessionSharesResponse.page:type_name -> kagent.api.v1.PageResponse
-	3,  // 24: kagent.api.v1.SessionService.CreateSession:input_type -> kagent.api.v1.CreateSessionRequest
-	5,  // 25: kagent.api.v1.SessionService.GetSession:input_type -> kagent.api.v1.GetSessionRequest
-	7,  // 26: kagent.api.v1.SessionService.ListSessions:input_type -> kagent.api.v1.ListSessionsRequest
-	9,  // 27: kagent.api.v1.SessionService.UpdateSessionName:input_type -> kagent.api.v1.UpdateSessionNameRequest
-	11, // 28: kagent.api.v1.SessionService.RevertSession:input_type -> kagent.api.v1.RevertSessionRequest
-	13, // 29: kagent.api.v1.SessionService.DeleteSession:input_type -> kagent.api.v1.DeleteSessionRequest
-	16, // 30: kagent.api.v1.SessionService.CreateSessionShare:input_type -> kagent.api.v1.CreateSessionShareRequest
-	18, // 31: kagent.api.v1.SessionService.ListSessionShares:input_type -> kagent.api.v1.ListSessionSharesRequest
-	20, // 32: kagent.api.v1.SessionService.RevokeSessionShare:input_type -> kagent.api.v1.RevokeSessionShareRequest
-	4,  // 33: kagent.api.v1.SessionService.CreateSession:output_type -> kagent.api.v1.CreateSessionResponse
-	6,  // 34: kagent.api.v1.SessionService.GetSession:output_type -> kagent.api.v1.GetSessionResponse
-	8,  // 35: kagent.api.v1.SessionService.ListSessions:output_type -> kagent.api.v1.ListSessionsResponse
-	10, // 36: kagent.api.v1.SessionService.UpdateSessionName:output_type -> kagent.api.v1.UpdateSessionNameResponse
-	12, // 37: kagent.api.v1.SessionService.RevertSession:output_type -> kagent.api.v1.RevertSessionResponse
-	14, // 38: kagent.api.v1.SessionService.DeleteSession:output_type -> kagent.api.v1.DeleteSessionResponse
-	17, // 39: kagent.api.v1.SessionService.CreateSessionShare:output_type -> kagent.api.v1.CreateSessionShareResponse
-	19, // 40: kagent.api.v1.SessionService.ListSessionShares:output_type -> kagent.api.v1.ListSessionSharesResponse
-	21, // 41: kagent.api.v1.SessionService.RevokeSessionShare:output_type -> kagent.api.v1.RevokeSessionShareResponse
-	33, // [33:42] is the sub-list for method output_type
-	24, // [24:33] is the sub-list for method input_type
-	24, // [24:24] is the sub-list for extension type_name
-	24, // [24:24] is the sub-list for extension extendee
-	0,  // [0:24] is the sub-list for field type_name
+	26, // 4: kagent.api.v1.Session.created_at:type_name -> google.protobuf.Timestamp
+	26, // 5: kagent.api.v1.Session.updated_at:type_name -> google.protobuf.Timestamp
+	2,  // 6: kagent.api.v1.Session.recovery:type_name -> kagent.api.v1.SessionRecovery
+	3,  // 7: kagent.api.v1.SessionRecovery.failed_messages:type_name -> kagent.api.v1.SessionMessageReference
+	27, // 8: kagent.api.v1.AgentProvenance.ref:type_name -> kagent.api.v1.ResourceReference
+	27, // 9: kagent.api.v1.CreateSessionRequest.agent:type_name -> kagent.api.v1.ResourceReference
+	1,  // 10: kagent.api.v1.CreateSessionResponse.session:type_name -> kagent.api.v1.Session
+	1,  // 11: kagent.api.v1.GetSessionResponse.session:type_name -> kagent.api.v1.Session
+	28, // 12: kagent.api.v1.ListSessionsRequest.page:type_name -> kagent.api.v1.PageRequest
+	27, // 13: kagent.api.v1.ListSessionsRequest.agent:type_name -> kagent.api.v1.ResourceReference
+	1,  // 14: kagent.api.v1.ListSessionsResponse.sessions:type_name -> kagent.api.v1.Session
+	29, // 15: kagent.api.v1.ListSessionsResponse.page:type_name -> kagent.api.v1.PageResponse
+	1,  // 16: kagent.api.v1.UpdateSessionNameResponse.session:type_name -> kagent.api.v1.Session
+	1,  // 17: kagent.api.v1.RevertSessionResponse.session:type_name -> kagent.api.v1.Session
+	1,  // 18: kagent.api.v1.DeleteSessionResponse.session:type_name -> kagent.api.v1.Session
+	26, // 19: kagent.api.v1.SessionShare.created_at:type_name -> google.protobuf.Timestamp
+	26, // 20: kagent.api.v1.SessionShare.expires_at:type_name -> google.protobuf.Timestamp
+	30, // 21: kagent.api.v1.CreateSessionShareRequest.ttl:type_name -> google.protobuf.Duration
+	17, // 22: kagent.api.v1.CreateSessionShareResponse.share:type_name -> kagent.api.v1.SessionShare
+	28, // 23: kagent.api.v1.ListSessionSharesRequest.page:type_name -> kagent.api.v1.PageRequest
+	17, // 24: kagent.api.v1.ListSessionSharesResponse.shares:type_name -> kagent.api.v1.SessionShare
+	29, // 25: kagent.api.v1.ListSessionSharesResponse.page:type_name -> kagent.api.v1.PageResponse
+	5,  // 26: kagent.api.v1.SessionService.CreateSession:input_type -> kagent.api.v1.CreateSessionRequest
+	7,  // 27: kagent.api.v1.SessionService.GetSession:input_type -> kagent.api.v1.GetSessionRequest
+	9,  // 28: kagent.api.v1.SessionService.ListSessions:input_type -> kagent.api.v1.ListSessionsRequest
+	11, // 29: kagent.api.v1.SessionService.UpdateSessionName:input_type -> kagent.api.v1.UpdateSessionNameRequest
+	13, // 30: kagent.api.v1.SessionService.RevertSession:input_type -> kagent.api.v1.RevertSessionRequest
+	15, // 31: kagent.api.v1.SessionService.DeleteSession:input_type -> kagent.api.v1.DeleteSessionRequest
+	18, // 32: kagent.api.v1.SessionService.CreateSessionShare:input_type -> kagent.api.v1.CreateSessionShareRequest
+	20, // 33: kagent.api.v1.SessionService.ListSessionShares:input_type -> kagent.api.v1.ListSessionSharesRequest
+	22, // 34: kagent.api.v1.SessionService.RevokeSessionShare:input_type -> kagent.api.v1.RevokeSessionShareRequest
+	6,  // 35: kagent.api.v1.SessionService.CreateSession:output_type -> kagent.api.v1.CreateSessionResponse
+	8,  // 36: kagent.api.v1.SessionService.GetSession:output_type -> kagent.api.v1.GetSessionResponse
+	10, // 37: kagent.api.v1.SessionService.ListSessions:output_type -> kagent.api.v1.ListSessionsResponse
+	12, // 38: kagent.api.v1.SessionService.UpdateSessionName:output_type -> kagent.api.v1.UpdateSessionNameResponse
+	14, // 39: kagent.api.v1.SessionService.RevertSession:output_type -> kagent.api.v1.RevertSessionResponse
+	16, // 40: kagent.api.v1.SessionService.DeleteSession:output_type -> kagent.api.v1.DeleteSessionResponse
+	19, // 41: kagent.api.v1.SessionService.CreateSessionShare:output_type -> kagent.api.v1.CreateSessionShareResponse
+	21, // 42: kagent.api.v1.SessionService.ListSessionShares:output_type -> kagent.api.v1.ListSessionSharesResponse
+	23, // 43: kagent.api.v1.SessionService.RevokeSessionShare:output_type -> kagent.api.v1.RevokeSessionShareResponse
+	35, // [35:44] is the sub-list for method output_type
+	26, // [26:35] is the sub-list for method input_type
+	26, // [26:26] is the sub-list for extension type_name
+	26, // [26:26] is the sub-list for extension extendee
+	0,  // [0:26] is the sub-list for field type_name
 }
 
 func init() { file_kagent_api_v1_sessions_proto_init() }
@@ -1409,13 +1542,14 @@ func file_kagent_api_v1_sessions_proto_init() {
 	}
 	file_kagent_api_v1_common_proto_init()
 	file_kagent_api_v1_runtime_proto_init()
+	file_kagent_api_v1_sessions_proto_msgTypes[2].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_kagent_api_v1_sessions_proto_rawDesc), len(file_kagent_api_v1_sessions_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   22,
+			NumMessages:   24,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/go/core/internal/database/session_recovery.go b/go/core/internal/database/session_recovery.go
index 549601d792..f4d79127e5 100644
--- a/go/core/internal/database/session_recovery.go
+++ b/go/core/internal/database/session_recovery.go
@@ -140,6 +140,9 @@ func (c *Client) BeginCrashedSessionRecovery(ctx context.Context, id string, gen
 		if guard.Busy {
 			return ErrConflict
 		}
+		if err := recordDiscardedSessionMessages(ctx, tx, row, session, snapshot); err != nil {
+			return err
+		}
 		row.State = apiv1.RuntimeState_RUNTIME_STATE_FAILED.String()
 		row.runtimeInstanceRow, err = beginRuntimeOperation(ctx, tx, row.runtimeInstanceRow, runtimeKindAgent, apiv1.RuntimeOperation_RUNTIME_OPERATION_RESUME)
 		if err != nil {
@@ -164,6 +167,79 @@ func (c *Client) BeginCrashedSessionRecovery(ctx context.Context, id string, gen
 	return operation, err
 }
 
+// The Session lock keeps annotations atomic with recovery admission and task settlement.
+//
+// The boundary is the newest history event tagged with the snapshot the Actor
+// reverts to. A runtime without a snapshot restarts from its initial state, so
+// every input is discarded. A snapshot that no event records leaves the
+// boundary unknown: a quiescence whose Substrate call succeeded but whose
+// commit did not, or a forked Actor reporting a different URI than its copied
+// boundary. Recovery then records no new annotations rather than marking the
+// whole transcript lost, because most of that history is usually in the
+// snapshot. Annotations from earlier reverts are kept either way.
+func recordDiscardedSessionMessages(ctx context.Context, tx pgx.Tx, row sessionRow, session *apiv1.Session, snapshot SessionTaskSnapshot) error {
+	type messageRow struct {
+		TaskID    string
+		MessageID string
+		Data      []byte
+	}
+	// An empty snapshot URI keeps boundary 0, so every message is discarded. A
+	// URI that no event records yields a NULL boundary, which selects nothing.
+	messages, err := queryMany(ctx, tx, `
+		WITH boundary AS (
+		    SELECT CASE WHEN $2::text = '' THEN 0 ELSE MAX(sequence) END AS sequence
+		    FROM session_task_event
+		    WHERE history_id = $1 AND snapshot_uri = $2
+		      AND snapshot_fidelity = $3 AND snapshot_atespace = $4)
+		SELECT e.task_id, e.message_id, e.data
+		FROM session_task_event e, boundary b
+		WHERE e.history_id = $1 AND e.message_id IS NOT NULL AND e.sequence > b.sequence
+		ORDER BY e.sequence
+	`, pgx.RowToStructByName[messageRow], row.HistoryID, snapshot.URI, snapshot.Fidelity, snapshot.Atespace)
+	if err != nil {
+		return fmt.Errorf("read discarded Session messages: %w", err)
+	}
+	if len(messages) == 0 {
+		return nil
+	}
+	failed := append([]*apiv1.SessionMessageReference(nil), session.GetRecovery().GetFailedMessages()...)
+	for _, row := range messages {
+		event, err := unmarshalSessionTaskEvent(row.Data)
+		if err != nil {
+			return err
+		}
+		message, ok := event.(*a2a.Message)
+		if !ok || message == nil || row.TaskID == "" || row.MessageID == "" || message.ID != row.MessageID || string(message.TaskID) != row.TaskID || message.ContextID != session.ContextId {
+			return fmt.Errorf("discarded Session message has inconsistent identity")
+		}
+		if message.Role == a2a.MessageRoleUser {
+			failed = append(failed, &apiv1.SessionMessageReference{TaskId: row.TaskID, MessageId: row.MessageID})
+		}
+	}
+	if len(failed) == 0 {
+		return nil
+	}
+	type identity struct{ taskID, messageID string }
+	seen := make(map[identity]struct{}, len(failed))
+	unique := make([]*apiv1.SessionMessageReference, 0, len(failed))
+	for _, message := range failed {
+		if message.GetTaskId() == "" || message.GetMessageId() == "" {
+			return fmt.Errorf("discarded Session message requires a task and message identity")
+		}
+		key := identity{message.TaskId, message.MessageId}
+		if _, exists := seen[key]; exists {
+			continue
+		}
+		seen[key] = struct{}{}
+		unique = append(unique, message)
+	}
+	if session.Recovery == nil {
+		session.Recovery = &apiv1.SessionRecovery{}
+	}
+	session.Recovery.FailedMessages = unique
+	return nil
+}
+
 // settleCrashedSessionTask retains captured waits and fails discarded nonterminal work.
 // Callers hold the Session lock; recovery cannot checkpoint discarded state.
 func settleCrashedSessionTask(ctx context.Context, tx pgx.Tx, session sessionRow, snapshot SessionTaskSnapshot) error {
diff --git a/go/core/internal/database/session_recovery_messages_test.go b/go/core/internal/database/session_recovery_messages_test.go
new file mode 100644
index 0000000000..b0be8dd167
--- /dev/null
+++ b/go/core/internal/database/session_recovery_messages_test.go
@@ -0,0 +1,138 @@
+package database
+
+import (
+	"testing"
+
+	"github.com/a2aproject/a2a-go/v2/a2a"
+	a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
+)
+
+func TestCrashedRecoveryRejectsMalformedMessageHistory(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		data         []byte
+		emptyPayload bool
+		mutate       func(*a2apb.Message)
+	}{
+		{name: "invalid protobuf", data: []byte{0xff}},
+		{name: "not a message", emptyPayload: true},
+		{name: "message identity", mutate: func(message *a2apb.Message) {
+			message.MessageId = "different-message"
+		}},
+		{name: "task identity", mutate: func(message *a2apb.Message) {
+			message.TaskId = "different-task"
+		}},
+		{name: "context identity", mutate: func(message *a2apb.Message) {
+			message.ContextId = "different-context"
+		}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			client := NewClient(setupTestDB(t), "public")
+			sessionFixture(t, client, t.Context(), "team-a", "revision", "assistant", "kagent")
+			session, task := waitingTaskFixture(t, client)
+			message := &a2apb.Message{
+				MessageId: task.History[0].ID, TaskId: string(task.ID), ContextId: session.ContextId,
+				Role: a2apb.Role_ROLE_USER,
+			}
+			if test.mutate != nil {
+				test.mutate(message)
+			}
+			data := test.data
+			if data == nil {
+				event := &a2apb.StreamResponse{}
+				if !test.emptyPayload {
+					event.Payload = &a2apb.StreamResponse_Message{Message: message}
+				}
+				var err error
+				data, err = proto.Marshal(event)
+				require.NoError(t, err)
+			}
+			// Ordinary task writes reject these malformed identities before persistence.
+			_, err := client.db.Exec(t.Context(), `
+				UPDATE session_task_event SET data = $2
+				WHERE task_id = $1 AND message_id = $3
+			`, string(task.ID), data, task.History[0].ID)
+			require.NoError(t, err)
+
+			_, err = client.BeginCrashedSessionRecovery(t.Context(), session.Id, lifecycleGeneration(t, client, session.Id), "actor-"+session.Id, SessionTaskSnapshot{})
+			require.Error(t, err)
+			current, err := client.GetSessionByID(t.Context(), session.Id)
+			require.NoError(t, err)
+			require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, current.State)
+			require.Equal(t, apiv1.RuntimeOperation_RUNTIME_OPERATION_NONE, current.Operation)
+			require.Empty(t, current.GetRecovery().GetFailedMessages())
+			require.Nil(t, current.Failure)
+		})
+	}
+}
+
+func TestCrashedRecoveryRollsBackAnnotationsWhenSettlementFails(t *testing.T) {
+	client := NewClient(setupTestDB(t), "public")
+	sessionFixture(t, client, t.Context(), "team-a", "revision", "assistant", "kagent")
+	session, task := waitingTaskFixture(t, client)
+	_, version, err := client.GetVersionedSessionTask(t.Context(), session.Id, string(task.ID))
+	require.NoError(t, err)
+	for _, state := range []a2a.TaskState{a2a.TaskStateWorking, a2a.TaskStateInputRequired} {
+		task.Status.State = state
+		version, err = client.UpdateSessionTask(t.Context(), session.Id, version, taskMutationHash(string(state)), task, task, "")
+		require.NoError(t, err)
+	}
+	require.NoError(t, client.SettleSessionTask(t.Context(), session.Id, string(task.ID), version))
+	stored, err := readSession(t.Context(), client.db, session.Id)
+	require.NoError(t, err)
+	before, err := readSessionTask(t.Context(), client.db, stored.HistoryID, string(task.ID))
+	require.NoError(t, err)
+	// Message history remains valid, so recovery first writes its annotations.
+	// Corrupting the task projection makes settlement fail later in that transaction.
+	_, err = client.db.Exec(t.Context(), `UPDATE session_task SET data = $2 WHERE history_id = $1 AND id = $3`, stored.HistoryID, []byte{0xff}, string(task.ID))
+	require.NoError(t, err)
+	_, err = client.BeginCrashedSessionRecovery(t.Context(), session.Id, lifecycleGeneration(t, client, session.Id), "actor-"+session.Id, SessionTaskSnapshot{})
+	require.Error(t, err)
+	current, err := client.GetSessionByID(t.Context(), session.Id)
+	require.NoError(t, err)
+	require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, current.State)
+	require.Equal(t, apiv1.RuntimeOperation_RUNTIME_OPERATION_NONE, current.Operation)
+	require.Empty(t, current.GetRecovery().GetFailedMessages())
+
+	_, err = client.db.Exec(t.Context(), `UPDATE session_task SET data = $2 WHERE history_id = $1 AND id = $3`, stored.HistoryID, before.Data, string(task.ID))
+	require.NoError(t, err)
+	work, err := client.ClaimSessionQuiescence(t.Context())
+	require.NoError(t, err, "failed recovery retains the original pending runtime boundary")
+	require.Equal(t, string(task.ID), work.TaskID)
+}
+
+// A snapshot no history event records leaves the retained boundary unknown, as
+// after a quiescence whose commit was lost or in a fork whose Actor reports a
+// different URI. Recovery must not then mark the whole transcript as lost.
+func TestCrashedRecoveryWithUnrecordedSnapshotAddsNoAnnotations(t *testing.T) {
+	client := NewClient(setupTestDB(t), "public")
+	sessionFixture(t, client, t.Context(), "team-a", "revision", "assistant", "kagent")
+	session, first := waitingTaskFixture(t, client)
+	actorUID := "actor-" + session.Id
+	// A runtime without a snapshot restarts from its initial state and loses every input.
+	completeCrashedRecovery(t, client, session.Id, actorUID)
+	earlier := []*apiv1.SessionMessageReference{{TaskId: string(first.ID), MessageId: first.History[0].ID}}
+	current, err := client.GetSessionByID(t.Context(), session.Id)
+	require.NoError(t, err)
+	require.True(t, proto.Equal(&apiv1.SessionRecovery{FailedMessages: earlier}, current.GetRecovery()))
+
+	second := newSessionTask(uuid.NewString(), "second")
+	second.ContextID = session.ContextId
+	_, err = client.CreateRuntimeTask(t.Context(), session.Id, taskMutationHash("after the first revert"), second, "")
+	require.NoError(t, err)
+	operation, err := client.BeginCrashedSessionRecovery(t.Context(), session.Id, lifecycleGeneration(t, client, session.Id), actorUID, SessionTaskSnapshot{
+		Atespace: "team-a", URI: "s3://snapshots/never-recorded", Fidelity: "SNAPSHOT_FIDELITY_VOLUMES",
+	})
+	require.NoError(t, err, "an unrecorded snapshot does not block recovery")
+	require.Equal(t, apiv1.RuntimeOperation_RUNTIME_OPERATION_RESUME, operation.Instance.Operation)
+	finishCrashedRecovery(t, client, session.Id, operation, actorUID)
+	current, err = client.GetSessionByID(t.Context(), session.Id)
+	require.NoError(t, err)
+	require.True(t, proto.Equal(&apiv1.SessionRecovery{FailedMessages: earlier}, current.GetRecovery()),
+		"an unknown boundary keeps earlier annotations and adds none: got %v", current.GetRecovery())
+}
diff --git a/go/core/internal/grpcserver/scheduledrun_test.go b/go/core/internal/grpcserver/scheduledrun_test.go
index 1d26566933..7c65949ed3 100644
--- a/go/core/internal/grpcserver/scheduledrun_test.go
+++ b/go/core/internal/grpcserver/scheduledrun_test.go
@@ -200,7 +200,7 @@ func scheduledRunTestServer(t *testing.T) (*database.Client, apiv1.ScheduledRunS
 		Listener: listener, Authenticator: &authimpl.InsecureAuthenticator{},
 		SystemService:       testSystemService(),
 		ScheduledRunService: scheduledrun.NewService(store, kube, &pkgauth.NoopAuthorizer{}),
-		SessionService:      sessionsvc.NewService(store, &pkgauth.NoopAuthorizer{}, nil),
+		SessionService:      sessionsvc.NewService(store, &pkgauth.NoopAuthorizer{}, storedRecoveryWorkflow{}),
 	})
 	require.NoError(t, err)
 	ctx, cancel := context.WithCancel(t.Context())
diff --git a/go/core/internal/grpcserver/session.go b/go/core/internal/grpcserver/session.go
index ddd925a139..747f72a90e 100644
--- a/go/core/internal/grpcserver/session.go
+++ b/go/core/internal/grpcserver/session.go
@@ -21,7 +21,7 @@ func (s *sessionServer) CreateSession(ctx context.Context, request *apiv1.Create
 }
 
 func (s *sessionServer) GetSession(ctx context.Context, request *apiv1.GetSessionRequest) (*apiv1.GetSessionResponse, error) {
-	session, err := s.service.Get(ctx, request.GetSessionId())
+	session, err := s.service.Observe(ctx, request.GetSessionId())
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/core/internal/grpcserver/session_recovery_test.go b/go/core/internal/grpcserver/session_recovery_test.go
index d737b784b9..ee96ce2827 100644
--- a/go/core/internal/grpcserver/session_recovery_test.go
+++ b/go/core/internal/grpcserver/session_recovery_test.go
@@ -130,6 +130,14 @@ func TestRevertSessionThroughGRPC(t *testing.T) {
 					require.NoError(t, store.SettleSessionTask(t.Context(), session.Id, string(task.ID), version))
 				}
 			}
+			observeRecovery := func(needed bool) *apiv1.Session {
+				t.Helper()
+				response, err := client.GetSession(owner, &apiv1.GetSessionRequest{SessionId: session.Id})
+				require.NoError(t, err)
+				require.Equal(t, new(needed), response.Session.GetRecovery().Needed)
+				return response.Session
+			}
+			require.Empty(t, observeRecovery(false).GetRecovery().FailedMessages)
 			request := &apiv1.RevertSessionRequest{SessionId: session.Id}
 			if test.noSnapshot {
 				control.actors[name].Status.Snapshots = nil
@@ -146,6 +154,11 @@ func TestRevertSessionThroughGRPC(t *testing.T) {
 			_, err = client.RevertSession(owner, request)
 			requireRevertRefusal(t, err, "KAGENT_SESSION_RUNTIME_NOT_CRASHED", "healthy runtimes cannot revert")
 			control.actors[name].Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
+			crashed := observeRecovery(true)
+			require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, crashed.State)
+			require.Equal(t, apiv1.RuntimeOperation_RUNTIME_OPERATION_NONE, crashed.Operation)
+			require.Nil(t, crashed.Failure)
+			require.Empty(t, crashed.GetRecovery().FailedMessages)
 			before := proto.CloneOf(control.actors[name])
 			control.actors[name].Metadata.Uid = "replacement-uid"
 			_, err = client.RevertSession(owner, request)
@@ -173,10 +186,20 @@ func TestRevertSessionThroughGRPC(t *testing.T) {
 			failed, err := client.GetSession(owner, &apiv1.GetSessionRequest{SessionId: session.Id})
 			require.NoError(t, err)
 			require.Equal(t, database.SessionRuntimeCrashed, failed.Session.Failure.Reason)
+			require.Equal(t, new(true), failed.Session.GetRecovery().Needed)
+			// No history event records "s3://snapshots/last-good", so only a
+			// runtime without a snapshot has a known boundary: its initial state.
+			var wantFailed []*apiv1.SessionMessageReference
+			if task != nil && test.noSnapshot {
+				wantFailed = []*apiv1.SessionMessageReference{{TaskId: string(task.ID), MessageId: task.History[0].ID}}
+			}
+			require.Equal(t, wantFailed, failed.Session.GetRecovery().FailedMessages)
 			ready, err := client.RevertSession(owner, request)
 			require.NoError(t, err)
 			require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, ready.Session.State)
 			require.Nil(t, ready.Session.Failure)
+			require.Nil(t, ready.Session.Recovery, "only GetSession returns recovery")
+			require.Equal(t, wantFailed, observeRecovery(false).GetRecovery().FailedMessages)
 			// A client whose success response was lost retries and learns that
 			// the runtime is already healthy, not that it cannot be recovered.
 			_, err = client.RevertSession(owner, request)
diff --git a/go/core/internal/grpcserver/session_test.go b/go/core/internal/grpcserver/session_test.go
index 14e426128e..25685649ee 100644
--- a/go/core/internal/grpcserver/session_test.go
+++ b/go/core/internal/grpcserver/session_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/proto"
 
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
 	"github.com/kagent-dev/kagent/go/core/internal/database"
@@ -93,7 +94,7 @@ func TestSessionPermissionsAndOwnershipThroughGeneratedClient(t *testing.T) {
 		{Id: bobID, Creator: "bob", Agent: &apiv1.AgentProvenance{Ref: agent}},
 	}}
 	connection := newConfigurationConnection(t, nil, func(config *Config) {
-		config.SessionService = sessionsvc.NewService(store, sessionAuthorizationPolicy{}, nil)
+		config.SessionService = sessionsvc.NewService(store, sessionAuthorizationPolicy{}, storedRecoveryWorkflow{})
 		config.ShareStore = &testShareStore{sessionShare: &apiv1.SessionShare{SessionId: aliceID}, ownerUserID: "alice"}
 	})
 	client := apiv1.NewSessionServiceClient(connection)
@@ -207,3 +208,30 @@ func TestSessionCreateNamespaceThroughGeneratedClient(t *testing.T) {
 		})
 	}
 }
+
+// storedRecoveryWorkflow stands in for the Actor workflow in transport tests
+// that never reach Substrate. GetSession reports stored recovery annotations
+// with needed unknown; lifecycle calls fail, because no test may reach them.
+type storedRecoveryWorkflow struct{}
+
+var errNoRuntimeWorkflow = errors.New("transport test has no runtime workflow")
+
+func (storedRecoveryWorkflow) Create(context.Context, *apiv1.Session) (*apiv1.Session, error) {
+	return nil, errNoRuntimeWorkflow
+}
+
+func (storedRecoveryWorkflow) Revert(context.Context, *apiv1.Session) (*apiv1.Session, error) {
+	return nil, errNoRuntimeWorkflow
+}
+
+func (storedRecoveryWorkflow) Delete(context.Context, *apiv1.Session) (*apiv1.Session, error) {
+	return nil, errNoRuntimeWorkflow
+}
+
+func (storedRecoveryWorkflow) Recovery(_ context.Context, session *apiv1.Session) (*apiv1.SessionRecovery, error) {
+	recovery := proto.CloneOf(session.GetRecovery())
+	if recovery != nil {
+		recovery.Needed = nil
+	}
+	return recovery, nil
+}
diff --git a/go/core/internal/grpcserver/task_store_test.go b/go/core/internal/grpcserver/task_store_test.go
index 753eaa2931..f69a161337 100644
--- a/go/core/internal/grpcserver/task_store_test.go
+++ b/go/core/internal/grpcserver/task_store_test.go
@@ -151,7 +151,7 @@ func TestRuntimeTaskStoreThroughGRPC(t *testing.T) {
 		Authenticator:    runtimeauth.NewProvider(&authimpl.InsecureAuthenticator{}, verifier, store, issuer.URL),
 		TaskStoreService: tasks,
 		MemoryService:    memoryservice.NewService(memories, policy),
-		SessionService:   sessionsvc.NewService(store, policy, nil),
+		SessionService:   sessionsvc.NewService(store, policy, storedRecoveryWorkflow{}),
 	})
 	require.NoError(t, err)
 	serverCtx, stopServer := context.WithCancel(t.Context())
diff --git a/go/core/internal/mcp/tasks_test.go b/go/core/internal/mcp/tasks_test.go
index a902b7cf98..aa0a1b9412 100644
--- a/go/core/internal/mcp/tasks_test.go
+++ b/go/core/internal/mcp/tasks_test.go
@@ -564,6 +564,12 @@ func (*fakeSessionStore) DeleteSessionShare(context.Context, string, string) err
 
 type fakeSessionWorkflow struct{}
 
+// Recovery panics because MCP resolves a Session before every invoke and task
+// call; observing the Actor there would add a Substrate round trip to each one.
+func (*fakeSessionWorkflow) Recovery(context.Context, *apiv1.Session) (*apiv1.SessionRecovery, error) {
+	panic("MCP must not observe Session recovery")
+}
+
 func (*fakeSessionWorkflow) Create(_ context.Context, session *apiv1.Session) (*apiv1.Session, error) {
 	return session, nil
 }
diff --git a/go/core/internal/service/session/recovery.go b/go/core/internal/service/session/recovery.go
index 4ef05bac34..c7fb718991 100644
--- a/go/core/internal/service/session/recovery.go
+++ b/go/core/internal/service/session/recovery.go
@@ -4,11 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/google/uuid"
 	ateapipb "go.ate.dev/substrate/api/ateapi/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
 	"github.com/kagent-dev/kagent/go/core/internal/database"
@@ -29,6 +31,33 @@ var (
 	errRuntimeUnrecoverable = fmt.Errorf("runtime cannot be recovered: %w", database.ErrFailedPrecondition)
 )
 
+// Recovery reports whether the Session's runtime needs a revert, from the same
+// checks Revert admits one by. A failed observation leaves needed unknown, so a
+// control-plane outage can neither offer recovery nor prevent history reads.
+func (w *ActorWorkflow) Recovery(ctx context.Context, session *apiv1.Session) (*apiv1.SessionRecovery, error) {
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	recovery := proto.CloneOf(session.GetRecovery())
+	if recovery == nil {
+		recovery = &apiv1.SessionRecovery{}
+	}
+	recovery.Needed = nil
+	if crashRecoveryAdmitted(session) {
+		recovery.Needed = new(true)
+		return recovery, nil
+	}
+	if !lifecycleIdle(session) {
+		recovery.Needed = new(false)
+		return recovery, nil
+	}
+	observed, err := w.observeCrash(ctx, session)
+	if err != nil {
+		return recovery, fmt.Errorf("observe Session recovery: %w", err)
+	}
+	recovery.Needed = new(observed.crashed())
+	return recovery, nil
+}
+
 // crashRecoveryAdmitted reports a revert already admitted for a confirmed crash.
 // Revert continues it without observing Substrate again.
 func crashRecoveryAdmitted(session *apiv1.Session) bool {
@@ -53,7 +82,9 @@ func (o crashObservation) crashed() bool {
 }
 
 // observeCrash reads the Session's recorded Actor and reports whether Substrate
-// confirms it crashed. Revert admits recovery only through this check. An Actor
+// confirms it crashed. Recovery offers a revert and Revert admits one through
+// this one check, so chat cannot offer a revert that Revert refuses, nor hide
+// one it would accept. An Actor
 // that is missing, replaced, or from another template is an error, never a
 // crash: reverting it would restore the wrong runtime. Those errors wrap
 // errRuntimeUnrecoverable; transient read failures do not.
diff --git a/go/core/internal/service/session/recovery_messages_test.go b/go/core/internal/service/session/recovery_messages_test.go
new file mode 100644
index 0000000000..3a9f8bc0f5
--- /dev/null
+++ b/go/core/internal/service/session/recovery_messages_test.go
@@ -0,0 +1,147 @@
+package session
+
+import (
+	"crypto/sha256"
+	"testing"
+	"time"
+
+	"github.com/a2aproject/a2a-go/v2/a2a"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	ateapipb "go.ate.dev/substrate/api/ateapi/v1"
+
+	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
+	"github.com/kagent-dev/kagent/go/core/internal/database"
+	"github.com/kagent-dev/kagent/go/core/internal/substrate"
+)
+
+func TestCrashedRecoveryMarksOnlyDiscardedMessagesInOneTask(t *testing.T) {
+	store, actors, session, task, version := crashedTurnFixture(t)
+	task.Status.State = a2a.TaskStateInputRequired
+	hash := sha256.Sum256([]byte("captured question"))
+	version, err := store.UpdateSessionTask(t.Context(), session.Id, version, hash[:], task, task, "")
+	require.NoError(t, err)
+	require.NoError(t, store.SettleSessionTask(t.Context(), session.Id, string(task.ID), version))
+	captureRecoveryMessageBoundary(t, store, actors, session)
+
+	answer := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("continue"))
+	answer.TaskID, answer.ContextID = task.ID, session.ContextId
+	reply := a2a.NewMessage(a2a.MessageRoleAgent, a2a.NewTextPart("done"))
+	reply.TaskID, reply.ContextID = task.ID, session.ContextId
+	// Repeated history must not move a captured input past the snapshot boundary.
+	task.History = append(task.History, answer, reply)
+	task.Status.State = a2a.TaskStateCompleted
+	hash = sha256.Sum256([]byte("unpublished completed answer"))
+	_, err = store.UpdateSessionTask(t.Context(), session.Id, version, hash[:], task, task, "")
+	require.NoError(t, err)
+	before, _, err := store.GetVersionedSessionTask(t.Context(), session.Id, string(task.ID))
+	require.NoError(t, err)
+
+	recovered, err := NewActorWorkflow(store, actors, make(chan struct{}, 1), time.Minute).Revert(t.Context(), session)
+	require.NoError(t, err)
+	require.Equal(t, [][2]string{{string(task.ID), answer.ID}}, recoveryMessageIdentities(recovered))
+	stored, err := store.GetSettledSessionTask(t.Context(), session.Id, string(task.ID), nil)
+	require.NoError(t, err)
+	require.Equal(t, a2a.TaskStateCompleted, stored.Status.State)
+	require.Equal(t, before.Status, stored.Status)
+	require.Equal(t, before.History, stored.History, "publishing recovery retains the completed turn's full history")
+	require.Equal(t, before.Artifacts, stored.Artifacts)
+	current, err := store.GetSessionByID(t.Context(), session.Id)
+	require.NoError(t, err)
+	require.Equal(t, recoveryMessageIdentities(recovered), recoveryMessageIdentities(current), "delivery failures survive a fresh Session read")
+}
+
+func TestCrashedRecoveryScopesMessageFailuresToTheirTask(t *testing.T) {
+	store, actors, session, retained, version := crashedTurnFixture(t)
+	retained.Status.State = a2a.TaskStateCompleted
+	hash := sha256.Sum256([]byte("retained completed turn"))
+	version, err := store.UpdateSessionTask(t.Context(), session.Id, version, hash[:], retained, retained, "")
+	require.NoError(t, err)
+	require.NoError(t, store.SettleSessionTask(t.Context(), session.Id, string(retained.ID), version))
+	captureRecoveryMessageBoundary(t, store, actors, session)
+	beforeRetained, err := store.GetSettledSessionTask(t.Context(), session.Id, string(retained.ID), nil)
+	require.NoError(t, err)
+
+	message := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("a different turn"))
+	message.ID, message.ContextID = retained.History[0].ID, session.ContextId
+	discarded := a2a.NewSubmittedTask(message, message)
+	discarded.Status.State = a2a.TaskStateCompleted
+	hash = sha256.Sum256([]byte("completed without a newer snapshot"))
+	version, err = store.CreateRuntimeTask(t.Context(), session.Id, hash[:], discarded, "")
+	require.NoError(t, err)
+	require.NoError(t, store.SettleSessionTask(t.Context(), session.Id, string(discarded.ID), version))
+	beforeDiscarded, err := store.GetSettledSessionTask(t.Context(), session.Id, string(discarded.ID), nil)
+	require.NoError(t, err)
+
+	workflow := NewActorWorkflow(store, actors, make(chan struct{}, 1), time.Minute)
+	recovered, err := workflow.Revert(t.Context(), session)
+	require.NoError(t, err)
+	want := [][2]string{{string(discarded.ID), message.ID}}
+	require.Equal(t, want, recoveryMessageIdentities(recovered))
+	for _, before := range []*a2a.Task{beforeRetained, beforeDiscarded} {
+		after, err := store.GetSettledSessionTask(t.Context(), session.Id, string(before.ID), nil)
+		require.NoError(t, err)
+		require.Equal(t, before, after, "recovery does not rewrite completed A2A tasks or their history")
+	}
+
+	// Reverting the same runtime interval must not duplicate delivery failures.
+	actors.actors[actorKey("team-a", substrate.ActorName(session.Id))].Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
+	recovered, err = workflow.Revert(t.Context(), recovered)
+	require.NoError(t, err)
+	require.Equal(t, want, recoveryMessageIdentities(recovered))
+
+	message = a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("a retained new turn"))
+	message.ContextID = session.ContextId
+	later := a2a.NewSubmittedTask(message, message)
+	later.Status.State = a2a.TaskStateCompleted
+	hash = sha256.Sum256([]byte("completed after recovery"))
+	version, err = store.CreateRuntimeTask(t.Context(), session.Id, hash[:], later, "")
+	require.NoError(t, err)
+	require.NoError(t, store.SettleSessionTask(t.Context(), session.Id, string(later.ID), version))
+	captureRecoveryMessageBoundary(t, store, actors, session)
+
+	message = a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("another lost input"))
+	message.ContextID = session.ContextId
+	lost := a2a.NewSubmittedTask(message, message)
+	hash = sha256.Sum256([]byte("new interrupted turn"))
+	_, err = store.CreateRuntimeTask(t.Context(), session.Id, hash[:], lost, "")
+	require.NoError(t, err)
+	actors.actors[actorKey("team-a", substrate.ActorName(session.Id))].Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
+	recovered, err = workflow.Revert(t.Context(), recovered)
+	require.NoError(t, err)
+	want = append(want, [2]string{string(lost.ID), message.ID})
+	require.Equal(t, want, recoveryMessageIdentities(recovered), "a newer rollback boundary does not erase earlier delivery failures")
+}
+
+func TestCrashedRecoveryWithoutSnapshotMarksAllInputs(t *testing.T) {
+	store, actors, session, task, _ := crashedTurnFixture(t)
+	recovered, err := NewActorWorkflow(store, actors, make(chan struct{}, 1), time.Minute).Revert(t.Context(), session)
+	require.NoError(t, err)
+	require.Equal(t, [][2]string{{string(task.ID), task.History[0].ID}}, recoveryMessageIdentities(recovered))
+}
+
+func captureRecoveryMessageBoundary(t *testing.T, store *lifecycleTestStore, actors *lifecycleTestActors, session *apiv1.Session) {
+	t.Helper()
+	work, err := store.ClaimSessionQuiescence(t.Context())
+	require.NoError(t, err)
+	require.Equal(t, session.Id, work.Session.Id)
+	snapshotURI := "s3://snapshots/" + uuid.NewString()
+	require.NoError(t, store.FinishSessionQuiescence(t.Context(), work, &database.SessionTaskSnapshot{
+		Atespace: "team-a", URI: snapshotURI, Fidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES.String(),
+	}))
+	actors.actors[actorKey("team-a", substrate.ActorName(session.Id))].Status.Snapshots = []*ateapipb.Snapshot{{
+		Generation: 1, Fidelity: ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_VOLUMES,
+		DurableSnapshot: &ateapipb.SnapshotStorage{
+			State:  ateapipb.SnapshotStorageState_SNAPSHOT_STORAGE_STATE_COMPLETED,
+			Object: &ateapipb.ObjectSnapshot{SnapshotUri: snapshotURI},
+		},
+	}}
+}
+
+func recoveryMessageIdentities(session *apiv1.Session) [][2]string {
+	result := make([][2]string, 0, len(session.GetRecovery().GetFailedMessages()))
+	for _, message := range session.GetRecovery().GetFailedMessages() {
+		result = append(result, [2]string{message.TaskId, message.MessageId})
+	}
+	return result
+}
diff --git a/go/core/internal/service/session/recovery_observation_test.go b/go/core/internal/service/session/recovery_observation_test.go
new file mode 100644
index 0000000000..5ef5601c2a
--- /dev/null
+++ b/go/core/internal/service/session/recovery_observation_test.go
@@ -0,0 +1,152 @@
+package session
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	ateapipb "go.ate.dev/substrate/api/ateapi/v1"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+
+	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
+	"github.com/kagent-dev/kagent/go/core/internal/database"
+	"github.com/kagent-dev/kagent/go/core/internal/substrate"
+)
+
+func TestSessionRecoveryObservation(t *testing.T) {
+	for _, test := range []struct {
+		name                       string
+		crashed, replaced, missing bool
+		readErr                    error
+		want                       *bool
+	}{
+		{name: "healthy", want: new(false)},
+		{name: "crashed", crashed: true, want: new(true)},
+		{name: "missing", missing: true},
+		{name: "replaced", crashed: true, replaced: true},
+		{name: "unavailable", readErr: status.Error(codes.Unavailable, "control plane unavailable")},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			store, session := lifecycleFixture(t)
+			base := &lifecycleTestActors{actors: map[string]*ateapipb.Actor{}}
+			actors := &recoveryTestActors{lifecycleTestActors: base, revert: base.RevertActor}
+			workflow := NewActorWorkflow(store, actors, make(chan struct{}, 1), time.Minute)
+			session, err := workflow.Create(t.Context(), session)
+			require.NoError(t, err)
+			key := actorKey("team-a", substrate.ActorName(session.Id))
+			if test.crashed {
+				base.actors[key].Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
+			}
+			if test.replaced {
+				base.actors[key].Metadata.Uid = "replacement"
+			}
+			if test.missing {
+				delete(base.actors, key)
+			}
+			actors.readErr = test.readErr
+			session.Recovery = &apiv1.SessionRecovery{FailedMessages: []*apiv1.SessionMessageReference{{TaskId: "task", MessageId: "message"}}}
+			before := proto.CloneOf(session)
+			observed, err := workflow.Recovery(t.Context(), session)
+			if test.want == nil {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, test.want, observed.Needed)
+			require.Equal(t, session.Recovery.FailedMessages, observed.FailedMessages)
+			require.True(t, proto.Equal(before, session), "observing recovery must not mutate the Session or its history")
+			stored, err := store.GetSessionByID(t.Context(), session.Id)
+			require.NoError(t, err)
+			require.Equal(t, apiv1.RuntimeState_RUNTIME_STATE_READY, stored.State)
+			require.Nil(t, stored.Recovery, "reads do not admit recovery or persist observations")
+
+			// What chat offers is what Revert does: needed admits a revert,
+			// healthy refuses one, and an unknown observation admits nothing.
+			// A missing or replaced Actor is refused for good, so a client stops
+			// retrying; only a failed read stays retryable.
+			_, err = workflow.Revert(t.Context(), session)
+			switch {
+			case test.missing || test.replaced:
+				require.ErrorIs(t, err, errRuntimeUnrecoverable)
+			case test.want == nil:
+				require.Error(t, err)
+				require.NotErrorIs(t, err, database.ErrFailedPrecondition)
+			case *test.want:
+				require.NoError(t, err)
+			default:
+				require.ErrorIs(t, err, errRuntimeNotCrashed)
+			}
+		})
+	}
+}
+
+func TestSessionObservePreservesHistoryWhenRecoveryObservationFails(t *testing.T) {
+	id := uuid.NewString()
+	session := &apiv1.Session{Id: id, State: apiv1.RuntimeState_RUNTIME_STATE_READY}
+	observation := &apiv1.SessionRecovery{FailedMessages: []*apiv1.SessionMessageReference{{TaskId: "task", MessageId: "message"}}}
+	service := NewService(&serviceTestStore{getResult: session}, serviceTestAuthorizer{}, serviceTestWorkflow{
+		recovery: observation, recoveryErr: status.Error(codes.Unavailable, "control plane unavailable"),
+	})
+	got, err := service.Observe(serviceTestContext("alice"), id)
+	require.NoError(t, err)
+	require.Equal(t, id, got.Id)
+	require.Equal(t, observation.FailedMessages, got.Recovery.FailedMessages)
+	require.Nil(t, got.Recovery.Needed, "an unavailable Actor must not be reported healthy")
+	require.Nil(t, session.Recovery, "the stored record remains unchanged")
+}
+
+// panicRecoveryWorkflow fails a test that observes Substrate outside GetSession.
+type panicRecoveryWorkflow struct{ serviceTestWorkflow }
+
+func (panicRecoveryWorkflow) Recovery(context.Context, *apiv1.Session) (*apiv1.SessionRecovery, error) {
+	panic("only GetSession observes recovery")
+}
+
+// Only GetSession observes Substrate or returns recovery. MCP resolves every
+// invoke through Get, and the annotations grow with each lossy revert, so any
+// other response carrying them would cost a control-plane round trip or ship
+// every conversation's annotations in a list.
+func TestSessionRecoveryOnlyInGetSession(t *testing.T) {
+	id := uuid.NewString()
+	annotations := []*apiv1.SessionMessageReference{{TaskId: "task", MessageId: "message"}}
+	stored := func() *apiv1.Session {
+		return &apiv1.Session{
+			Id: id, State: apiv1.RuntimeState_RUNTIME_STATE_READY,
+			Recovery: &apiv1.SessionRecovery{FailedMessages: annotations},
+		}
+	}
+	ctx := serviceTestContext("alice")
+	store := &serviceTestStore{getResult: stored(), sessions: []*apiv1.Session{stored()}}
+	service := NewService(store, serviceTestAuthorizer{}, panicRecoveryWorkflow{})
+
+	got, err := service.Get(ctx, id)
+	require.NoError(t, err)
+	require.Equal(t, annotations, got.GetRecovery().GetFailedMessages(), "Get serves the stored record unchanged")
+
+	listed, err := service.List(ctx, ListRequest{})
+	require.NoError(t, err)
+	require.Len(t, listed.Sessions, 1)
+	require.Nil(t, listed.Sessions[0].Recovery)
+	require.NotNil(t, store.sessions[0].Recovery, "omitting recovery does not mutate the stored record")
+
+	reverted, err := service.Revert(ctx, id)
+	require.NoError(t, err)
+	require.Nil(t, reverted.Recovery)
+	deleted, err := service.Delete(ctx, id)
+	require.NoError(t, err)
+	require.Nil(t, deleted.Recovery)
+
+	// Observing reports the workflow's result without changing the stored record.
+	store.getResult = stored()
+	store.getResult.Recovery.Needed = new(true)
+	observation := &apiv1.SessionRecovery{FailedMessages: annotations, Needed: new(false)}
+	observed, err := NewService(store, serviceTestAuthorizer{}, serviceTestWorkflow{recovery: observation}).Observe(ctx, id)
+	require.NoError(t, err)
+	require.Equal(t, new(false), observed.Recovery.Needed)
+	require.Equal(t, annotations, observed.Recovery.FailedMessages)
+	require.Equal(t, new(true), store.getResult.Recovery.Needed, "observing does not mutate the stored record")
+}
diff --git a/go/core/internal/service/session/service.go b/go/core/internal/service/session/service.go
index bb5cd1eae5..8f4b563737 100644
--- a/go/core/internal/service/session/service.go
+++ b/go/core/internal/service/session/service.go
@@ -22,6 +22,7 @@ import (
 	"github.com/kagent-dev/kagent/go/core/internal/database"
 	"github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
+	"github.com/kagent-dev/kagent/go/pkg/logging"
 )
 
 const (
@@ -43,6 +44,7 @@ type store interface {
 type sessionWorkflow interface {
 	Create(context.Context, *apiv1.Session) (*apiv1.Session, error)
 	Revert(context.Context, *apiv1.Session) (*apiv1.Session, error)
+	Recovery(context.Context, *apiv1.Session) (*apiv1.SessionRecovery, error)
 	Delete(context.Context, *apiv1.Session) (*apiv1.Session, error)
 }
 
@@ -142,10 +144,43 @@ func (s *Service) Create(ctx context.Context, agent *apiv1.ResourceReference, re
 	return session, nil
 }
 
+// Get reads an authorized Session as stored. MCP resolves a Session's Agent
+// through it before every invoke and task call, so it must not wait on Substrate.
 func (s *Service) Get(ctx context.Context, id string) (*apiv1.Session, error) {
 	return s.getAuthorized(ctx, id, auth.VerbGet)
 }
 
+// Observe reads an authorized Session with its recovery observation. It is the
+// only Session read that calls Substrate, and GetSession is its only caller: a
+// slow control plane then delays opening one conversation, not every MCP call.
+// A failed observation leaves needed unset rather than failing the read, so an
+// unavailable control plane cannot hide the history it would let a user recover.
+func (s *Service) Observe(ctx context.Context, id string) (*apiv1.Session, error) {
+	session, err := s.getAuthorized(ctx, id, auth.VerbGet)
+	if err != nil {
+		return nil, err
+	}
+	session = proto.CloneOf(session)
+	session.Recovery, err = s.workflow.Recovery(ctx, session)
+	if err != nil {
+		logging.FromContext(ctx).WarnContext(ctx, "observe session recovery", "session_id", session.Id, "error", err)
+	}
+	return session, nil
+}
+
+// withoutRecovery drops recovery from every response but GetSession. The
+// annotations live in the Session record and grow with each revert that
+// discards input, so a list response would otherwise carry them for every
+// conversation, and every other response would report needed as unknown.
+func withoutRecovery(session *apiv1.Session) *apiv1.Session {
+	if session.GetRecovery() == nil {
+		return session
+	}
+	session = proto.CloneOf(session)
+	session.Recovery = nil
+	return session
+}
+
 // getAuthorized loads a Session authorized for the requested operation. Reads are
 // independent of runtime readiness: viewing a suspended conversation must not
 // provision a worker. Lifecycle and A2A operations use this same access policy.
@@ -196,7 +231,7 @@ func (s *Service) Rename(ctx context.Context, id, name string) (*apiv1.Session,
 	if err != nil {
 		return nil, serviceerrors.NewInternal("Failed to rename Session", err)
 	}
-	return session, nil
+	return withoutRecovery(session), nil
 }
 
 func (s *Service) List(ctx context.Context, request ListRequest) (ListResult, error) {
@@ -220,7 +255,7 @@ func (s *Service) List(ctx context.Context, request ListRequest) (ListResult, er
 		result := ListResult{Sessions: []*apiv1.Session{}}
 		if session.Id > afterID && (request.Agent == nil ||
 			proto.Equal(session.GetAgent().GetRef(), request.Agent)) {
-			result.Sessions = append(result.Sessions, session)
+			result.Sessions = append(result.Sessions, withoutRecovery(session))
 		}
 		return result, nil
 	}
@@ -253,7 +288,7 @@ func (s *Service) List(ctx context.Context, request ListRequest) (ListResult, er
 				}
 				return ListResult{}, err
 			}
-			result.Sessions = append(result.Sessions, session)
+			result.Sessions = append(result.Sessions, withoutRecovery(session))
 			if len(result.Sessions) > pageSize {
 				result.Sessions = result.Sessions[:pageSize]
 				result.NextPageToken = encodePageToken(result.Sessions[pageSize-1].Id)
@@ -281,7 +316,7 @@ func (s *Service) Delete(ctx context.Context, id string) (*apiv1.Session, error)
 	if err != nil {
 		return nil, serviceerrors.NewUnavailable("Failed to delete Session", err)
 	}
-	return session, nil
+	return withoutRecovery(session), nil
 }
 
 func (s *Service) Revert(ctx context.Context, id string) (*apiv1.Session, error) {
@@ -309,7 +344,7 @@ func (s *Service) Revert(ctx context.Context, id string) (*apiv1.Session, error)
 	if err != nil {
 		return nil, serviceerrors.NewUnavailable("Failed to revert Session", err)
 	}
-	return session, nil
+	return withoutRecovery(session), nil
 }
 
 // revertRefusalReason names the ErrorInfo reason for a revert that can never
diff --git a/go/core/internal/service/session/service_test.go b/go/core/internal/service/session/service_test.go
index f5b6270ce5..fc341d2b99 100644
--- a/go/core/internal/service/session/service_test.go
+++ b/go/core/internal/service/session/service_test.go
@@ -126,7 +126,15 @@ func (s *serviceTestStore) DeleteSessionShare(_ context.Context, id, _ string) e
 	return nil
 }
 
-type serviceTestWorkflow struct{ err error }
+type serviceTestWorkflow struct {
+	err         error
+	recovery    *apiv1.SessionRecovery
+	recoveryErr error
+}
+
+func (w serviceTestWorkflow) Recovery(context.Context, *apiv1.Session) (*apiv1.SessionRecovery, error) {
+	return w.recovery, w.recoveryErr
+}
 
 func (w serviceTestWorkflow) Create(_ context.Context, session *apiv1.Session) (*apiv1.Session, error) {
 	return session, w.err
@@ -537,7 +545,7 @@ func TestServiceGetScopesReadsToAuthorizedIdentity(t *testing.T) {
 			stored := &apiv1.Session{Id: id, State: apiv1.RuntimeState_RUNTIME_STATE_SUSPENDED}
 			store := &serviceTestStore{getResult: stored}
 			authorizer := &recordingAuthorizer{denied: map[string]bool{id: test.denied}}
-			// A suspended read must never touch a workflow.
+			// Get never touches a workflow: MCP resolves every invoke through it.
 			service := NewService(store, authorizer, nil)
 			result, err := service.Get(ctx, id)
 			if test.code != "" {
diff --git a/proto/kagent/api/v1/sessions.proto b/proto/kagent/api/v1/sessions.proto
index 971fd45102..bea59434e9 100644
--- a/proto/kagent/api/v1/sessions.proto
+++ b/proto/kagent/api/v1/sessions.proto
@@ -50,6 +50,25 @@ message Session {
   // Output only. A2A conversation ID, equal to this session's ID.
   // Forks receive a new conversation ID and new task IDs.
   string context_id = 14;
+  // Output only. Set only by GetSession; every other response leaves it unset.
+  SessionRecovery recovery = 16;
+}
+
+message SessionRecovery {
+  // Whether the recorded runtime crashed and needs a revert. Unset when
+  // GetSession could not observe the runtime; clients must not offer a revert then.
+  optional bool needed = 1;
+  // User messages the restored runtime does not remember, from every revert of
+  // this Session. Each appears once. Their A2A history remains available,
+  // including completed task results, so the list lasts as long as the
+  // Session and grows only when a revert discards input.
+  repeated SessionMessageReference failed_messages = 3;
+}
+
+// Message IDs are unique within a task, not across a conversation.
+message SessionMessageReference {
+  string task_id = 1;
+  string message_id = 2;
 }
 
 message AgentProvenance {
diff --git a/ui/src/generated/kagent/api/v1/sessions_pb.ts b/ui/src/generated/kagent/api/v1/sessions_pb.ts
index fc1a06d15c..3e75f9b773 100644
--- a/ui/src/generated/kagent/api/v1/sessions_pb.ts
+++ b/ui/src/generated/kagent/api/v1/sessions_pb.ts
@@ -17,7 +17,7 @@ import type { Message } from "@bufbuild/protobuf";
  * Describes the file kagent/api/v1/sessions.proto.
  */
 export const file_kagent_api_v1_sessions: GenFile = /*@__PURE__*/
-  fileDesc("ChxrYWdlbnQvYXBpL3YxL3Nlc3Npb25zLnByb3RvEg1rYWdlbnQuYXBpLnYxIioKB0ZhaWx1cmUSDgoGcmVhc29uGAEgASgJEg8KB21lc3NhZ2UYAiABKAkikgMKB1Nlc3Npb24SCgoCaWQYASABKAkSDwoHY3JlYXRvchgCIAEoCRItCgVhZ2VudBgPIAEoCzIeLmthZ2VudC5hcGkudjEuQWdlbnRQcm92ZW5hbmNlEhkKEXByZXBhcmVkX3JldmlzaW9uGAUgASgJEhUKDWEyYV9hdXRob3JpdHkYBiABKAkSKgoFc3RhdGUYByABKA4yGy5rYWdlbnQuYXBpLnYxLlJ1bnRpbWVTdGF0ZRIyCglvcGVyYXRpb24YCCABKA4yHy5rYWdlbnQuYXBpLnYxLlJ1bnRpbWVPcGVyYXRpb24SJwoHZmFpbHVyZRgJIAEoCzIWLmthZ2VudC5hcGkudjEuRmFpbHVyZRIuCgpjcmVhdGVkX2F0GAogASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIuCgp1cGRhdGVkX2F0GAsgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIMCgRuYW1lGA0gASgJEhIKCmNvbnRleHRfaWQYDiABKAkiTQoPQWdlbnRQcm92ZW5hbmNlEi0KA3JlZhgBIAEoCzIgLmthZ2VudC5hcGkudjEuUmVzb3VyY2VSZWZlcmVuY2USCwoDdWlkGAIgASgJIrwBChRDcmVhdGVTZXNzaW9uUmVxdWVzdBI3CgVhZ2VudBgFIAEoCzIgLmthZ2VudC5hcGkudjEuUmVzb3VyY2VSZWZlcmVuY2VCBrpIA8gBARIeCgpyZXF1ZXN0X2lkGAMgASgJQgq6SAdyBRABGIABEksKBG5hbWUYBCABKAlCPbpIOnI4GMgBMjNeKD86JHxbXlxwe1p9XHB7Q2N9XSg/OlteXHB7Q2N9XSpbXlxwe1p9XHB7Q2N9XSk/KSQiQAoVQ3JlYXRlU2Vzc2lvblJlc3BvbnNlEicKB3Nlc3Npb24YASABKAsyFi5rYWdlbnQuYXBpLnYxLlNlc3Npb24iMQoRR2V0U2Vzc2lvblJlcXVlc3QSHAoKc2Vzc2lvbl9pZBgBIAEoCUIIukgFcgOwAQEiPQoSR2V0U2Vzc2lvblJlc3BvbnNlEicKB3Nlc3Npb24YASABKAsyFi5rYWdlbnQuYXBpLnYxLlNlc3Npb24ihgEKE0xpc3RTZXNzaW9uc1JlcXVlc3QSFAoMYWxsX2NyZWF0b3JzGAIgASgIEigKBHBhZ2UYAyABKAsyGi5rYWdlbnQuYXBpLnYxLlBhZ2VSZXF1ZXN0Ei8KBWFnZW50GAYgASgLMiAua2FnZW50LmFwaS52MS5SZXNvdXJjZVJlZmVyZW5jZSJrChRMaXN0U2Vzc2lvbnNSZXNwb25zZRIoCghzZXNzaW9ucxgBIAMoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbhIpCgRwYWdlGAIgASgLMhsua2FnZW50LmFwaS52MS5QYWdlUmVzcG9uc2UihQEKGFVwZGF0ZVNlc3Npb25OYW1lUmVxdWVzdBIcCgpzZXNzaW9uX2lkGAEgASgJQgi6SAVyA7ABARJLCgRuYW1lGAIgASgJQj26SDpyOBjIATIzXig/OiR8W15ccHtafVxwe0NjfV0oPzpbXlxwe0NjfV0qW15ccHtafVxwe0NjfV0pPykkIkQKGVVwZGF0ZVNlc3Npb25OYW1lUmVzcG9uc2USJwoHc2Vzc2lvbhgBIAEoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbiI0ChRSZXZlcnRTZXNzaW9uUmVxdWVzdBIcCgpzZXNzaW9uX2lkGAEgASgJQgi6SAVyA7ABASJAChVSZXZlcnRTZXNzaW9uUmVzcG9uc2USJwoHc2Vzc2lvbhgBIAEoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbiI0ChREZWxldGVTZXNzaW9uUmVxdWVzdBIcCgpzZXNzaW9uX2lkGAEgASgJQgi6SAVyA7ABASJAChVEZWxldGVTZXNzaW9uUmVzcG9uc2USJwoHc2Vzc2lvbhgBIAEoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbiKOAQoMU2Vzc2lvblNoYXJlEgoKAmlkGAEgASgJEhIKCnNlc3Npb25faWQYAiABKAkSLgoKY3JlYXRlZF9hdBgDIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXASLgoKZXhwaXJlc19hdBgEIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXAiawoZQ3JlYXRlU2Vzc2lvblNoYXJlUmVxdWVzdBIcCgpzZXNzaW9uX2lkGAEgASgJQgi6SAVyA7ABARIwCgN0dGwYAiABKAsyGS5nb29nbGUucHJvdG9idWYuRHVyYXRpb25CCLpIBaoBAioAIlcKGkNyZWF0ZVNlc3Npb25TaGFyZVJlc3BvbnNlEioKBXNoYXJlGAEgASgLMhsua2FnZW50LmFwaS52MS5TZXNzaW9uU2hhcmUSDQoFdG9rZW4YAiABKAkiYgoYTGlzdFNlc3Npb25TaGFyZXNSZXF1ZXN0EhwKCnNlc3Npb25faWQYASABKAlCCLpIBXIDsAEBEigKBHBhZ2UYAiABKAsyGi5rYWdlbnQuYXBpLnYxLlBhZ2VSZXF1ZXN0InMKGUxpc3RTZXNzaW9uU2hhcmVzUmVzcG9uc2USKwoGc2hhcmVzGAEgAygLMhsua2FnZW50LmFwaS52MS5TZXNzaW9uU2hhcmUSKQoEcGFnZRgCIAEoCzIbLmthZ2VudC5hcGkudjEuUGFnZVJlc3BvbnNlIjcKGVJldm9rZVNlc3Npb25TaGFyZVJlcXVlc3QSGgoIc2hhcmVfaWQYASABKAlCCLpIBXIDsAEBIhwKGlJldm9rZVNlc3Npb25TaGFyZVJlc3BvbnNlMvYGCg5TZXNzaW9uU2VydmljZRJaCg1DcmVhdGVTZXNzaW9uEiMua2FnZW50LmFwaS52MS5DcmVhdGVTZXNzaW9uUmVxdWVzdBokLmthZ2VudC5hcGkudjEuQ3JlYXRlU2Vzc2lvblJlc3BvbnNlElEKCkdldFNlc3Npb24SIC5rYWdlbnQuYXBpLnYxLkdldFNlc3Npb25SZXF1ZXN0GiEua2FnZW50LmFwaS52MS5HZXRTZXNzaW9uUmVzcG9uc2USVwoMTGlzdFNlc3Npb25zEiIua2FnZW50LmFwaS52MS5MaXN0U2Vzc2lvbnNSZXF1ZXN0GiMua2FnZW50LmFwaS52MS5MaXN0U2Vzc2lvbnNSZXNwb25zZRJmChFVcGRhdGVTZXNzaW9uTmFtZRInLmthZ2VudC5hcGkudjEuVXBkYXRlU2Vzc2lvbk5hbWVSZXF1ZXN0Gigua2FnZW50LmFwaS52MS5VcGRhdGVTZXNzaW9uTmFtZVJlc3BvbnNlEloKDVJldmVydFNlc3Npb24SIy5rYWdlbnQuYXBpLnYxLlJldmVydFNlc3Npb25SZXF1ZXN0GiQua2FnZW50LmFwaS52MS5SZXZlcnRTZXNzaW9uUmVzcG9uc2USWgoNRGVsZXRlU2Vzc2lvbhIjLmthZ2VudC5hcGkudjEuRGVsZXRlU2Vzc2lvblJlcXVlc3QaJC5rYWdlbnQuYXBpLnYxLkRlbGV0ZVNlc3Npb25SZXNwb25zZRJpChJDcmVhdGVTZXNzaW9uU2hhcmUSKC5rYWdlbnQuYXBpLnYxLkNyZWF0ZVNlc3Npb25TaGFyZVJlcXVlc3QaKS5rYWdlbnQuYXBpLnYxLkNyZWF0ZVNlc3Npb25TaGFyZVJlc3BvbnNlEmYKEUxpc3RTZXNzaW9uU2hhcmVzEicua2FnZW50LmFwaS52MS5MaXN0U2Vzc2lvblNoYXJlc1JlcXVlc3QaKC5rYWdlbnQuYXBpLnYxLkxpc3RTZXNzaW9uU2hhcmVzUmVzcG9uc2USaQoSUmV2b2tlU2Vzc2lvblNoYXJlEigua2FnZW50LmFwaS52MS5SZXZva2VTZXNzaW9uU2hhcmVSZXF1ZXN0Gikua2FnZW50LmFwaS52MS5SZXZva2VTZXNzaW9uU2hhcmVSZXNwb25zZUI9WjtnaXRodWIuY29tL2thZ2VudC1kZXYva2FnZW50L2dvL2FwaS9nZW4va2FnZW50L2FwaS92MTthcGl2MWIGcHJvdG8z", [file_buf_validate_validate, file_google_protobuf_duration, file_google_protobuf_timestamp, file_kagent_api_v1_common, file_kagent_api_v1_runtime]);
+  fileDesc("ChxrYWdlbnQvYXBpL3YxL3Nlc3Npb25zLnByb3RvEg1rYWdlbnQuYXBpLnYxIioKB0ZhaWx1cmUSDgoGcmVhc29uGAEgASgJEg8KB21lc3NhZ2UYAiABKAkixAMKB1Nlc3Npb24SCgoCaWQYASABKAkSDwoHY3JlYXRvchgCIAEoCRItCgVhZ2VudBgPIAEoCzIeLmthZ2VudC5hcGkudjEuQWdlbnRQcm92ZW5hbmNlEhkKEXByZXBhcmVkX3JldmlzaW9uGAUgASgJEhUKDWEyYV9hdXRob3JpdHkYBiABKAkSKgoFc3RhdGUYByABKA4yGy5rYWdlbnQuYXBpLnYxLlJ1bnRpbWVTdGF0ZRIyCglvcGVyYXRpb24YCCABKA4yHy5rYWdlbnQuYXBpLnYxLlJ1bnRpbWVPcGVyYXRpb24SJwoHZmFpbHVyZRgJIAEoCzIWLmthZ2VudC5hcGkudjEuRmFpbHVyZRIuCgpjcmVhdGVkX2F0GAogASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIuCgp1cGRhdGVkX2F0GAsgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIMCgRuYW1lGA0gASgJEhIKCmNvbnRleHRfaWQYDiABKAkSMAoIcmVjb3ZlcnkYECABKAsyHi5rYWdlbnQuYXBpLnYxLlNlc3Npb25SZWNvdmVyeSJyCg9TZXNzaW9uUmVjb3ZlcnkSEwoGbmVlZGVkGAEgASgISACIAQESPwoPZmFpbGVkX21lc3NhZ2VzGAMgAygLMiYua2FnZW50LmFwaS52MS5TZXNzaW9uTWVzc2FnZVJlZmVyZW5jZUIJCgdfbmVlZGVkIj4KF1Nlc3Npb25NZXNzYWdlUmVmZXJlbmNlEg8KB3Rhc2tfaWQYASABKAkSEgoKbWVzc2FnZV9pZBgCIAEoCSJNCg9BZ2VudFByb3ZlbmFuY2USLQoDcmVmGAEgASgLMiAua2FnZW50LmFwaS52MS5SZXNvdXJjZVJlZmVyZW5jZRILCgN1aWQYAiABKAkivAEKFENyZWF0ZVNlc3Npb25SZXF1ZXN0EjcKBWFnZW50GAUgASgLMiAua2FnZW50LmFwaS52MS5SZXNvdXJjZVJlZmVyZW5jZUIGukgDyAEBEh4KCnJlcXVlc3RfaWQYAyABKAlCCrpIB3IFEAEYgAESSwoEbmFtZRgEIAEoCUI9ukg6cjgYyAEyM14oPzokfFteXHB7Wn1ccHtDY31dKD86W15ccHtDY31dKlteXHB7Wn1ccHtDY31dKT8pJCJAChVDcmVhdGVTZXNzaW9uUmVzcG9uc2USJwoHc2Vzc2lvbhgBIAEoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbiIxChFHZXRTZXNzaW9uUmVxdWVzdBIcCgpzZXNzaW9uX2lkGAEgASgJQgi6SAVyA7ABASI9ChJHZXRTZXNzaW9uUmVzcG9uc2USJwoHc2Vzc2lvbhgBIAEoCzIWLmthZ2VudC5hcGkudjEuU2Vzc2lvbiKGAQoTTGlzdFNlc3Npb25zUmVxdWVzdBIUCgxhbGxfY3JlYXRvcnMYAiABKAgSKAoEcGFnZRgDIAEoCzIaLmthZ2VudC5hcGkudjEuUGFnZVJlcXVlc3QSLwoFYWdlbnQYBiABKAsyIC5rYWdlbnQuYXBpLnYxLlJlc291cmNlUmVmZXJlbmNlImsKFExpc3RTZXNzaW9uc1Jlc3BvbnNlEigKCHNlc3Npb25zGAEgAygLMhYua2FnZW50LmFwaS52MS5TZXNzaW9uEikKBHBhZ2UYAiABKAsyGy5rYWdlbnQuYXBpLnYxLlBhZ2VSZXNwb25zZSKFAQoYVXBkYXRlU2Vzc2lvbk5hbWVSZXF1ZXN0EhwKCnNlc3Npb25faWQYASABKAlCCLpIBXIDsAEBEksKBG5hbWUYAiABKAlCPbpIOnI4GMgBMjNeKD86JHxbXlxwe1p9XHB7Q2N9XSg/OlteXHB7Q2N9XSpbXlxwe1p9XHB7Q2N9XSk/KSQiRAoZVXBkYXRlU2Vzc2lvbk5hbWVSZXNwb25zZRInCgdzZXNzaW9uGAEgASgLMhYua2FnZW50LmFwaS52MS5TZXNzaW9uIjQKFFJldmVydFNlc3Npb25SZXF1ZXN0EhwKCnNlc3Npb25faWQYASABKAlCCLpIBXIDsAEBIkAKFVJldmVydFNlc3Npb25SZXNwb25zZRInCgdzZXNzaW9uGAEgASgLMhYua2FnZW50LmFwaS52MS5TZXNzaW9uIjQKFERlbGV0ZVNlc3Npb25SZXF1ZXN0EhwKCnNlc3Npb25faWQYASABKAlCCLpIBXIDsAEBIkAKFURlbGV0ZVNlc3Npb25SZXNwb25zZRInCgdzZXNzaW9uGAEgASgLMhYua2FnZW50LmFwaS52MS5TZXNzaW9uIo4BCgxTZXNzaW9uU2hhcmUSCgoCaWQYASABKAkSEgoKc2Vzc2lvbl9pZBgCIAEoCRIuCgpjcmVhdGVkX2F0GAMgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIuCgpleHBpcmVzX2F0GAQgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcCJrChlDcmVhdGVTZXNzaW9uU2hhcmVSZXF1ZXN0EhwKCnNlc3Npb25faWQYASABKAlCCLpIBXIDsAEBEjAKA3R0bBgCIAEoCzIZLmdvb2dsZS5wcm90b2J1Zi5EdXJhdGlvbkIIukgFqgECKgAiVwoaQ3JlYXRlU2Vzc2lvblNoYXJlUmVzcG9uc2USKgoFc2hhcmUYASABKAsyGy5rYWdlbnQuYXBpLnYxLlNlc3Npb25TaGFyZRINCgV0b2tlbhgCIAEoCSJiChhMaXN0U2Vzc2lvblNoYXJlc1JlcXVlc3QSHAoKc2Vzc2lvbl9pZBgBIAEoCUIIukgFcgOwAQESKAoEcGFnZRgCIAEoCzIaLmthZ2VudC5hcGkudjEuUGFnZVJlcXVlc3QicwoZTGlzdFNlc3Npb25TaGFyZXNSZXNwb25zZRIrCgZzaGFyZXMYASADKAsyGy5rYWdlbnQuYXBpLnYxLlNlc3Npb25TaGFyZRIpCgRwYWdlGAIgASgLMhsua2FnZW50LmFwaS52MS5QYWdlUmVzcG9uc2UiNwoZUmV2b2tlU2Vzc2lvblNoYXJlUmVxdWVzdBIaCghzaGFyZV9pZBgBIAEoCUIIukgFcgOwAQEiHAoaUmV2b2tlU2Vzc2lvblNoYXJlUmVzcG9uc2Uy9gYKDlNlc3Npb25TZXJ2aWNlEloKDUNyZWF0ZVNlc3Npb24SIy5rYWdlbnQuYXBpLnYxLkNyZWF0ZVNlc3Npb25SZXF1ZXN0GiQua2FnZW50LmFwaS52MS5DcmVhdGVTZXNzaW9uUmVzcG9uc2USUQoKR2V0U2Vzc2lvbhIgLmthZ2VudC5hcGkudjEuR2V0U2Vzc2lvblJlcXVlc3QaIS5rYWdlbnQuYXBpLnYxLkdldFNlc3Npb25SZXNwb25zZRJXCgxMaXN0U2Vzc2lvbnMSIi5rYWdlbnQuYXBpLnYxLkxpc3RTZXNzaW9uc1JlcXVlc3QaIy5rYWdlbnQuYXBpLnYxLkxpc3RTZXNzaW9uc1Jlc3BvbnNlEmYKEVVwZGF0ZVNlc3Npb25OYW1lEicua2FnZW50LmFwaS52MS5VcGRhdGVTZXNzaW9uTmFtZVJlcXVlc3QaKC5rYWdlbnQuYXBpLnYxLlVwZGF0ZVNlc3Npb25OYW1lUmVzcG9uc2USWgoNUmV2ZXJ0U2Vzc2lvbhIjLmthZ2VudC5hcGkudjEuUmV2ZXJ0U2Vzc2lvblJlcXVlc3QaJC5rYWdlbnQuYXBpLnYxLlJldmVydFNlc3Npb25SZXNwb25zZRJaCg1EZWxldGVTZXNzaW9uEiMua2FnZW50LmFwaS52MS5EZWxldGVTZXNzaW9uUmVxdWVzdBokLmthZ2VudC5hcGkudjEuRGVsZXRlU2Vzc2lvblJlc3BvbnNlEmkKEkNyZWF0ZVNlc3Npb25TaGFyZRIoLmthZ2VudC5hcGkudjEuQ3JlYXRlU2Vzc2lvblNoYXJlUmVxdWVzdBopLmthZ2VudC5hcGkudjEuQ3JlYXRlU2Vzc2lvblNoYXJlUmVzcG9uc2USZgoRTGlzdFNlc3Npb25TaGFyZXMSJy5rYWdlbnQuYXBpLnYxLkxpc3RTZXNzaW9uU2hhcmVzUmVxdWVzdBooLmthZ2VudC5hcGkudjEuTGlzdFNlc3Npb25TaGFyZXNSZXNwb25zZRJpChJSZXZva2VTZXNzaW9uU2hhcmUSKC5rYWdlbnQuYXBpLnYxLlJldm9rZVNlc3Npb25TaGFyZVJlcXVlc3QaKS5rYWdlbnQuYXBpLnYxLlJldm9rZVNlc3Npb25TaGFyZVJlc3BvbnNlQj1aO2dpdGh1Yi5jb20va2FnZW50LWRldi9rYWdlbnQvZ28vYXBpL2dlbi9rYWdlbnQvYXBpL3YxO2FwaXYxYgZwcm90bzM", [file_buf_validate_validate, file_google_protobuf_duration, file_google_protobuf_timestamp, file_kagent_api_v1_common, file_kagent_api_v1_runtime]);
 
 /**
  * @generated from message kagent.api.v1.Failure
@@ -111,6 +111,13 @@ export type Session = Message<"kagent.api.v1.Session"> & {
    * @generated from field: string context_id = 14;
    */
   contextId: string;
+
+  /**
+   * Output only. Set only by GetSession; every other response leaves it unset.
+   *
+   * @generated from field: kagent.api.v1.SessionRecovery recovery = 16;
+   */
+  recovery?: SessionRecovery | undefined;
 };
 
 /**
@@ -120,6 +127,60 @@ export type Session = Message<"kagent.api.v1.Session"> & {
 export const SessionSchema: GenMessage<Session> = /*@__PURE__*/
   messageDesc(file_kagent_api_v1_sessions, 1);
 
+/**
+ * @generated from message kagent.api.v1.SessionRecovery
+ */
+export type SessionRecovery = Message<"kagent.api.v1.SessionRecovery"> & {
+  /**
+   * Whether the recorded runtime crashed and needs a revert. Unset when
+   * GetSession could not observe the runtime; clients must not offer a revert then.
+   *
+   * @generated from field: optional bool needed = 1;
+   */
+  needed?: boolean | undefined;
+
+  /**
+   * User messages the restored runtime does not remember, from every revert of
+   * this Session. Each appears once. Their A2A history remains available,
+   * including completed task results, so the list lasts as long as the
+   * Session and grows only when a revert discards input.
+   *
+   * @generated from field: repeated kagent.api.v1.SessionMessageReference failed_messages = 3;
+   */
+  failedMessages: SessionMessageReference[];
+};
+
+/**
+ * Describes the message kagent.api.v1.SessionRecovery.
+ * Use `create(SessionRecoverySchema)` to create a new message.
+ */
+export const SessionRecoverySchema: GenMessage<SessionRecovery> = /*@__PURE__*/
+  messageDesc(file_kagent_api_v1_sessions, 2);
+
+/**
+ * Message IDs are unique within a task, not across a conversation.
+ *
+ * @generated from message kagent.api.v1.SessionMessageReference
+ */
+export type SessionMessageReference = Message<"kagent.api.v1.SessionMessageReference"> & {
+  /**
+   * @generated from field: string task_id = 1;
+   */
+  taskId: string;
+
+  /**
+   * @generated from field: string message_id = 2;
+   */
+  messageId: string;
+};
+
+/**
+ * Describes the message kagent.api.v1.SessionMessageReference.
+ * Use `create(SessionMessageReferenceSchema)` to create a new message.
+ */
+export const SessionMessageReferenceSchema: GenMessage<SessionMessageReference> = /*@__PURE__*/
+  messageDesc(file_kagent_api_v1_sessions, 3);
+
 /**
  * @generated from message kagent.api.v1.AgentProvenance
  */
@@ -140,7 +201,7 @@ export type AgentProvenance = Message<"kagent.api.v1.AgentProvenance"> & {
  * Use `create(AgentProvenanceSchema)` to create a new message.
  */
 export const AgentProvenanceSchema: GenMessage<AgentProvenance> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 2);
+  messageDesc(file_kagent_api_v1_sessions, 4);
 
 /**
  * @generated from message kagent.api.v1.CreateSessionRequest
@@ -169,7 +230,7 @@ export type CreateSessionRequest = Message<"kagent.api.v1.CreateSessionRequest">
  * Use `create(CreateSessionRequestSchema)` to create a new message.
  */
 export const CreateSessionRequestSchema: GenMessage<CreateSessionRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 3);
+  messageDesc(file_kagent_api_v1_sessions, 5);
 
 /**
  * @generated from message kagent.api.v1.CreateSessionResponse
@@ -186,7 +247,7 @@ export type CreateSessionResponse = Message<"kagent.api.v1.CreateSessionResponse
  * Use `create(CreateSessionResponseSchema)` to create a new message.
  */
 export const CreateSessionResponseSchema: GenMessage<CreateSessionResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 4);
+  messageDesc(file_kagent_api_v1_sessions, 6);
 
 /**
  * @generated from message kagent.api.v1.GetSessionRequest
@@ -203,7 +264,7 @@ export type GetSessionRequest = Message<"kagent.api.v1.GetSessionRequest"> & {
  * Use `create(GetSessionRequestSchema)` to create a new message.
  */
 export const GetSessionRequestSchema: GenMessage<GetSessionRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 5);
+  messageDesc(file_kagent_api_v1_sessions, 7);
 
 /**
  * @generated from message kagent.api.v1.GetSessionResponse
@@ -220,7 +281,7 @@ export type GetSessionResponse = Message<"kagent.api.v1.GetSessionResponse"> & {
  * Use `create(GetSessionResponseSchema)` to create a new message.
  */
 export const GetSessionResponseSchema: GenMessage<GetSessionResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 6);
+  messageDesc(file_kagent_api_v1_sessions, 8);
 
 /**
  * @generated from message kagent.api.v1.ListSessionsRequest
@@ -251,7 +312,7 @@ export type ListSessionsRequest = Message<"kagent.api.v1.ListSessionsRequest"> &
  * Use `create(ListSessionsRequestSchema)` to create a new message.
  */
 export const ListSessionsRequestSchema: GenMessage<ListSessionsRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 7);
+  messageDesc(file_kagent_api_v1_sessions, 9);
 
 /**
  * @generated from message kagent.api.v1.ListSessionsResponse
@@ -273,7 +334,7 @@ export type ListSessionsResponse = Message<"kagent.api.v1.ListSessionsResponse">
  * Use `create(ListSessionsResponseSchema)` to create a new message.
  */
 export const ListSessionsResponseSchema: GenMessage<ListSessionsResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 8);
+  messageDesc(file_kagent_api_v1_sessions, 10);
 
 /**
  * @generated from message kagent.api.v1.UpdateSessionNameRequest
@@ -298,7 +359,7 @@ export type UpdateSessionNameRequest = Message<"kagent.api.v1.UpdateSessionNameR
  * Use `create(UpdateSessionNameRequestSchema)` to create a new message.
  */
 export const UpdateSessionNameRequestSchema: GenMessage<UpdateSessionNameRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 9);
+  messageDesc(file_kagent_api_v1_sessions, 11);
 
 /**
  * @generated from message kagent.api.v1.UpdateSessionNameResponse
@@ -315,7 +376,7 @@ export type UpdateSessionNameResponse = Message<"kagent.api.v1.UpdateSessionName
  * Use `create(UpdateSessionNameResponseSchema)` to create a new message.
  */
 export const UpdateSessionNameResponseSchema: GenMessage<UpdateSessionNameResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 10);
+  messageDesc(file_kagent_api_v1_sessions, 12);
 
 /**
  * @generated from message kagent.api.v1.RevertSessionRequest
@@ -332,7 +393,7 @@ export type RevertSessionRequest = Message<"kagent.api.v1.RevertSessionRequest">
  * Use `create(RevertSessionRequestSchema)` to create a new message.
  */
 export const RevertSessionRequestSchema: GenMessage<RevertSessionRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 11);
+  messageDesc(file_kagent_api_v1_sessions, 13);
 
 /**
  * @generated from message kagent.api.v1.RevertSessionResponse
@@ -349,7 +410,7 @@ export type RevertSessionResponse = Message<"kagent.api.v1.RevertSessionResponse
  * Use `create(RevertSessionResponseSchema)` to create a new message.
  */
 export const RevertSessionResponseSchema: GenMessage<RevertSessionResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 12);
+  messageDesc(file_kagent_api_v1_sessions, 14);
 
 /**
  * @generated from message kagent.api.v1.DeleteSessionRequest
@@ -366,7 +427,7 @@ export type DeleteSessionRequest = Message<"kagent.api.v1.DeleteSessionRequest">
  * Use `create(DeleteSessionRequestSchema)` to create a new message.
  */
 export const DeleteSessionRequestSchema: GenMessage<DeleteSessionRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 13);
+  messageDesc(file_kagent_api_v1_sessions, 15);
 
 /**
  * @generated from message kagent.api.v1.DeleteSessionResponse
@@ -383,7 +444,7 @@ export type DeleteSessionResponse = Message<"kagent.api.v1.DeleteSessionResponse
  * Use `create(DeleteSessionResponseSchema)` to create a new message.
  */
 export const DeleteSessionResponseSchema: GenMessage<DeleteSessionResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 14);
+  messageDesc(file_kagent_api_v1_sessions, 16);
 
 /**
  * Grants read-only access to a session, including A2A get, list, and subscribe.
@@ -419,7 +480,7 @@ export type SessionShare = Message<"kagent.api.v1.SessionShare"> & {
  * Use `create(SessionShareSchema)` to create a new message.
  */
 export const SessionShareSchema: GenMessage<SessionShare> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 15);
+  messageDesc(file_kagent_api_v1_sessions, 17);
 
 /**
  * @generated from message kagent.api.v1.CreateSessionShareRequest
@@ -445,7 +506,7 @@ export type CreateSessionShareRequest = Message<"kagent.api.v1.CreateSessionShar
  * Use `create(CreateSessionShareRequestSchema)` to create a new message.
  */
 export const CreateSessionShareRequestSchema: GenMessage<CreateSessionShareRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 16);
+  messageDesc(file_kagent_api_v1_sessions, 18);
 
 /**
  * @generated from message kagent.api.v1.CreateSessionShareResponse
@@ -469,7 +530,7 @@ export type CreateSessionShareResponse = Message<"kagent.api.v1.CreateSessionSha
  * Use `create(CreateSessionShareResponseSchema)` to create a new message.
  */
 export const CreateSessionShareResponseSchema: GenMessage<CreateSessionShareResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 17);
+  messageDesc(file_kagent_api_v1_sessions, 19);
 
 /**
  * @generated from message kagent.api.v1.ListSessionSharesRequest
@@ -491,7 +552,7 @@ export type ListSessionSharesRequest = Message<"kagent.api.v1.ListSessionSharesR
  * Use `create(ListSessionSharesRequestSchema)` to create a new message.
  */
 export const ListSessionSharesRequestSchema: GenMessage<ListSessionSharesRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 18);
+  messageDesc(file_kagent_api_v1_sessions, 20);
 
 /**
  * @generated from message kagent.api.v1.ListSessionSharesResponse
@@ -513,7 +574,7 @@ export type ListSessionSharesResponse = Message<"kagent.api.v1.ListSessionShares
  * Use `create(ListSessionSharesResponseSchema)` to create a new message.
  */
 export const ListSessionSharesResponseSchema: GenMessage<ListSessionSharesResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 19);
+  messageDesc(file_kagent_api_v1_sessions, 21);
 
 /**
  * @generated from message kagent.api.v1.RevokeSessionShareRequest
@@ -530,7 +591,7 @@ export type RevokeSessionShareRequest = Message<"kagent.api.v1.RevokeSessionShar
  * Use `create(RevokeSessionShareRequestSchema)` to create a new message.
  */
 export const RevokeSessionShareRequestSchema: GenMessage<RevokeSessionShareRequest> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 20);
+  messageDesc(file_kagent_api_v1_sessions, 22);
 
 /**
  * @generated from message kagent.api.v1.RevokeSessionShareResponse
@@ -543,7 +604,7 @@ export type RevokeSessionShareResponse = Message<"kagent.api.v1.RevokeSessionSha
  * Use `create(RevokeSessionShareResponseSchema)` to create a new message.
  */
 export const RevokeSessionShareResponseSchema: GenMessage<RevokeSessionShareResponse> = /*@__PURE__*/
-  messageDesc(file_kagent_api_v1_sessions, 21);
+  messageDesc(file_kagent_api_v1_sessions, 23);
 
 /**
  * Lifecycle mutations execute inline. After a retryable error, repeat the same