Keep original recording facts across verified Agent session cutovers

internal/agent/recording_client.go

@@ -5,9 +5,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
+	"time"
 
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	"git.ipao.vip/rogee/go-sip/internal/session"
 
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -53,16 +58,19 @@ func (c RecordingClient) RequestUpload(ctx context.Context, asset *agentpb.Asset
 		asset.GetExecutionId() != c.SourceEventID || asset.GetCallId() != c.SourceEventID || asset.GetAssetId() == "" || asset.GetSizeBytes() <= 0 || asset.GetChecksumSha256() == "" {
 		return nil, errors.New("recording token request does not describe the approved call and original asset")
 	}
-	response, err := c.Client.RequestRecordingUpload(ctx, &agentpb.RequestRecordingUploadRequest{
-		Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
-		UploadId: uploadID, Asset: proto.Clone(asset).(*agentpb.AssetDescriptor),
+	originalAsset := proto.Clone(asset).(*agentpb.AssetDescriptor)
+	response, err := reReportVerifiedSessionCutover(ctx, c, meta, "upload/"+uploadID, func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.RequestRecordingUploadResponse, error) {
+		return c.Client.RequestRecordingUpload(callCtx, &agentpb.RequestRecordingUploadRequest{
+			Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
+			UploadId: uploadID, Asset: proto.Clone(originalAsset).(*agentpb.AssetDescriptor),
+		})
 	})
 	if err != nil {
 		return nil, fmt.Errorf("request original recording upload token: %w", err)
 	}
 	grant := response.GetGrant()
 	if grant == nil || grant.GetUploadId() != uploadID || grant.GetBucket() == "" || grant.GetObjectKey() == "" ||
-		grant.GetMaxBytes() != asset.GetSizeBytes() || grant.GetRequiredChecksumSha256() != asset.GetChecksumSha256() {
+		grant.GetMaxBytes() != originalAsset.GetSizeBytes() || grant.GetRequiredChecksumSha256() != originalAsset.GetChecksumSha256() {
 		return nil, fmt.Errorf("%w: Dispatcher returned a different recording asset", ErrUploadGrantInvalid)
 	}
 	return grant, nil
@@ -73,8 +81,10 @@ func (c RecordingClient) ReportEnded(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	response, err := c.Client.ReportCallEnded(ctx, &agentpb.ReportCallEndedRequest{
-		Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
+	response, err := reReportVerifiedSessionCutover(ctx, c, meta, "ended", func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.ReportCallEndedResponse, error) {
+		return c.Client.ReportCallEnded(callCtx, &agentpb.ReportCallEndedRequest{
+			Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
+		})
 	})
 	if err != nil {
 		return fmt.Errorf("report confirmed call end: %w", err)
@@ -93,14 +103,21 @@ func (c RecordingClient) ReportFinal(ctx context.Context, payload []byte, upload
 	if len(payload) == 0 {
 		return "", errors.New("final call result payload is required")
 	}
-	request := &agentpb.ReportCallResultRequest{
-		Meta: meta, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
-		ResultPayloadJson: bytes.Clone(payload),
-	}
+	originalPayload := bytes.Clone(payload)
+	var originalUpload *agentpb.UploadObservation
 	if upload != nil {
-		request.Upload = proto.Clone(upload).(*agentpb.UploadObservation)
+		originalUpload = proto.Clone(upload).(*agentpb.UploadObservation)
 	}
-	response, err := c.Client.ReportCallResult(ctx, request)
+	response, err := reReportVerifiedSessionCutover(ctx, c, meta, "result", func(callCtx context.Context, current *agentpb.RequestMeta) (*agentpb.ReportCallResultResponse, error) {
+		request := &agentpb.ReportCallResultRequest{
+			Meta: current, DispatcherId: c.DispatcherID, TenantId: c.TenantID, SourceEventId: c.SourceEventID,
+			ResultPayloadJson: bytes.Clone(originalPayload),
+		}
+		if originalUpload != nil {
+			request.Upload = proto.Clone(originalUpload).(*agentpb.UploadObservation)
+		}
+		return c.Client.ReportCallResult(callCtx, request)
+	})
 	if err != nil {
 		return "", fmt.Errorf("persist unique call result: %w", err)
 	}
@@ -109,3 +126,44 @@ func (c RecordingClient) ReportFinal(ctx context.Context, payload []byte, upload
 	}
 	return "", errors.New("Dispatcher did not persist the unique call result")
 }
+
+func reReportVerifiedSessionCutover[T any](ctx context.Context, c RecordingClient, original *agentpb.RequestMeta, action string, send func(context.Context, *agentpb.RequestMeta) (T, error)) (T, error) {
+	response, err := send(ctx, original)
+	if !isVerifiedSessionTransition(err) {
+		return response, err
+	}
+	log.Printf("Agent fact fenced during session transition event=%s agent=%s generation=%d action=%s", c.SourceEventID, original.AgentId, original.SessionGeneration, action)
+	retryCtx, cancel := context.WithTimeout(ctx, 6*time.Second)
+	defer cancel()
+	attempts := 1
+	for {
+		timer := time.NewTimer(100 * time.Millisecond)
+		select {
+		case <-retryCtx.Done():
+			timer.Stop()
+			log.Printf("Agent fact re-report deadline event=%s agent=%s action=%s attempts=%d", c.SourceEventID, original.AgentId, action, attempts)
+			var zero T
+			return zero, fmt.Errorf("Agent fact re-report not confirmed: %w", errors.Join(err, retryCtx.Err()))
+		case <-timer.C:
+		}
+		current, sessionErr := c.requestMeta(retryCtx, action)
+		if sessionErr != nil {
+			var zero T
+			return zero, fmt.Errorf("obtain renewed Agent session for original fact: %w", sessionErr)
+		}
+		if current.AgentId != original.AgentId || current.CellId != original.CellId || current.BootId != original.BootId || current.DispatcherEpoch != original.DispatcherEpoch {
+			var zero T
+			return zero, fmt.Errorf("%w: Agent identity changed during session transition", ErrRecordingClientUnavailable)
+		}
+		response, err = send(retryCtx, current)
+		attempts++
+		if !isVerifiedSessionTransition(err) {
+			return response, err
+		}
+	}
+}
+
+func isVerifiedSessionTransition(err error) bool {
+	st, ok := status.FromError(err)
+	return ok && st.Code() == codes.Unauthenticated && st.Message() == session.GenerationTransitionMessage
+}

internal/agent/recording_client_transition_test.go

@@ -0,0 +1,202 @@
+package agent
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+const verifiedSessionCutover = "verified Agent session generation changed"
+
+type cutoverRecordingRPC struct {
+	agentpb.AgentControlServiceClient
+	failure        error
+	lateFailure    error
+	alwaysReject   bool
+	sameGeneration bool
+	changeIdentity string
+	generation     uint64
+	agentID        string
+	cellID         string
+	boot           string
+	epoch          string
+	metas          []*agentpb.RequestMeta
+	assets         []*agentpb.AssetDescriptor
+	uploadIDs      []string
+	payloads       [][]byte
+}
+
+func (s *cutoverRecordingRPC) receive(meta *agentpb.RequestMeta) error {
+	s.metas = append(s.metas, proto.Clone(meta).(*agentpb.RequestMeta))
+	if len(s.metas) == 1 {
+		if !s.sameGeneration {
+			s.generation++
+		}
+		switch s.changeIdentity {
+		case "agent":
+			s.agentID = "different-agent"
+		case "cell":
+			s.cellID = "different-cell"
+		case "boot":
+			s.boot = "different-boot"
+		case "epoch":
+			s.epoch = "different-epoch"
+		}
+		return s.failure
+	}
+	if s.lateFailure != nil {
+		return s.lateFailure
+	}
+	if s.alwaysReject {
+		return s.failure
+	}
+	return nil
+}
+
+func (s *cutoverRecordingRPC) RequestRecordingUpload(_ context.Context, req *agentpb.RequestRecordingUploadRequest, _ ...grpc.CallOption) (*agentpb.RequestRecordingUploadResponse, error) {
+	s.uploadIDs = append(s.uploadIDs, req.UploadId)
+	s.assets = append(s.assets, proto.Clone(req.Asset).(*agentpb.AssetDescriptor))
+	if err := s.receive(req.Meta); err != nil {
+		return nil, err
+	}
+	return &agentpb.RequestRecordingUploadResponse{Grant: &agentpb.UploadGrant{
+		UploadId: req.UploadId, Bucket: "mock-bucket", ObjectKey: "mock-key",
+		MaxBytes: req.Asset.SizeBytes, RequiredChecksumSha256: req.Asset.ChecksumSha256,
+	}}, nil
+}
+
+func (s *cutoverRecordingRPC) ReportCallEnded(_ context.Context, req *agentpb.ReportCallEndedRequest, _ ...grpc.CallOption) (*agentpb.ReportCallEndedResponse, error) {
+	if err := s.receive(req.Meta); err != nil {
+		return nil, err
+	}
+	return &agentpb.ReportCallEndedResponse{Receipt: &agentpb.OperationReceipt{FactId: req.SourceEventId, Result: agentpb.ResultCode_RESULT_CODE_APPLIED}}, nil
+}
+
+func (s *cutoverRecordingRPC) ReportCallResult(_ context.Context, req *agentpb.ReportCallResultRequest, _ ...grpc.CallOption) (*agentpb.ReportCallResultResponse, error) {
+	s.payloads = append(s.payloads, bytes.Clone(req.ResultPayloadJson))
+	if err := s.receive(req.Meta); err != nil {
+		return nil, err
+	}
+	return &agentpb.ReportCallResultResponse{Receipt: &agentpb.OperationReceipt{FactId: "result-fact", Result: agentpb.ResultCode_RESULT_CODE_ACCEPTED}}, nil
+}
+
+func newCutoverRecordingClient(s *cutoverRecordingRPC) RecordingClient {
+	s.generation = 1
+	s.agentID = "agent-1"
+	s.cellID = "cell-1"
+	s.boot = "same-boot"
+	s.epoch = "epoch-1"
+	return RecordingClient{
+		Client: s, DispatcherID: "dispatcher-1", TenantID: 1, SourceEventID: "call-1",
+		Session: func(context.Context) (*agentpb.RequestMeta, error) {
+			return &agentpb.RequestMeta{AgentId: s.agentID, CellId: s.cellID, BootId: s.boot, DispatcherEpoch: s.epoch, SessionGeneration: s.generation}, nil
+		},
+	}
+}
+
+func TestRecordingClientReReportsOnlyOriginalFactAfterVerifiedSessionCutover(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		operation string
+		report    func(RecordingClient) error
+	}{
+		{"upload grant", "call-1/upload/up-1", func(c RecordingClient) error {
+			_, err := c.RequestUpload(context.Background(), &agentpb.AssetDescriptor{
+				Kind: agentpb.AssetKind_ASSET_KIND_RECORDING, ExecutionId: "call-1", CallId: "call-1", AssetId: "recording-1", SizeBytes: 4, ChecksumSha256: "digest",
+			}, "up-1")
+			return err
+		}},
+		{"confirmed end", "call-1/ended", func(c RecordingClient) error { return c.ReportEnded(context.Background()) }},
+		{"final result", "call-1/result", func(c RecordingClient) error {
+			_, err := c.ReportFinal(context.Background(), []byte(`{"call_id":"call-1"}`), nil)
+			return err
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover)}
+			if err := tc.report(newCutoverRecordingClient(transport)); err != nil {
+				t.Fatalf("original fact not delivered after session cutover: %v", err)
+			}
+			if len(transport.metas) != 2 {
+				t.Fatalf("want exactly one bounded re-report, got %d requests", len(transport.metas))
+			}
+			old, current := transport.metas[0], transport.metas[1]
+			if old.SessionGeneration != 1 || current.SessionGeneration != 2 ||
+				old.AgentId != current.AgentId || old.CellId != current.CellId || old.BootId != current.BootId || old.DispatcherEpoch != current.DispatcherEpoch ||
+				old.OperationId != tc.operation || current.OperationId != tc.operation || old.IdempotencyKey != tc.operation || current.IdempotencyKey != tc.operation {
+				t.Fatalf("session cutover changed fact identity or accepted an unbound Agent: first=%v second=%v", old, current)
+			}
+			if tc.name == "upload grant" && (len(transport.assets) != 2 || !proto.Equal(transport.assets[0], transport.assets[1]) || transport.uploadIDs[0] != "up-1" || transport.uploadIDs[1] != "up-1") {
+				t.Fatal("upload grant request changed original asset/upload identity")
+			}
+			if tc.name == "final result" && (len(transport.payloads) != 2 || !bytes.Equal(transport.payloads[0], transport.payloads[1])) {
+				t.Fatal("re-report changed original final result payload")
+			}
+		})
+	}
+}
+
+func TestRecordingClientDoesNotRetryUnverifiedOrUnknownOutcomes(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		failure  error
+		identity string
+	}{
+		{"permission denied", status.Error(codes.PermissionDenied, "inactive Agent"), ""},
+		{"timeout with unknown result", status.Error(codes.DeadlineExceeded, "result unknown"), ""},
+		{"unavailable with unknown result", status.Error(codes.Unavailable, "result unknown"), ""},
+		{"unverified unauthenticated", status.Error(codes.Unauthenticated, "mTLS peer missing"), ""},
+		{"different Agent after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "agent"},
+		{"different Cell after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "cell"},
+		{"different boot after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "boot"},
+		{"different epoch after cutover", status.Error(codes.Unauthenticated, verifiedSessionCutover), "epoch"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			transport := &cutoverRecordingRPC{failure: tc.failure, changeIdentity: tc.identity}
+			err := newCutoverRecordingClient(transport).ReportEnded(context.Background())
+			if err == nil || len(transport.metas) != 1 {
+				t.Fatalf("unauthorized or uncertain call end was re-reported: requests=%d error=%v", len(transport.metas), err)
+			}
+		})
+	}
+}
+
+func TestRecordingClientStopsWhenSecondOutcomeIsUnknown(t *testing.T) {
+	transport := &cutoverRecordingRPC{
+		failure:     status.Error(codes.Unauthenticated, verifiedSessionCutover),
+		lateFailure: status.Error(codes.DeadlineExceeded, "result unknown after re-report"),
+	}
+	err := newCutoverRecordingClient(transport).ReportEnded(context.Background())
+	if status.Code(err) != codes.DeadlineExceeded || len(transport.metas) != 2 {
+		t.Fatalf("unknown second outcome caused an unsafe retry: requests=%d error=%v", len(transport.metas), err)
+	}
+}
+
+func TestRecordingClientAllowsOriginalGenerationUntilDispatcherCatchesUp(t *testing.T) {
+	transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover), sameGeneration: true}
+	if err := newCutoverRecordingClient(transport).ReportEnded(context.Background()); err != nil {
+		t.Fatalf("original fact was not delivered after Dispatcher session caught up: %v", err)
+	}
+	if len(transport.metas) != 2 || transport.metas[0].SessionGeneration != transport.metas[1].SessionGeneration {
+		t.Fatalf("re-report changed an otherwise still-active Agent generation: %d requests", len(transport.metas))
+	}
+}
+
+func TestRecordingClientSessionCutoverReReportIsBounded(t *testing.T) {
+	transport := &cutoverRecordingRPC{failure: status.Error(codes.Unauthenticated, verifiedSessionCutover), alwaysReject: true}
+	ctx, cancel := context.WithTimeout(context.Background(), 280*time.Millisecond)
+	defer cancel()
+	err := newCutoverRecordingClient(transport).ReportEnded(ctx)
+	if !errors.Is(err, context.DeadlineExceeded) || len(transport.metas) < 2 || len(transport.metas) > 4 {
+		t.Fatalf("re-report did not stop at caller deadline: requests=%d error=%v", len(transport.metas), err)
+	}
+}

internal/dispatcher/agent.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	agentsession "git.ipao.vip/rogee/go-sip/internal/session"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 	"git.ipao.vip/rogee/go-sip/internal/tenant"
 	"github.com/google/uuid"
@@ -206,9 +207,17 @@ func (c *AgentCoordinator) AuthorizeInboundMeta(meta *agentpb.RequestMeta) error
 	c.mu.Unlock()
 	if !active || !registered || client == nil || session.ExpiresAtUnixMs <= c.now().UnixMilli() ||
 		meta.CellId != session.CellID || meta.BootId != session.BootID ||
-		meta.DispatcherEpoch != session.DispatcherEpoch || meta.SessionGeneration != session.SessionGeneration {
+		meta.DispatcherEpoch != session.DispatcherEpoch {
 		return fmt.Errorf("Agent report does not match a current activated session: %w", store.ErrCommandConflict)
 	}
+	if meta.SessionGeneration != session.SessionGeneration {
+		if meta.SessionGeneration > 0 &&
+			((meta.SessionGeneration < session.SessionGeneration && session.SessionGeneration-meta.SessionGeneration == 1) ||
+				(meta.SessionGeneration > session.SessionGeneration && meta.SessionGeneration-session.SessionGeneration == 1)) {
+			return fmt.Errorf("Agent fact was fenced during session transition: %w: %w", store.ErrCommandConflict, agentsession.ErrGenerationTransition)
+		}
+		return fmt.Errorf("Agent report has an unapproved session generation: %w", store.ErrCommandConflict)
+	}
 	return nil
 }
 

internal/dispatcher/agent_inbound_session_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	"git.ipao.vip/rogee/go-sip/internal/session"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 	"google.golang.org/protobuf/proto"
 )
@@ -32,20 +33,26 @@ func TestInboundAgentFactRequiresCurrentUnexpiredActivatedSession(t *testing.T)
 		t.Fatalf("active Agent was refused: %v", err)
 	}
 	for _, tc := range []struct {
-		name   string
-		change func(*agentpb.RequestMeta)
+		name       string
+		change     func(*agentpb.RequestMeta)
+		transition bool
 	}{
-		{"old boot", func(m *agentpb.RequestMeta) { m.BootId = "boot-old" }},
-		{"old generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 2 }},
-		{"other cell", func(m *agentpb.RequestMeta) { m.CellId = "cell-other" }},
-		{"other epoch", func(m *agentpb.RequestMeta) { m.DispatcherEpoch = "epoch-other" }},
-		{"unregistered Agent", func(m *agentpb.RequestMeta) { m.AgentId = "agent-other" }},
+		{"old boot", func(m *agentpb.RequestMeta) { m.BootId = "boot-old" }, false},
+		{"old generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 2 }, true},
+		{"next generation before activation response", func(m *agentpb.RequestMeta) { m.SessionGeneration = 4 }, true},
+		{"distant generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 7 }, false},
+		{"missing generation", func(m *agentpb.RequestMeta) { m.SessionGeneration = 0 }, false},
+		{"other cell", func(m *agentpb.RequestMeta) { m.CellId = "cell-other" }, false},
+		{"other epoch", func(m *agentpb.RequestMeta) { m.DispatcherEpoch = "epoch-other" }, false},
+		{"unregistered Agent", func(m *agentpb.RequestMeta) { m.AgentId = "agent-other" }, false},
+		{"old boot and generation", func(m *agentpb.RequestMeta) { m.BootId = "boot-old"; m.SessionGeneration = 2 }, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			changed := proto.Clone(meta).(*agentpb.RequestMeta)
 			tc.change(changed)
-			if err := coordinator.AuthorizeInboundMeta(changed); !errors.Is(err, store.ErrCommandConflict) {
-				t.Fatalf("unbound Agent report passed: %v", err)
+			err := coordinator.AuthorizeInboundMeta(changed)
+			if !errors.Is(err, store.ErrCommandConflict) || errors.Is(err, session.ErrGenerationTransition) != tc.transition {
+				t.Fatalf("Agent report refusal classification: got %v, transition=%t", err, tc.transition)
 			}
 		})
 	}
@@ -53,4 +60,9 @@ func TestInboundAgentFactRequiresCurrentUnexpiredActivatedSession(t *testing.T)
 	if err := coordinator.AuthorizeInboundMeta(meta); !errors.Is(err, store.ErrCommandConflict) {
 		t.Fatalf("expired Agent report passed: %v", err)
 	}
+	stale := proto.Clone(meta).(*agentpb.RequestMeta)
+	stale.SessionGeneration = 2
+	if err := coordinator.AuthorizeInboundMeta(stale); errors.Is(err, session.ErrGenerationTransition) {
+		t.Fatalf("expired Agent session misclassified as a transient cutover: %v", err)
+	}
 }

internal/rpc/recording_server.go

@@ -12,6 +12,7 @@ import (
 
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
 	"git.ipao.vip/rogee/go-sip/internal/oss"
+	"git.ipao.vip/rogee/go-sip/internal/session"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 
 	"google.golang.org/grpc/codes"
@@ -52,6 +53,10 @@ func (s *RecordingServer) authorize(ctx context.Context, meta *agentpb.RequestMe
 		return err
 	}
 	if err := s.AuthorizeSession(meta); err != nil {
+		if errors.Is(err, session.ErrGenerationTransition) {
+			log.Printf("recording RPC session transition dispatcher=%s event=%s agent=%s generation=%d", dispatcherID, sourceEventID, meta.GetAgentId(), meta.GetSessionGeneration())
+			return status.Error(codes.Unauthenticated, session.GenerationTransitionMessage)
+		}
 		log.Printf("recording RPC denied dispatcher=%s event=%s reason=inactive_agent_session", dispatcherID, sourceEventID)
 		return status.Error(codes.PermissionDenied, "Agent session is not active for this Dispatcher")
 	}

internal/rpc/recording_server_flow_test.go

@@ -15,6 +15,7 @@ import (
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
 	"git.ipao.vip/rogee/go-sip/internal/configread"
 	"git.ipao.vip/rogee/go-sip/internal/oss"
+	"git.ipao.vip/rogee/go-sip/internal/session"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 
 	"google.golang.org/grpc/codes"
@@ -130,6 +131,47 @@ func recordingResultPayload(t *testing.T, task configread.Snapshot, grant *agent
 	return raw
 }
 
+func TestRecordingServerClassifiesOnlyVerifiedSessionCutover(t *testing.T) {
+	server, database, verified, request, _ := recordingRPCFixture(t)
+	server.AuthorizeSession = func(*agentpb.RequestMeta) error { return session.ErrGenerationTransition }
+	for _, tc := range []struct {
+		name string
+		call func(context.Context) error
+	}{
+		{"grant", func(ctx context.Context) error { _, err := server.RequestRecordingUpload(ctx, request); return err }},
+		{"call end", func(ctx context.Context) error {
+			_, err := server.ReportCallEnded(ctx, &agentpb.ReportCallEndedRequest{Meta: request.Meta, DispatcherId: request.DispatcherId, TenantId: request.TenantId, SourceEventId: request.SourceEventId})
+			return err
+		}},
+		{"final result", func(ctx context.Context) error {
+			_, err := server.ReportCallResult(ctx, &agentpb.ReportCallResultRequest{Meta: request.Meta, DispatcherId: request.DispatcherId, TenantId: request.TenantId, SourceEventId: request.SourceEventId})
+			return err
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.call(verified)
+			if status.Code(err) != codes.Unauthenticated || status.Convert(err).Message() != "verified Agent session generation changed" {
+				t.Fatalf("verified session transition must be a distinct definite refusal: %v", err)
+			}
+			if err := tc.call(context.Background()); status.Code(err) != codes.Unauthenticated || status.Convert(err).Message() == "verified Agent session generation changed" {
+				t.Fatalf("missing peer misclassified as verified session transition: %v", err)
+			}
+		})
+	}
+	if _, err := database.LoadRecordingUpload(request.DispatcherId, request.SourceEventId); !errors.Is(err, store.ErrUploadNotFound) {
+		t.Fatalf("denied transition created an upload grant: %v", err)
+	}
+	server.TrustedFingerprints = map[string]struct{}{"different": {}}
+	if _, err := server.RequestRecordingUpload(verified, request); status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("wrong mTLS fingerprint was treated as retriable: %v", err)
+	}
+	server.TrustedFingerprints = map[string]struct{}{CertificateFingerprint(&x509.Certificate{Raw: []byte("isolated-test-agent-peer")}): {}}
+	server.AuthorizeSession = func(*agentpb.RequestMeta) error { return errors.New("unapproved Agent identity") }
+	if _, err := server.RequestRecordingUpload(verified, request); status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("unrelated authorization refusal was treated as retriable: %v", err)
+	}
+}
+
 func TestRecordingServerAuthenticatesOriginalGrantAndExplicitReissue(t *testing.T) {
 	server, database, ctx, request, _ := recordingRPCFixture(t)
 	if _, err := server.RequestRecordingUpload(context.Background(), request); status.Code(err) != codes.Unauthenticated {

internal/rpc/recording_transport_test.go

@@ -3,15 +3,19 @@ package rpc
 import (
 	"context"
 	"net"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
 	"git.ipao.vip/rogee/go-sip/internal/agent"
+	"git.ipao.vip/rogee/go-sip/internal/session"
+	"git.ipao.vip/rogee/go-sip/internal/store"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/test/bufconn"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *testing.T) {
@@ -28,6 +32,22 @@ func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *tes
 		t.Fatal(err)
 	}
 	server.TrustedFingerprints = map[string]struct{}{CertificateFingerprint(clientCert): {}}
+	var agentGeneration, activeGeneration atomic.Uint64
+	var fencedReports atomic.Int32
+	agentGeneration.Store(request.Meta.SessionGeneration)
+	activeGeneration.Store(request.Meta.SessionGeneration + 1)
+	server.AuthorizeSession = func(meta *agentpb.RequestMeta) error {
+		active := activeGeneration.Load()
+		if meta.SessionGeneration == active {
+			return nil
+		}
+		if meta.SessionGeneration > 0 && meta.SessionGeneration < active && active-meta.SessionGeneration == 1 {
+			fencedReports.Add(1)
+			agentGeneration.Store(active)
+			return session.ErrGenerationTransition
+		}
+		return store.ErrCommandConflict
+	}
 	listener := bufconn.Listen(1 << 20)
 	grpcServer := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)))
 	agentpb.RegisterAgentControlServiceServer(grpcServer, server)
@@ -45,20 +65,29 @@ func TestRecordingFactsCrossRealMutualTLSUnaryTransportWithoutExternalOSS(t *tes
 	defer cancel()
 	client := agent.RecordingClient{
 		Client: agentpb.NewAgentControlServiceClient(conn), DispatcherID: request.DispatcherId, TenantID: request.TenantId, SourceEventID: request.SourceEventId,
-		Session: func(context.Context) (*agentpb.RequestMeta, error) { return request.Meta, nil },
+		Session: func(context.Context) (*agentpb.RequestMeta, error) {
+			meta := proto.Clone(request.Meta).(*agentpb.RequestMeta)
+			meta.SessionGeneration = agentGeneration.Load()
+			return meta, nil
+		},
 	}
 	grant, err := client.RequestUpload(ctx, request.Asset, request.UploadId)
 	if err != nil || grant.GetBucket() != "mock-bucket" {
 		t.Fatalf("mTLS Agent request did not bind Dispatcher-owned bucket: bucket=%q err=%v", grant.GetBucket(), err)
 	}
+	activeGeneration.Add(1)
 	if err := client.ReportEnded(ctx); err != nil {
 		t.Fatalf("mTLS Agent end did not durably acknowledge original execution: %v", err)
 	}
 	observation := &agentpb.UploadObservation{UploadId: request.UploadId, RecordingId: request.Asset.AssetId, PutStatusCode: 200, SizeBytes: request.Asset.SizeBytes, ChecksumSha256: request.Asset.ChecksumSha256}
+	activeGeneration.Add(1)
 	resultID, err := client.ReportFinal(ctx, recordingResultPayload(t, snapshot, grant), observation)
 	if err != nil || resultID == "" {
 		t.Fatalf("mTLS Agent final result not persisted: fact=%q err=%v", resultID, err)
 	}
+	if fencedReports.Load() != 3 || agentGeneration.Load() != activeGeneration.Load() {
+		t.Fatalf("three mTLS recording facts did not cross the verified Agent session cutovers: fenced=%d current=%d", fencedReports.Load(), agentGeneration.Load())
+	}
 	if request.Meta.OperationId != "request-1" || request.Meta.IdempotencyKey != "request-1" {
 		t.Fatal("concurrent Agent session metadata was mutated by the recording client")
 	}

internal/session/transition.go

@@ -0,0 +1,12 @@
+// Package session shares the verified Agent cutover refusal across the
+// Dispatcher, recording RPC server, and Agent without coupling their lifecycles.
+package session
+
+import "errors"
+
+// GenerationTransitionMessage is the sole retryable recording RPC refusal.
+const GenerationTransitionMessage = "verified Agent session generation changed"
+
+// ErrGenerationTransition marks an adjacent active-session mismatch; it does
+// not authorize the rejected fact or bypass the latest-generation fence.
+var ErrGenerationTransition = errors.New(GenerationTransitionMessage)

proto/ERRORS.md

@@ -10,7 +10,7 @@ contract does not change SaaS/MQ commands, ownership, or application receipts.
 | gRPC status | Required caller behavior |
 | --- | --- |
 | `InvalidArgument` | Correct the request; do not replay it unchanged. |
-| `Unauthenticated` | Re-establish the verified mTLS/active session before more work. |
+| `Unauthenticated` | Re-establish the verified mTLS/active session before more work. Only the exact, verified session-generation cutover refusal described below permits a bounded re-report of the original fact. |
 | `PermissionDenied` | Stop; no self-reported identity or unapproved peer is accepted. |
 | `FailedPrecondition` | Keep admission closed until the missing state, approval, or loaded SIP revision is resolved. |
 | `Aborted` / `AlreadyExists` | Preserve the original execution identity; conflicting content must not originate another call. |
@@ -49,6 +49,17 @@ require separate evidence.
 
 ## Recording and final result
 
+- If a verified mTLS Agent peer reports `ReportCallEnded`, `ReportCallResult`,
+  or `RequestRecordingUpload` across an adjacent active-session generation
+  change for the same Agent, Cell, boot, and Dispatcher epoch, the Dispatcher
+  **rejects** the fact with `Unauthenticated: verified Agent session generation
+  changed`. The Agent may fetch its current session and re-report only that
+  original idempotent fact for at most six seconds (or the earlier caller
+  deadline), retaining the source event, operation/idempotency key, upload ID,
+  asset and result payload. A different identity, expired session, unrelated
+  `Unauthenticated`, `PermissionDenied`, or an unknown/timed-out RPC result does
+  not authorize a retry. This exception never re-originates, resends an OSS
+  PUT, or weakens the latest-generation fence.
 - `RequestRecordingUpload` issues a restricted, short-lived target for the
   original recording. The Agent makes an explicit request for a new grant
   after expiry; there is no automatic token renewal, SaaS upload session, or

proto/manifest.json

@@ -24,8 +24,8 @@
     },
     {
       "path": "proto/ERRORS.md",
-      "bytes": 4160,
-      "sha256": "04bcaa5a821fcdee25f5eeed61b1609541493b072d0a327c5c2b518fde15910f"
+      "bytes": 5108,
+      "sha256": "024f53a02c01151f7223ad49ad2d32c73d7701497dfd032bed0de820d72c7dc4"
     },
     {
       "path": "proto/README.md",