Implement checkpoint blob handling in forwarder service

- Added support for checkpointing phases and blob management in the forwarder.
- Introduced new types and methods for handling checkpoint blobs, including queuing and publishing checkpoints.
- Enhanced the projector to build checkpoint projections with content-addressed blobs.
- Implemented tests to ensure proper checkpoint blob synchronization and handling of cancellation scenarios.

internal/backend/forwarder/actor.go

@@ -23,6 +23,7 @@ const (
 	TurnPhaseWaitingExternal TurnPhase = "waiting_external"
 	TurnPhaseAwaitingUser    TurnPhase = "awaiting_user"
 	TurnPhaseCompacting      TurnPhase = "compacting"
+	TurnPhaseCheckpointing   TurnPhase = "checkpointing"
 	TurnPhaseCompleted       TurnPhase = "completed"
 	TurnPhaseFailed          TurnPhase = "failed"
 	TurnPhaseCanceled        TurnPhase = "canceled"
@@ -66,6 +67,7 @@ const (
 	streamTimerNonStreamingRecovery streamTimerKind = "non_streaming_recovery"
 	streamTimerShellForeground      streamTimerKind = "shell_foreground"
 	streamTimerShellTransportClose  streamTimerKind = "shell_transport_close"
+	streamTimerCheckpointBlobs      streamTimerKind = "checkpoint_blobs"
 	streamTimerOrphanCancel         streamTimerKind = "orphan_cancel"
 )
 
@@ -318,6 +320,9 @@ func (service *Service) handleStreamCommand(stream *ActiveStream, command stream
 	case streamCommandCancel:
 		return service.handleCancelIntent(command.Intent)
 	case streamCommandMetadata:
+		if strings.TrimSpace(command.Intent.Kind) == "kv_result" {
+			return service.handleCheckpointBlobResult(stream, command.Intent.KVClientMessage)
+		}
 		return service.handleMetadataIntent(command.Intent)
 	case streamCommandExecResult:
 		return service.handleExecResult(command.Intent)
@@ -1003,6 +1008,8 @@ func (service *Service) handleTimerEvent(stream *ActiveStream, payload *streamTi
 			return nil
 		}
 		return service.recoverShellWithoutTerminal(stream, current, shellRecoveryReasonTransportClosed)
+	case streamTimerCheckpointBlobs:
+		return service.handleCheckpointBlobTimeout(stream)
 	case streamTimerOrphanCancel:
 		stream.mu.Lock()
 		subscriberCount := len(stream.Subscribers)